"""R04-T02 — unit tests for the typed agent event stream. The events replace the untyped ``{"type": ...}`` dicts the runtime emits today, but ``ui/chat_panel.py::_on_event`` still dispatches on those dicts until R08. So the contract under test is two-sided: each event must be a real typed value AND must serialise back to the exact legacy shape the widget already reads — same wire name, same keys, same optional-key behaviour. """ from __future__ import annotations from dataclasses import FrozenInstanceError import pytest from cowork_local.domain.agents.agent_event import ( AssistantMessageCompletedEvent, ErrorEvent, HistoryReadyEvent, NoticeEvent, OutputsAddedEvent, OutputsRemovedEvent, PlanStep, PlanUpdatedEvent, ReasoningChunkEvent, TextChunkEvent, ToolCallFinishedEvent, ToolCallStartedEvent, ToolOutputChunkEvent, ToolPreview, TurnCompletedEvent, ) from cowork_local.domain.agents.agent_event_codec import from_legacy_dict # -- base contract --------------------------------------------------------- # def test_events_reject_mutation() -> None: event = TextChunkEvent(delta="hello") with pytest.raises(FrozenInstanceError): event.delta = "goodbye" # -- legacy wire compatibility --------------------------------------------- # def test_text_chunk_serialises_as_the_legacy_text_event() -> None: assert TextChunkEvent(delta="hi").to_legacy_dict() == {"type": "text", "delta": "hi"} def test_reasoning_chunk_serialises_as_the_legacy_reasoning_event() -> None: assert ReasoningChunkEvent(delta="hmm").to_legacy_dict() == { "type": "reasoning", "delta": "hmm"} def test_assistant_message_completed_serialises_as_assistant_done() -> None: # Fires once per provider call, so several times in a tool-using turn — it # is NOT the end of the turn (that is TurnCompletedEvent). assert AssistantMessageCompletedEvent(content="done").to_legacy_dict() == { "type": "assistant_done", "content": "done"} def test_tool_call_started_serialises_with_the_legacy_id_and_args_keys() -> None: event = ToolCallStartedEvent( call_id="call_1", name="write_file", arguments={"path": "a.md"}, preview=ToolPreview(kind="diff", title="Create file: a.md", text="+ hi"), ) assert event.to_legacy_dict() == { "type": "tool_proposed", "id": "call_1", "name": "write_file", "args": {"path": "a.md"}, "preview": {"kind": "diff", "title": "Create file: a.md", "text": "+ hi"}, } def test_tool_call_started_omits_the_preview_when_there_is_none() -> None: event = ToolCallStartedEvent(call_id="call_1", name="read_file") assert "preview" not in event.to_legacy_dict() def test_tool_output_chunk_serialises_as_the_legacy_tool_output_event() -> None: event = ToolOutputChunkEvent(call_id="call_1", name="run_command", delta="line\n") assert event.to_legacy_dict() == { "type": "tool_output", "id": "call_1", "name": "run_command", "delta": "line\n"} def test_tool_call_finished_serialises_as_the_legacy_tool_result_event() -> None: event = ToolCallFinishedEvent( call_id="call_1", name="save_file", ok=True, output="saved", path="C:/out/a.md", produced=["C:/out/b.pptx"], ) assert event.to_legacy_dict() == { "type": "tool_result", "id": "call_1", "name": "save_file", "ok": True, "output": "saved", "path": "C:/out/a.md", "produced": ["C:/out/b.pptx"], } def test_tool_call_finished_omits_path_and_produced_when_empty() -> None: # chat_agent only sets these keys when they exist; emitting them as None # would make ``ev.get("path")`` truthy checks read differently downstream. legacy = ToolCallFinishedEvent(call_id="c", name="read_file", ok=True).to_legacy_dict() assert "path" not in legacy assert "produced" not in legacy def test_plan_updated_serialises_steps_back_to_title_status_dicts() -> None: event = PlanUpdatedEvent(steps=(PlanStep(title="Read config", status="done"), PlanStep(title="Patch it", status="running"))) assert event.to_legacy_dict() == { "type": "plan_set", "steps": [{"title": "Read config", "status": "done"}, {"title": "Patch it", "status": "running"}], } def test_notice_serialises_with_its_level() -> None: assert NoticeEvent(text="reading page 2/9", level="progress").to_legacy_dict() == { "type": "notice", "level": "progress", "text": "reading page 2/9"} def test_notice_defaults_to_the_info_level() -> None: assert NoticeEvent(text="compacted").to_legacy_dict()["level"] == "info" def test_outputs_added_and_removed_serialise_their_path_lists() -> None: assert OutputsAddedEvent(paths=("a.md",)).to_legacy_dict() == { "type": "outputs_added", "paths": ["a.md"]} assert OutputsRemovedEvent(paths=("tmp.py",)).to_legacy_dict() == { "type": "outputs_removed", "paths": ["tmp.py"]} def test_history_ready_serialises_its_session_id() -> None: assert HistoryReadyEvent(session_id="s7").to_legacy_dict() == { "type": "history_ready", "session_id": "s7"} # -- events introduced by R04 (no legacy consumer) ------------------------- # def test_turn_completed_carries_the_final_answer_and_step_count() -> None: event = TurnCompletedEvent(final_text="all done", steps_used=3) assert event.to_legacy_dict() == { "type": "turn_completed", "final_text": "all done", "steps_used": 3, "cancelled": False, "budget_exhausted": False} def test_error_event_is_fatal_unless_marked_recoverable() -> None: assert ErrorEvent(message="boom").recoverable is False assert ErrorEvent(message="rate limited", recoverable=True).recoverable is True # -- parsing legacy dicts back into events --------------------------------- # _ROUND_TRIP_CASES = [ TextChunkEvent(delta="hi"), ReasoningChunkEvent(delta="hmm"), AssistantMessageCompletedEvent(content="done"), ToolCallStartedEvent(call_id="c", name="run_command", arguments={"command": "ls"}, preview=ToolPreview(kind="command", title="Run", text="ls")), ToolCallStartedEvent(call_id="c", name="read_file"), ToolOutputChunkEvent(call_id="c", name="run_command", delta="out"), ToolCallFinishedEvent(call_id="c", name="save_file", ok=True, output="ok", path="a.md", produced=["b.md"]), ToolCallFinishedEvent(call_id="c", name="read_file", ok=False, output="missing"), PlanUpdatedEvent(steps=(PlanStep(title="Step", status="pending"),)), NoticeEvent(text="warned", level="warning"), OutputsAddedEvent(paths=("a.md",)), OutputsRemovedEvent(paths=("tmp.py",)), HistoryReadyEvent(session_id="s7"), TurnCompletedEvent(final_text="done", steps_used=2, cancelled=True), ErrorEvent(message="boom", recoverable=True), ] @pytest.mark.parametrize("event", _ROUND_TRIP_CASES, ids=lambda e: type(e).__name__) def test_every_event_survives_a_round_trip_through_the_legacy_dict(event) -> None: assert from_legacy_dict(event.to_legacy_dict()) == event def test_unknown_event_types_parse_to_none_instead_of_raising() -> None: # Co4E emits its own vocabulary (node_status, stage_text, run_done) which R04 # deliberately leaves alone; a bridge must be able to pass those through # untouched rather than crash on them. assert from_legacy_dict({"type": "node_status", "node_id": "n1"}) is None assert from_legacy_dict({"type": ""}) is None assert from_legacy_dict("not a dict") is None def test_missing_payload_keys_parse_to_empty_values() -> None: # Defensive: a truncated event from an older emitter must not kill the turn. assert from_legacy_dict({"type": "text"}) == TextChunkEvent(delta="") assert from_legacy_dict({"type": "tool_result", "id": "c", "name": "x"}) == ( ToolCallFinishedEvent(call_id="c", name="x", ok=False, output="")) def test_plan_steps_from_legacy_drop_entries_without_a_title() -> None: # normalize_plan_steps already clamps upstream; this only guards the parse # path so a hand-written dict cannot produce a titleless step. event = from_legacy_dict({"type": "plan_set", "steps": [{"title": "Real", "status": "done"}, {"status": "done"}]}) assert event == PlanUpdatedEvent(steps=(PlanStep(title="Real", status="done"),))