feat(R04): add the immutable turn snapshot and typed agent event stream

R04-T01 — `domain/agents/conversation_execution_request.py`: a frozen
snapshot of everything one chat turn needs. Turn inputs previously lived in a
closure plus a 15-key ctx dict inside `ui/chat_panel.py::_start_turn`, and the
worker thread kept reading the widget back while it ran, so every later click
was visible to work already in flight. The request also owns the prompt
composition rules (instruction prefix separator, session notes, model-switch
review note) that were inline in that closure.

R04-T02 — `domain/agents/agent_event.py`: 13 frozen event types replacing the
untyped `{"type": ...}` dicts, whose only specification was the 130-line
if/elif chain in `_on_event`. Each event serialises back to the exact legacy
dict, so the presentation layer is untouched; `agent_event_codec.py` parses the
other way and is a temporary shim, isolated so R08 can delete it in one move.
`assistant_done` is deliberately NOT the end of a turn (it fires once per
provider call), so it maps to AssistantMessageCompletedEvent while the new
TurnCompletedEvent reports the turn itself.

R04-T03 (part) — `domain/agents/agent_result.py`: one named outcome for a
finished turn, replacing the message list / 3-tuple / reconstructed-from-side-
effects trio the three callers each read differently.

Verification: 66 tests. Beyond the unit tests,
`tests/integration/test_agent_event_bridge.py` runs the REAL `run_cowork` loop
offline and asserts every dict it emits is recognised and round-trips
byte-for-byte — a guard against an event type nobody modelled or a key whose
meaning silently drifted.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/integration/test_agent_event_bridge.py

@@ -0,0 +1,113 @@
+"""R04-T02 — the typed event vocabulary vs. what the real runtime emits.
+
+The unit tests pin each event against the shape I *read* out of
+``core/chat_agent.py``. This one removes the reading: it runs the actual
+``run_cowork`` loop offline (FakeProvider, real tool execution, real cleanup)
+and asserts every dict it emits is recognised by :func:`from_legacy_dict` and
+survives a round trip byte-for-byte.
+
+That makes it a guard against the two failure modes a hand-written vocabulary
+has: an event type nobody modelled, and a key that silently changes meaning.
+Either one would surface here as a failure instead of as a blank chat bubble
+after R04-T03 starts routing events through the typed layer.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+from cowork_local.core import chat_agent
+from cowork_local.domain.agents.agent_event_codec import from_legacy_dict
+from cowork_local.tests.fakes.fake_provider import FakeProvider
+
+
+def _run_turn_and_collect(tmp_path: Path, provider: FakeProvider) -> List[Dict[str, Any]]:
+    """Run one real ``run_cowork`` turn offline and return every emitted dict."""
+    output_dir = tmp_path / "output"
+    output_dir.mkdir(parents=True, exist_ok=True)
+    emitted: List[Dict[str, Any]] = []
+
+    chat_agent.run_cowork(
+        provider=provider,
+        messages=[{"role": "user", "content": "make me a report"}],
+        output_dir=output_dir,
+        emit=emitted.append,
+        # security_config=None disables the AI guardrail layers, which is the
+        # documented behaviour for headless callers and keeps this test offline.
+        security_config=None,
+        title="Report",
+    )
+    return emitted
+
+
+def _reporting_turn(tmp_path: Path) -> List[Dict[str, Any]]:
+    """A turn that streams text, calls save_file, then answers — the common path."""
+    provider = FakeProvider()
+    provider.queue_response(
+        content="Writing it now.",
+        chunks=["Writing ", "it now."],
+        tool_calls=[{"id": "call_1", "name": "save_file",
+                     "arguments": {"filename": "report.md", "content": "# Report\n"}}],
+    )
+    provider.queue_response(content="Saved to report.md.", chunks=["Saved to report.md."])
+    return _run_turn_and_collect(tmp_path, provider)
+
+
+def test_the_runtime_emits_only_event_types_the_domain_layer_models(tmp_path: Path) -> None:
+    emitted = _reporting_turn(tmp_path)
+
+    unmodelled = sorted({e["type"] for e in emitted if from_legacy_dict(e) is None})
+
+    assert unmodelled == [], f"run_cowork emits event types R04-T02 does not model: {unmodelled}"
+
+
+def test_every_emitted_event_round_trips_without_losing_a_key(tmp_path: Path) -> None:
+    emitted = _reporting_turn(tmp_path)
+    assert emitted, "the turn produced no events at all — the fixture is wrong"
+
+    for raw in emitted:
+        event = from_legacy_dict(raw)
+        assert event is not None, raw
+        assert event.to_legacy_dict() == raw, f"round trip changed the {raw['type']} event"
+
+
+def test_a_tool_using_turn_really_exercises_the_tool_events(tmp_path: Path) -> None:
+    # Guards the test above from passing trivially: if the fixture ever stopped
+    # calling a tool, the round-trip check would only cover text events.
+    types = {e["type"] for e in _reporting_turn(tmp_path)}
+
+    assert {"text", "assistant_done", "tool_proposed", "tool_result"} <= types
+
+
+def test_reasoning_events_from_a_thinking_model_round_trip(tmp_path: Path) -> None:
+    # A separate fixture because only reasoning models emit these, and the
+    # common-path turn above would otherwise never cover the event.
+    provider = FakeProvider()
+    provider.queue_response(content="42", chunks=["42"], reasoning="Let me think...")
+
+    emitted = _run_turn_and_collect(tmp_path, provider)
+
+    reasoning_events = [e for e in emitted if e["type"] == "reasoning"]
+    assert reasoning_events, "a reasoning model produced no reasoning event"
+    for raw in reasoning_events:
+        assert from_legacy_dict(raw).to_legacy_dict() == raw
+
+
+def test_plan_events_from_the_real_update_plan_tool_round_trip(tmp_path: Path) -> None:
+    provider = FakeProvider()
+    provider.queue_response(
+        content="Planning.",
+        tool_calls=[{"id": "call_1", "name": "update_plan",
+                     "arguments": {"steps": [{"title": "Draft", "status": "running"},
+                                             {"title": "Review", "status": "pending"}]}}],
+    )
+    provider.queue_response(content="Done.")
+
+    emitted = _run_turn_and_collect(tmp_path, provider)
+
+    plan_events = [e for e in emitted if e["type"] == "plan_set"]
+    assert plan_events, "update_plan did not produce a plan_set event"
+    for raw in plan_events:
+        assert from_legacy_dict(raw).to_legacy_dict() == raw

tests/unit/test_agent_event.py

@@ -0,0 +1,209 @@
+"""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"),))

tests/unit/test_agent_result.py

@@ -0,0 +1,101 @@
+"""R04-T03 (a) — unit tests for the value a finished turn returns.
+
+Two callers need different things out of one turn today:
+``ui/chat_panel.py::_finalize_turn`` wants the message list, while
+``core/task_executors.py::_run_agent`` returns a
+``(answer_text, timed_out, incomplete_reason)`` tuple assembled by hand. This
+type is what both read instead, so "what happened in that turn?" has one answer
+with names on it.
+"""
+
+from __future__ import annotations
+
+from dataclasses import FrozenInstanceError
+
+import pytest
+from cowork_local.domain.agents.agent_event import PlanStep, TurnCompletedEvent
+from cowork_local.domain.agents.agent_result import AgentResult
+
+
+def test_result_rejects_mutation() -> None:
+    result = AgentResult(steps_used=1)
+
+    with pytest.raises(FrozenInstanceError):
+        result.steps_used = 2
+
+
+def test_messages_are_frozen_into_a_tuple() -> None:
+    live = [{"role": "user", "content": "hi"}]
+
+    result = AgentResult(messages=live)
+    live.append({"role": "assistant", "content": "later"})
+
+    assert result.messages == ({"role": "user", "content": "hi"},)
+
+
+def test_final_text_is_the_last_non_empty_assistant_message() -> None:
+    # A turn ends on a tool message often enough (cancelled mid-loop) that the
+    # answer cannot simply be messages[-1].
+    result = AgentResult(messages=[
+        {"role": "assistant", "content": "first pass"},
+        {"role": "assistant", "content": "the answer"},
+        {"role": "tool", "tool_call_id": "c", "name": "read_file", "content": "..."},
+    ])
+
+    assert result.final_text == "the answer"
+
+
+def test_final_text_skips_a_blank_assistant_message() -> None:
+    result = AgentResult(messages=[
+        {"role": "assistant", "content": "the answer"},
+        {"role": "assistant", "content": "   "},
+    ])
+
+    assert result.final_text == "the answer"
+
+
+def test_final_text_is_empty_when_the_model_never_answered() -> None:
+    assert AgentResult(messages=[{"role": "user", "content": "hi"}]).final_text == ""
+
+
+def test_a_plain_finished_turn_is_ok() -> None:
+    assert AgentResult(messages=[{"role": "assistant", "content": "done"}]).ok is True
+
+
+def test_a_cancelled_turn_is_not_ok() -> None:
+    assert AgentResult(cancelled=True).ok is False
+
+
+def test_a_failed_turn_is_not_ok_and_keeps_its_message() -> None:
+    result = AgentResult(error="SecurityBlocked: nope")
+
+    assert result.ok is False
+    assert result.error == "SecurityBlocked: nope"
+
+
+def test_hitting_the_step_ceiling_is_reported_separately_from_cancelling() -> None:
+    # "Stopped because the safety limit was reached" and "the user pressed Stop"
+    # need different wording in the transcript, so they stay separate flags.
+    result = AgentResult(budget_exhausted=True, steps_used=30)
+
+    assert result.budget_exhausted is True
+    assert result.cancelled is False
+
+
+def test_result_converts_to_the_turn_completed_event() -> None:
+    result = AgentResult(
+        messages=[{"role": "assistant", "content": "done"}],
+        steps_used=3, cancelled=False, budget_exhausted=True,
+    )
+
+    assert result.to_turn_completed_event() == TurnCompletedEvent(
+        final_text="done", steps_used=3, cancelled=False, budget_exhausted=True)
+
+
+def test_plan_steps_are_frozen_into_a_tuple() -> None:
+    steps = [PlanStep(title="Draft", status="done")]
+
+    result = AgentResult(plan_steps=steps)
+    steps.append(PlanStep(title="Review"))
+
+    assert result.plan_steps == (PlanStep(title="Draft", status="done"),)

tests/unit/test_conversation_execution_request.py

@@ -0,0 +1,132 @@
+"""R04-T01 — unit tests for the immutable turn snapshot.
+
+The snapshot exists so a turn already running cannot be altered by the UI the
+user keeps clicking on. These tests pin exactly that: the object refuses
+mutation, it copies the mutable collections handed to it at submit time, and it
+owns the prompt-composition rules that were inline in
+``ui/chat_panel.py::_start_turn``'s worker closure (prefix separator, session
+notes, model-switch review note).
+"""
+
+from __future__ import annotations
+
+from dataclasses import FrozenInstanceError
+from pathlib import Path
+
+import pytest
+from cowork_local.domain.agents.conversation_execution_request import (
+    ConversationExecutionRequest,
+)
+
+
+def _request(**overrides) -> ConversationExecutionRequest:
+    """A minimal valid request; each test overrides only what it exercises."""
+    base = {"turn_id": "t1", "session_id": "s1"}
+    base.update(overrides)
+    return ConversationExecutionRequest(**base)
+
+
+# -- immutability ---------------------------------------------------------- #
+def test_request_rejects_mutation_after_construction() -> None:
+    request = _request(model="gpt-4o-mini")
+
+    with pytest.raises(FrozenInstanceError):
+        request.model = "claude-sonnet-4-6"
+
+
+def test_turn_id_is_required() -> None:
+    with pytest.raises(ValueError):
+        ConversationExecutionRequest(turn_id="", session_id="s1")
+
+
+def test_session_id_is_required() -> None:
+    with pytest.raises(ValueError):
+        ConversationExecutionRequest(turn_id="t1", session_id="")
+
+
+# -- snapshotting mutable UI state ---------------------------------------- #
+def test_attachments_are_snapshotted_away_from_the_caller_list() -> None:
+    picked = ["a.docx"]
+
+    request = _request(attachments=picked)
+    picked.append("b.pdf")   # the composer clears/refills its own list next turn
+
+    assert request.attachments == ("a.docx",)
+
+
+def test_messages_are_snapshotted_away_from_the_live_history_list() -> None:
+    history = [{"role": "user", "content": "earlier"}]
+
+    request = _request(messages=history)
+    history.append({"role": "assistant", "content": "later"})
+
+    assert len(request.messages) == 1
+    assert isinstance(request.messages, tuple)
+
+
+def test_allowed_tools_none_means_every_tool_stays_available() -> None:
+    # None and () must stay distinguishable: None = no restriction, () = deny
+    # every built-in tool. Coercing None to () would silently disarm the agent.
+    assert _request().allowed_tools is None
+    assert _request(allowed_tools=[]).allowed_tools == ()
+
+
+def test_output_paths_accept_strings_and_normalise_to_path() -> None:
+    request = _request(output_dir="out/t1", home_output_root="out")
+
+    assert request.output_dir == Path("out/t1")
+    assert request.home_output_root == Path("out")
+
+
+# -- derived turn policy --------------------------------------------------- #
+def test_effective_max_steps_uses_the_interactive_cap_by_default() -> None:
+    assert _request(max_steps=30, completion_max_steps=200).effective_max_steps == 30
+
+
+def test_effective_max_steps_lifts_the_cap_when_running_to_completion() -> None:
+    request = _request(max_steps=30, completion_max_steps=200, run_to_completion=True)
+
+    assert request.effective_max_steps == 200
+
+
+def test_permission_gate_is_required_only_in_confirm_mode() -> None:
+    assert _request(gate_mode="confirm").requires_permission_gate is True
+    assert _request(gate_mode="auto").requires_permission_gate is False
+
+
+def test_has_prompt_ignores_whitespace_only_input() -> None:
+    assert _request(prompt="   \n ").has_prompt is False
+    assert _request(prompt="do it").has_prompt is True
+
+
+# -- prompt composition (moved out of the widget's worker closure) --------- #
+def test_user_content_returns_the_body_unchanged_without_prefix_or_notes() -> None:
+    assert _request().user_content("the body") == "the body"
+
+
+def test_user_content_separates_the_instruction_prefix_from_the_body() -> None:
+    request = _request(instruction_prefix="SKILL RULES")
+
+    assert request.user_content("the body") == "SKILL RULES\n\n---\n\nthe body"
+
+
+def test_user_content_appends_session_notes_after_the_body() -> None:
+    request = _request(session_notes="Files produced earlier: a.md")
+
+    assert request.user_content("the body") == "the body\n\nFiles produced earlier: a.md"
+
+
+def test_user_content_falls_back_to_session_notes_when_the_body_is_empty() -> None:
+    # An attachment-only turn has no typed text, so the notes must not be
+    # prefixed with a stray blank line.
+    request = _request(session_notes="Files produced earlier: a.md")
+
+    assert request.user_content("") == "Files produced earlier: a.md"
+
+
+def test_user_content_puts_the_review_note_ahead_of_everything_else() -> None:
+    request = _request(instruction_prefix="SKILL RULES", review_note="[Note: switched]")
+
+    content = request.user_content("the body")
+
+    assert content == "[Note: switched]\n\nSKILL RULES\n\n---\n\nthe body"