"""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"]