tests/test_transcript_to_tasks.py

"""Tests for `app.transcript_to_tasks.translate`.

Coverage targets (one test per row, see
`specs/realtime-progress-via-gateway-ws.md` §6 + §8.2):

- state alphabet dispatch (delta / final / error / aborted / unknown)
- frame-shape rejection (non-event, wrong event name, missing payload)
- tool_use block → AddTask (per tool name in the mapping table)
- tool_result block → CompleteTask correlated by id
- message-level tool-result form (role=tool, toolCallId at root)
- _INTERNAL_TOOLS_SKIP suppression
- icon allowlist enforcement (only bolt / agent emitted)
- tool_use idempotency on replay
- malformed envelopes never raise
- sessions_spawn special-case (icon=agent, agentId+task)

Wire shape modelled after `specs/m1-t13-ws-tap-evidence/raw-envelopes.jsonl`
(observed delta state) + the gateway sanitizer source that preserves
tool_use / tool_result blocks at state="final".
"""
from __future__ import annotations

from app.transcript_to_tasks import (
    AddTask,
    CompleteTask,
    MarkRunTerminal,
    PerTurnState,
    translate,
)


# --------------------------------------------------------------------------
# Fixture helpers — model the canonical post-sanitizer `chat` envelope.
# --------------------------------------------------------------------------


def _envelope(
    *,
    state: str = "final",
    role: str = "assistant",
    content: object = None,
    session_key: str = "agent:fullstack-developer:subagent:abc",
    run_id: str = "run-1",
    seq: int = 1,
    msg_extra: dict | None = None,
    payload_extra: dict | None = None,
) -> dict:
    msg: dict = {"role": role}
    if content is not None:
        msg["content"] = content
    if msg_extra:
        msg.update(msg_extra)
    payload = {
        "runId": run_id,
        "sessionKey": session_key,
        "seq": seq,
        "state": state,
        "message": msg,
    }
    if payload_extra:
        payload.update(payload_extra)
    return {"type": "event", "event": "chat", "payload": payload, "seq": seq}


# --------------------------------------------------------------------------
# Frame-shape guards
# --------------------------------------------------------------------------


def test_non_event_frame_ignored() -> None:
    assert translate({"type": "res", "ok": True}, PerTurnState()) == []


def test_wrong_event_name_ignored() -> None:
    env = _envelope()
    env["event"] = "agent"
    assert translate(env, PerTurnState()) == []


def test_missing_payload_ignored() -> None:
    assert translate({"type": "event", "event": "chat"}, PerTurnState()) == []


def test_non_dict_input_ignored() -> None:
    assert translate("not a dict", PerTurnState()) == []  # type: ignore[arg-type]
    assert translate(None, PerTurnState()) == []  # type: ignore[arg-type]


def test_translator_never_raises_on_garbage() -> None:
    # Random shapes shouldn't crash.
    cases: list = [
        {},
        {"type": "event", "event": "chat", "payload": "not a dict"},
        {"type": "event", "event": "chat", "payload": {"state": "final", "message": "str"}},
        {
            "type": "event",
            "event": "chat",
            "payload": {
                "state": "final",
                "message": {"role": "assistant", "content": [1, 2, 3]},
            },
        },
        {
            "type": "event",
            "event": "chat",
            "payload": {
                "state": "final",
                "message": {"role": "assistant", "content": [{"type": None}]},
            },
        },
    ]
    for c in cases:
        translate(c, PerTurnState())  # no exception => pass


# --------------------------------------------------------------------------
# State machine dispatch
# --------------------------------------------------------------------------


def test_delta_state_ignored() -> None:
    env = _envelope(state="delta", content=[{"type": "text", "text": "hi"}])
    assert translate(env, PerTurnState()) == []


def test_unknown_state_ignored() -> None:
    env = _envelope(state="weird", content=[{"type": "text", "text": "hi"}])
    assert translate(env, PerTurnState()) == []


def test_error_state_yields_terminal_with_message() -> None:
    env = _envelope(state="error", payload_extra={"errorMessage": "model timed out"})
    out = translate(env, PerTurnState())
    assert len(out) == 1
    m = out[0]
    assert isinstance(m, MarkRunTerminal)
    assert m.suffix is not None
    assert m.suffix.title == "Run errored: model timed out"
    assert m.suffix.icon == "bolt"


def test_error_state_truncates_long_message() -> None:
    long = "x" * 500
    env = _envelope(state="error", payload_extra={"errorMessage": long})
    m = translate(env, PerTurnState())[0]
    assert isinstance(m, MarkRunTerminal)
    assert m.suffix is not None
    assert len(m.suffix.title) <= len("Run errored: ") + 80


def test_aborted_state_yields_terminal() -> None:
    env = _envelope(state="aborted")
    out = translate(env, PerTurnState())
    assert out == [MarkRunTerminal(suffix=AddTask(title="Aborted", icon="bolt"))]


# --------------------------------------------------------------------------
# tool_use mapping table (one test per row in §6.2)
# --------------------------------------------------------------------------


def _final_with_tool_use(name: str, inp: dict, tid: str = "tu-1") -> dict:
    return _envelope(
        content=[{"type": "tool_use", "id": tid, "name": name, "input": inp}]
    )


def test_read_tool_use_maps_to_reading_basename() -> None:
    out = translate(
        _final_with_tool_use("Read", {"file_path": "/home/raza/foo/bar.py"}),
        PerTurnState(),
    )
    assert out == [AddTask(title="Reading bar.py", icon="bolt", tool_call_id="tu-1")]


def test_write_tool_use_maps_to_writing_basename() -> None:
    out = translate(
        _final_with_tool_use("Write", {"file_path": "/tmp/spec.md"}),
        PerTurnState(),
    )
    assert out == [AddTask(title="Writing spec.md", icon="bolt", tool_call_id="tu-1")]


def test_edit_tool_use_maps_to_editing_basename() -> None:
    out = translate(
        _final_with_tool_use("Edit", {"file_path": "app/main.py"}),
        PerTurnState(),
    )
    assert out == [AddTask(title="Editing main.py", icon="bolt", tool_call_id="tu-1")]


def test_bash_tool_use_strips_leading_cd_and_truncates_words() -> None:
    out = translate(
        _final_with_tool_use(
            "Bash",
            {"command": "cd /repo && pytest -q tests/test_progress_workflow.py -v --tb=short"},
        ),
        PerTurnState(),
    )
    assert isinstance(out[0], AddTask)
    title = out[0].title
    assert title.startswith("Running: pytest -q tests/test_progress_workflow.py")
    assert out[0].icon == "bolt"


def test_grep_tool_use_truncates_long_pattern() -> None:
    pat = "a" * 100
    out = translate(_final_with_tool_use("Grep", {"pattern": pat}), PerTurnState())
    assert isinstance(out[0], AddTask)
    assert out[0].title == f'Searching: "{"a" * 40}"'


def test_glob_tool_use() -> None:
    out = translate(
        _final_with_tool_use("Glob", {"pattern": "**/*.py"}), PerTurnState()
    )
    assert out[0] == AddTask(
        title='Finding files: "**/*.py"', icon="bolt", tool_call_id="tu-1"
    )


def test_websearch_tool_use_uses_query_field() -> None:
    out = translate(
        _final_with_tool_use("WebSearch", {"query": "openclaw progress events"}),
        PerTurnState(),
    )
    assert out[0].title.startswith('Searching the web: "openclaw progress events')
    assert out[0].icon == "bolt"


def test_webfetch_tool_use_extracts_host() -> None:
    out = translate(
        _final_with_tool_use("WebFetch", {"url": "https://docs.openai.com/guides/x"}),
        PerTurnState(),
    )
    assert out[0] == AddTask(
        title="Fetching docs.openai.com", icon="bolt", tool_call_id="tu-1"
    )


def test_sessions_spawn_tool_use_uses_agent_icon_and_agentid() -> None:
    out = translate(
        _final_with_tool_use(
            "sessions_spawn",
            {
                "agentId": "andy",
                "task": "Subscribe to gateway WS and translate envelopes into tasks",
            },
        ),
        PerTurnState(),
    )
    assert isinstance(out[0], AddTask)
    assert out[0].icon == "agent"
    assert out[0].title.startswith("Delegating to andy:")
    assert "Subscribe to gateway WS" in out[0].title
    assert len(out[0].title) <= len("Delegating to andy: ") + 60


def test_unknown_tool_use_defaults_to_calling_name() -> None:
    out = translate(
        _final_with_tool_use("MagicNewTool", {"foo": "bar"}), PerTurnState()
    )
    assert out[0] == AddTask(
        title="Calling MagicNewTool", icon="bolt", tool_call_id="tu-1"
    )


# --------------------------------------------------------------------------
# Internal tools skip
# --------------------------------------------------------------------------


def test_internal_tools_are_skipped() -> None:
    for name in ("sessions_send", "sessions_history", "sessions_yield"):
        out = translate(
            _final_with_tool_use(name, {"sessionKey": "agent:x"}), PerTurnState()
        )
        assert out == [], f"expected skip for {name}, got {out}"


# --------------------------------------------------------------------------
# tool_result correlation
# --------------------------------------------------------------------------


def test_tool_use_then_tool_result_completes_task() -> None:
    state = PerTurnState()
    env_use = _final_with_tool_use("Read", {"file_path": "/a/b.py"})
    out_use = translate(env_use, state)
    assert out_use == [AddTask(title="Reading b.py", icon="bolt", tool_call_id="tu-1")]
    assert "tu-1" in state.open_tool_uses

    env_res = _envelope(
        role="tool",
        content=[{"type": "tool_result", "tool_use_id": "tu-1", "content": "ok"}],
    )
    out_res = translate(env_res, state)
    assert out_res == [CompleteTask(tool_call_id="tu-1")]
    assert "tu-1" not in state.open_tool_uses


def test_tool_result_without_matching_open_is_dropped() -> None:
    state = PerTurnState()
    env_res = _envelope(
        role="tool",
        content=[{"type": "tool_result", "tool_use_id": "missing"}],
    )
    assert translate(env_res, state) == []


def test_message_level_tool_result_form() -> None:
    state = PerTurnState()
    translate(_final_with_tool_use("Read", {"file_path": "x"}, tid="z9"), state)
    env_res = _envelope(
        role="tool",
        msg_extra={"toolCallId": "z9", "content": "..."},
        content=[],  # no content blocks; toolCallId at root
    )
    out = translate(env_res, state)
    assert out == [CompleteTask(tool_call_id="z9")]


def test_tool_use_idempotent_on_replay() -> None:
    state = PerTurnState()
    env = _final_with_tool_use("Read", {"file_path": "x"}, tid="dup")
    first = translate(env, state)
    second = translate(env, state)  # same envelope re-delivered post-reconnect
    assert len(first) == 1
    assert second == []
    assert state.open_tool_uses["dup"] == "Read"


# --------------------------------------------------------------------------
# Icon allowlist guard (only bolt / agent emitted)
# --------------------------------------------------------------------------


def test_only_bolt_and_agent_icons_emitted() -> None:
    cases = [
        ("Read", {"file_path": "x"}),
        ("Write", {"file_path": "x"}),
        ("Edit", {"file_path": "x"}),
        ("Bash", {"command": "ls"}),
        ("Grep", {"pattern": "x"}),
        ("Glob", {"pattern": "x"}),
        ("WebSearch", {"query": "x"}),
        ("WebFetch", {"url": "https://x.com"}),
        ("sessions_spawn", {"agentId": "andy", "task": "x"}),
        ("Some-Random-Tool", {}),
    ]
    allowed = {"bolt", "agent"}
    for name, inp in cases:
        out = translate(_final_with_tool_use(name, inp), PerTurnState())
        assert out, name
        assert isinstance(out[0], AddTask)
        assert out[0].icon in allowed, (name, out[0].icon)


# --------------------------------------------------------------------------
# Per-turn state tracking — seen_session_keys
# --------------------------------------------------------------------------


def test_seen_session_keys_records_inbound_session() -> None:
    state = PerTurnState()
    translate(
        _envelope(state="delta", session_key="agent:foo:subagent:1"), state
    )
    translate(
        _envelope(state="final", session_key="agent:bar:subagent:2"), state
    )
    assert state.seen_session_keys == {"agent:foo:subagent:1", "agent:bar:subagent:2"}


# --------------------------------------------------------------------------
# Multi-block final (one envelope carrying multiple tool_use blocks)
# --------------------------------------------------------------------------


def test_toolcall_transcript_shape_yields_add_and_complete() -> None:
    """Option 1 (spec §6.1b): sessions_history transcript shape uses
    `type:"toolcall"` (no underscore) + `block.arguments` (not `input`),
    and co-locates tool_result on the SAME assistant message in one
    content array. The translator must recognise this shape end-to-end."""
    state = PerTurnState()
    env = _envelope(
        role="assistant",
        content=[
            {
                "type": "toolcall",
                "id": "toolu_01",
                "name": "Bash",
                "arguments": {"command": "ls /tmp", "description": "list tmp"},
            },
            {
                "type": "tool_result",
                "tool_use_id": "toolu_01",
                "content": "a\nb\nc",
                "name": "Bash",
                "is_error": False,
            },
        ],
    )
    out = translate(env, state)
    # Expect exactly: AddTask("Running: ls /tmp", bolt, toolu_01) +
    # CompleteTask("toolu_01"). Single envelope, both mutations.
    assert len(out) == 2, out
    assert isinstance(out[0], AddTask)
    assert out[0].title.startswith("Running: ls /tmp")
    assert out[0].icon == "bolt"
    assert out[0].tool_call_id == "toolu_01"
    assert isinstance(out[1], CompleteTask)
    assert out[1].tool_call_id == "toolu_01"


def test_toolcall_with_no_arguments_falls_back_safely() -> None:
    """toolcall block with neither `input` nor `arguments` must not crash —
    title defaults gracefully."""
    state = PerTurnState()
    env = _envelope(
        role="assistant",
        content=[{"type": "toolcall", "id": "x", "name": "Read"}],
    )
    out = translate(env, state)
    assert len(out) == 1
    assert isinstance(out[0], AddTask)
    # No file_path → title is "Reading " (empty basename, stripped).
    assert out[0].icon == "bolt"
    assert out[0].tool_call_id == "x"


def test_toolcall_uses_input_when_both_input_and_arguments_present() -> None:
    """Defensive: if a block carries BOTH `input` and `arguments` (mixed
    shape), the translator prefers `input` (the WS-canonical name)."""
    state = PerTurnState()
    env = _envelope(
        role="assistant",
        content=[
            {
                "type": "toolcall",
                "id": "x",
                "name": "Read",
                "input": {"file_path": "/from/input/a.py"},
                "arguments": {"file_path": "/from/args/b.py"},
            }
        ],
    )
    out = translate(env, state)
    assert isinstance(out[0], AddTask)
    assert out[0].title == "Reading a.py"


def test_assistant_role_tool_result_completes_open_task() -> None:
    """Spec §6.1b: transcript co-locates tool_result with the toolcall on
    the SAME assistant message. The translator's per-block dispatch must
    handle tool_result regardless of message role (assistant included)."""
    state = PerTurnState()
    # First: add a tool task via separate envelope.
    add_env = _envelope(
        role="assistant",
        content=[
            {
                "type": "toolcall",
                "id": "z",
                "name": "Grep",
                "arguments": {"pattern": "needle"},
            }
        ],
    )
    translate(add_env, state)
    assert "z" in state.open_tool_uses
    # Then: tool_result block on a SEPARATE assistant message.
    res_env = _envelope(
        role="assistant",
        content=[
            {"type": "tool_result", "tool_use_id": "z", "content": "match found"}
        ],
    )
    out = translate(res_env, state)
    assert out == [CompleteTask(tool_call_id="z")]
    assert "z" not in state.open_tool_uses


def test_multi_tool_use_in_one_final_envelope_yields_all() -> None:
    env = _envelope(
        content=[
            {"type": "text", "text": "I'll do these in parallel:"},
            {"type": "tool_use", "id": "a", "name": "Read", "input": {"file_path": "a.py"}},
            {"type": "tool_use", "id": "b", "name": "Grep", "input": {"pattern": "foo"}},
        ]
    )
    out = translate(env, PerTurnState())
    assert len(out) == 2
    assert all(isinstance(m, AddTask) for m in out)
    assert [m.tool_call_id for m in out if isinstance(m, AddTask)] == ["a", "b"]