diff --git a/domain/agents/agent_event.py b/domain/agents/agent_event.py new file mode 100644 index 0000000..824ab88 --- /dev/null +++ b/domain/agents/agent_event.py @@ -0,0 +1,358 @@ +"""Typed events a turn emits while it runs (R04-T02). + +The runtime currently speaks in bare dicts: ``emit({"type": "tool_result", "id": +..., "ok": ...})``. Nothing declares which keys a given type carries, so the +only specification is the 130-line ``if/elif`` chain in +``ui/chat_panel.py::_on_event`` — and a typo in an emitter surfaces as a widget +that silently renders nothing. + +This module makes the vocabulary explicit. Each event is a frozen dataclass with +real fields, and each one knows how to serialise itself back to the exact legacy +dict the widget already reads (:meth:`AgentEvent.to_legacy_dict`), with +:func:`from_legacy_dict` parsing the other way. That two-way bridge is what lets +R04 introduce typed events WITHOUT touching the presentation layer — decomposing +``_on_event`` into a renderer is R08-T01's job, and forcing both changes into one +PR is exactly the "rewrite everything at once" the refactor plan forbids. + +Scope note: this covers the interactive/scheduled **Cowork turn** vocabulary +(the ``run_cowork`` path R04 unifies). Co4E's own node events (``node_status``, +``stage_text``, ``run_done``) belong to ``Co4EWorkflowService`` in R07-T06 and +are deliberately left as dicts here — :func:`from_legacy_dict` returns ``None`` +for them so a bridge can pass them straight through. + +Layer rules (``docs/architecture/ADR-001-layered-architecture.md``): domain +layer, standard library only. No PySide6, no ``core/*`` imports. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, ClassVar, Dict, Iterable, Optional, Tuple + +# Notice levels. "progress" is special-cased by the UI (it retargets the live +# thinking indicator instead of adding a bubble), so the vocabulary is pinned +# here rather than left to each emitter's string literal. +NOTICE_INFO = "info" +NOTICE_WARNING = "warning" +NOTICE_PROGRESS = "progress" + + +# --------------------------------------------------------------------------- # +# Value objects shared by several events. +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class ToolPreview: + """The human-readable preview of a proposed tool call. + + Mirrors ``core/tools.py::describe_action``'s return shape exactly (three + string keys, nothing else), so wrapping it in a type is lossless. ``kind`` + drives which bubble the UI renders: "diff" -> coloured before/after, + "command" -> terminal block, "info" -> plain text. + """ + + kind: str = "info" + title: str = "" + text: str = "" + + def to_dict(self) -> Dict[str, str]: + return {"kind": self.kind, "title": self.title, "text": self.text} + + @classmethod + def from_dict(cls, raw: Any) -> Optional["ToolPreview"]: + """Parse a legacy preview dict; ``None`` when there was none. + + A non-dict value degrades to ``None`` rather than raising: a malformed + preview must cost the user a nicer bubble, never the whole turn. + """ + if not isinstance(raw, dict) or not raw: + return None + return cls(kind=str(raw.get("kind", "info")), title=str(raw.get("title", "")), + text=str(raw.get("text", ""))) + + +@dataclass(frozen=True) +class PlanStep: + """One entry of the agent's ``update_plan`` checklist. + + ``status`` is kept a plain string on purpose: ``core/plan.py`` already owns + validation (clamping anything unknown to "pending" against + pending/running/done/error), and duplicating that vocabulary here would give + the app two sources of truth to drift apart. + """ + + title: str + status: str = "pending" + + def to_dict(self) -> Dict[str, str]: + return {"title": self.title, "status": self.status} + + +def _as_str_tuple(values: Iterable[Any]) -> Tuple[str, ...]: + """Freeze an iterable of paths into a tuple of strings. + + Emitters hand us live lists (``record["outputs"]``, ``_cleanup``'s result); + copying decouples the event from later mutation of that list. + """ + return tuple(str(v) for v in (values or ())) + + +# --------------------------------------------------------------------------- # +# Base class. +# --------------------------------------------------------------------------- # +class AgentEvent: + """Base for every turn event. + + Not a dataclass itself (it holds no data) — subclasses are the frozen + dataclasses. ``EVENT_TYPE`` is the legacy wire name, which stays the single + identifier shared between the typed world and the dict world. + """ + + EVENT_TYPE: ClassVar[str] = "" + + def _payload(self) -> Dict[str, Any]: + """Type-specific keys of the legacy dict (without ``type``).""" + return {} + + def to_legacy_dict(self) -> Dict[str, Any]: + """The exact dict shape ``ui/chat_panel.py::_on_event`` dispatches on.""" + return {"type": self.EVENT_TYPE, **self._payload()} + + +# --------------------------------------------------------------------------- # +# Streaming events. +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class TextChunkEvent(AgentEvent): + """A fragment of the assistant's visible answer.""" + + EVENT_TYPE: ClassVar[str] = "text" + delta: str = "" + + def _payload(self) -> Dict[str, Any]: + return {"delta": self.delta} + + +@dataclass(frozen=True) +class ReasoningChunkEvent(AgentEvent): + """A fragment of a reasoning model's thinking, shown in a collapsed box.""" + + EVENT_TYPE: ClassVar[str] = "reasoning" + delta: str = "" + + def _payload(self) -> Dict[str, Any]: + return {"delta": self.delta} + + +@dataclass(frozen=True) +class AssistantMessageCompletedEvent(AgentEvent): + """One assistant message finished streaming. + + Emitted once per provider call, so a tool-using turn produces SEVERAL of + these — it marks an autosave point, not the end of the turn. The end of the + turn is :class:`TurnCompletedEvent`. + """ + + EVENT_TYPE: ClassVar[str] = "assistant_done" + content: str = "" + + def _payload(self) -> Dict[str, Any]: + return {"content": self.content} + + +# --------------------------------------------------------------------------- # +# Tool-call lifecycle. +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class ToolCallStartedEvent(AgentEvent): + """A tool call is about to run (after any security/permission gate). + + Field names are the typed ones (``call_id``, ``arguments``); the legacy keys + ``id``/``args`` are produced only at the serialisation boundary, so new code + never has to shadow the ``id`` builtin. + """ + + EVENT_TYPE: ClassVar[str] = "tool_proposed" + call_id: str = "" + name: str = "" + arguments: Dict[str, Any] = field(default_factory=dict) + preview: Optional[ToolPreview] = None + + def _payload(self) -> Dict[str, Any]: + payload: Dict[str, Any] = {"id": self.call_id, "name": self.name, + "args": dict(self.arguments)} + # Omitted rather than sent as None: the widget does + # ``preview = ev.get("preview") or {}`` and an absent key is the shape it + # already handles for tools without a preview. + if self.preview is not None: + payload["preview"] = self.preview.to_dict() + return payload + + +@dataclass(frozen=True) +class ToolOutputChunkEvent(AgentEvent): + """Live stdout/stderr from a running command, appended to its step bubble.""" + + EVENT_TYPE: ClassVar[str] = "tool_output" + call_id: str = "" + name: str = "" + delta: str = "" + + def _payload(self) -> Dict[str, Any]: + return {"id": self.call_id, "name": self.name, "delta": self.delta} + + +@dataclass(frozen=True) +class ToolCallFinishedEvent(AgentEvent): + """A tool call returned. ``path``/``produced`` name files it created.""" + + EVENT_TYPE: ClassVar[str] = "tool_result" + call_id: str = "" + name: str = "" + ok: bool = False + output: str = "" + path: str = "" # the single file this call wrote, if any + produced: Tuple[str, ...] = () # extra deliverables a command produced + + def __post_init__(self) -> None: + # Callers pass a live list; freeze it so the event cannot change later. + object.__setattr__(self, "produced", _as_str_tuple(self.produced)) + + def _payload(self) -> Dict[str, Any]: + payload: Dict[str, Any] = {"id": self.call_id, "name": self.name, + "ok": self.ok, "output": self.output} + # Both keys stay ABSENT when empty, matching what chat_agent emits today: + # downstream code tests them with ``ev.get(...)`` truthiness and iterates + # ``ev.get("produced", [])``, so adding empty values would be a change. + if self.path: + payload["path"] = self.path + if self.produced: + payload["produced"] = list(self.produced) + return payload + + +# --------------------------------------------------------------------------- # +# Side-channel events (plan, notices, output folder). +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class PlanUpdatedEvent(AgentEvent): + """The agent published a new version of its step checklist (full list).""" + + EVENT_TYPE: ClassVar[str] = "plan_set" + steps: Tuple[PlanStep, ...] = () + + def __post_init__(self) -> None: + object.__setattr__(self, "steps", tuple(self.steps or ())) + + def _payload(self) -> Dict[str, Any]: + return {"steps": [s.to_dict() for s in self.steps]} + + +@dataclass(frozen=True) +class NoticeEvent(AgentEvent): + """An aside outside the model's own answer. + + Three sources today: context auto-compaction (info), a blocked + security check (warning), and attachment reading progress (progress). + """ + + EVENT_TYPE: ClassVar[str] = "notice" + text: str = "" + level: str = NOTICE_INFO + + def _payload(self) -> Dict[str, Any]: + return {"level": self.level, "text": self.text} + + +@dataclass(frozen=True) +class OutputsAddedEvent(AgentEvent): + """Deliverables appeared in the turn's output folder.""" + + EVENT_TYPE: ClassVar[str] = "outputs_added" + paths: Tuple[str, ...] = () + + def __post_init__(self) -> None: + object.__setattr__(self, "paths", _as_str_tuple(self.paths)) + + def _payload(self) -> Dict[str, Any]: + return {"paths": list(self.paths)} + + +@dataclass(frozen=True) +class OutputsRemovedEvent(AgentEvent): + """Intermediate/generator files were cleaned up — drop them from Output.""" + + EVENT_TYPE: ClassVar[str] = "outputs_removed" + paths: Tuple[str, ...] = () + + def __post_init__(self) -> None: + object.__setattr__(self, "paths", _as_str_tuple(self.paths)) + + def _payload(self) -> Dict[str, Any]: + return {"paths": list(self.paths)} + + +@dataclass(frozen=True) +class HistoryReadyEvent(AgentEvent): + """The turn's conversation now exists on disk and can be opened. + + Emitted by the unattended (Schedule Task) path so the scheduler refreshes + History only once the session is really there. + """ + + EVENT_TYPE: ClassVar[str] = "history_ready" + session_id: str = "" + + def _payload(self) -> Dict[str, Any]: + return {"session_id": self.session_id} + + +# --------------------------------------------------------------------------- # +# Turn-level events introduced by R04 (no legacy consumer yet). +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class TurnCompletedEvent(AgentEvent): + """The whole turn ended — exactly once per turn. + + Nothing consumes ``"turn_completed"`` yet: the widget's ``if/elif`` chain + simply has no branch for it, so emitting it is inert until R08 wires a + renderer. It exists now because the state it carries (was the turn + cancelled? did it hit the step ceiling?) is currently reconstructed by the + UI from side effects rather than being told to it. + """ + + EVENT_TYPE: ClassVar[str] = "turn_completed" + final_text: str = "" + steps_used: int = 0 + cancelled: bool = False + budget_exhausted: bool = False # stopped at effective_max_steps + + def _payload(self) -> Dict[str, Any]: + return {"final_text": self.final_text, "steps_used": self.steps_used, + "cancelled": self.cancelled, "budget_exhausted": self.budget_exhausted} + + +@dataclass(frozen=True) +class ErrorEvent(AgentEvent): + """The turn hit an error. + + ``recoverable`` separates "this turn is over" from "something failed but the + loop carried on" — a distinction the current code loses, because both end up + as a bare ``except Exception`` plus a text bubble. + """ + + EVENT_TYPE: ClassVar[str] = "error" + message: str = "" + recoverable: bool = False + + def _payload(self) -> Dict[str, Any]: + return {"message": self.message, "recoverable": self.recoverable} + + +__all__ = [ + "NOTICE_INFO", "NOTICE_WARNING", "NOTICE_PROGRESS", + "AgentEvent", "ToolPreview", "PlanStep", + "TextChunkEvent", "ReasoningChunkEvent", "AssistantMessageCompletedEvent", + "ToolCallStartedEvent", "ToolOutputChunkEvent", "ToolCallFinishedEvent", + "PlanUpdatedEvent", "NoticeEvent", "OutputsAddedEvent", "OutputsRemovedEvent", + "HistoryReadyEvent", "TurnCompletedEvent", "ErrorEvent", +] diff --git a/domain/agents/agent_event_codec.py b/domain/agents/agent_event_codec.py new file mode 100644 index 0000000..4caf845 --- /dev/null +++ b/domain/agents/agent_event_codec.py @@ -0,0 +1,123 @@ +"""Legacy dict -> typed :mod:`agent_event` translation (R04-T02). + +Kept in its own module for two reasons. It is a **temporary compatibility +shim**: once R08-T01 turns ``ui/chat_panel.py::_on_event`` into an event +renderer that consumes typed events directly, nothing needs to parse dicts any +more and this whole file gets deleted — a deletion that stays trivial only while +it is isolated. And it keeps ``agent_event.py`` inside the 400-LOC limit the +architecture rules impose, without diluting either file's single job: one +declares the vocabulary, the other bridges it to the old wire format. + +Serialisation the other way lives on the events themselves +(``AgentEvent.to_legacy_dict``), because an event has to be emittable without +anyone importing a codec. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Tuple + +from .agent_event import ( + AgentEvent, + AssistantMessageCompletedEvent, + ErrorEvent, + HistoryReadyEvent, + NOTICE_INFO, + NoticeEvent, + OutputsAddedEvent, + OutputsRemovedEvent, + PlanStep, + PlanUpdatedEvent, + ReasoningChunkEvent, + TextChunkEvent, + ToolCallFinishedEvent, + ToolCallStartedEvent, + ToolOutputChunkEvent, + ToolPreview, + TurnCompletedEvent, +) + + +def _plan_steps_from_legacy(raw: Any) -> Tuple[PlanStep, ...]: + """Parse the legacy ``steps`` list, dropping anything unusable. + + A step with no title cannot be rendered or ticked off, so it is discarded + instead of becoming a blank row in the Plan panel. + """ + if not isinstance(raw, list): + return () + steps: List[PlanStep] = [] + for item in raw: + if not isinstance(item, dict): + continue + title = str(item.get("title", "")).strip() + if not title: + continue + steps.append(PlanStep(title=title, status=str(item.get("status", "pending")))) + return tuple(steps) + + +def _parse_tool_started(raw: Dict[str, Any]) -> ToolCallStartedEvent: + """Rebuild a ``tool_proposed`` event, mapping ``id``/``args`` to typed names.""" + args = raw.get("args") + return ToolCallStartedEvent( + call_id=str(raw.get("id", "")), name=str(raw.get("name", "")), + arguments=dict(args) if isinstance(args, dict) else {}, + preview=ToolPreview.from_dict(raw.get("preview")), + ) + + +def _parse_tool_finished(raw: Dict[str, Any]) -> ToolCallFinishedEvent: + """Rebuild a ``tool_result`` event; the optional file keys may be absent.""" + return ToolCallFinishedEvent( + call_id=str(raw.get("id", "")), name=str(raw.get("name", "")), + ok=bool(raw.get("ok", False)), output=str(raw.get("output", "")), + path=str(raw.get("path", "") or ""), produced=raw.get("produced") or (), + ) + + +# One parser per wire name. A table (rather than an if/elif chain) keeps adding +# an event a single-line change and makes the supported set introspectable. +_PARSERS = { + TextChunkEvent.EVENT_TYPE: lambda raw: TextChunkEvent(delta=str(raw.get("delta", ""))), + ReasoningChunkEvent.EVENT_TYPE: lambda raw: ReasoningChunkEvent( + delta=str(raw.get("delta", ""))), + AssistantMessageCompletedEvent.EVENT_TYPE: lambda raw: AssistantMessageCompletedEvent( + content=str(raw.get("content", ""))), + ToolCallStartedEvent.EVENT_TYPE: _parse_tool_started, + ToolOutputChunkEvent.EVENT_TYPE: lambda raw: ToolOutputChunkEvent( + call_id=str(raw.get("id", "")), name=str(raw.get("name", "")), + delta=str(raw.get("delta", ""))), + ToolCallFinishedEvent.EVENT_TYPE: _parse_tool_finished, + PlanUpdatedEvent.EVENT_TYPE: lambda raw: PlanUpdatedEvent( + steps=_plan_steps_from_legacy(raw.get("steps"))), + NoticeEvent.EVENT_TYPE: lambda raw: NoticeEvent( + text=str(raw.get("text", "")), level=str(raw.get("level", NOTICE_INFO))), + OutputsAddedEvent.EVENT_TYPE: lambda raw: OutputsAddedEvent(paths=raw.get("paths") or ()), + OutputsRemovedEvent.EVENT_TYPE: lambda raw: OutputsRemovedEvent(paths=raw.get("paths") or ()), + HistoryReadyEvent.EVENT_TYPE: lambda raw: HistoryReadyEvent( + session_id=str(raw.get("session_id", ""))), + TurnCompletedEvent.EVENT_TYPE: lambda raw: TurnCompletedEvent( + final_text=str(raw.get("final_text", "")), steps_used=int(raw.get("steps_used", 0) or 0), + cancelled=bool(raw.get("cancelled", False)), + budget_exhausted=bool(raw.get("budget_exhausted", False))), + ErrorEvent.EVENT_TYPE: lambda raw: ErrorEvent( + message=str(raw.get("message", "")), recoverable=bool(raw.get("recoverable", False))), +} + + +def from_legacy_dict(payload: Any) -> Optional[AgentEvent]: + """Parse an emitted dict into a typed event, or ``None`` if it isn't ours. + + ``None`` (rather than an exception) is the contract that makes incremental + adoption possible: a bridge sitting between the runtime and the widget can + type the events it recognises and forward everything else — Co4E's node + events, or anything a future emitter adds — completely untouched. + """ + if not isinstance(payload, dict): + return None + parser = _PARSERS.get(str(payload.get("type", ""))) + return parser(payload) if parser is not None else None + + +__all__ = ["from_legacy_dict"] diff --git a/domain/agents/agent_result.py b/domain/agents/agent_result.py new file mode 100644 index 0000000..a6e26ae --- /dev/null +++ b/domain/agents/agent_result.py @@ -0,0 +1,86 @@ +"""What one finished turn produced (R04-T03). + +The outcome of a turn is currently spread over three shapes: ``run_cowork`` +returns the mutated message list, ``task_executors._run_agent`` returns a +``(answer_text, timed_out, incomplete_reason)`` tuple, and the UI reconstructs +the rest (did it get cancelled? did it hit the ceiling?) from side effects. Each +caller therefore knows a slightly different amount about the same turn. + +:class:`AgentResult` is the single answer. Frozen, like the request that started +the turn, so a result cannot be edited into disagreeing with what actually +happened. + +Layer rules (``docs/architecture/ADR-001-layered-architecture.md``): domain +layer — standard library plus sibling domain types only. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Dict, Tuple + +from .agent_event import PlanStep, TurnCompletedEvent + + +@dataclass(frozen=True) +class AgentResult: + """The outcome of one conversation turn.""" + + # The conversation AFTER the turn (system prompt, history, the new user + # message, every assistant reply and tool result). + messages: Tuple[Dict[str, Any], ...] = () + steps_used: int = 0 # provider calls this turn consumed + cancelled: bool = False # the user pressed Stop + budget_exhausted: bool = False # stopped at effective_max_steps + # The agent's final checklist, so a caller can ask "did it really finish?" + # (``core/plan.py::plan_incomplete_reason``) without replaying the events. + plan_steps: Tuple[PlanStep, ...] = () + # Non-empty when the turn ended on a failure. A string rather than the + # exception: the domain layer must not depend on where the error came from, + # and the message is what every consumer (bubble, error.txt, audit) shows. + error: str = "" + + def __post_init__(self) -> None: + """Freeze the collections the runtime hands over. + + Both arrive as live lists that the caller keeps appending to after the + turn (the UI merges messages back into its own history), so copying here + is what keeps a result a record rather than a moving target. + """ + object.__setattr__(self, "messages", tuple(self.messages or ())) + object.__setattr__(self, "plan_steps", tuple(self.plan_steps or ())) + + @property + def final_text(self) -> str: + """The answer to show the user. + + Scans backwards for the last assistant message with real content, which + is not the same as ``messages[-1]``: a turn that was cancelled or that + ran out of steps mid-loop ends on a tool message, and a reasoning-only + reply leaves a blank assistant message behind. Same rule as + ``core/task_executors.py::_last_assistant_text``, which this replaces. + """ + for message in reversed(self.messages): + if message.get("role") == "assistant" and (message.get("content") or "").strip(): + return str(message["content"]) + return "" + + @property + def ok(self) -> bool: + """Whether the turn ran to a normal end. + + Hitting the step ceiling still counts as ok: the agent did work and + produced an answer, it just was not allowed to keep going — which the + transcript says in its own note rather than by failing the turn. + """ + return not self.error and not self.cancelled + + def to_turn_completed_event(self) -> TurnCompletedEvent: + """The end-of-turn event carrying this outcome to subscribers.""" + return TurnCompletedEvent( + final_text=self.final_text, steps_used=self.steps_used, + cancelled=self.cancelled, budget_exhausted=self.budget_exhausted, + ) + + +__all__ = ["AgentResult"] diff --git a/domain/agents/conversation_execution_request.py b/domain/agents/conversation_execution_request.py new file mode 100644 index 0000000..9b106e3 --- /dev/null +++ b/domain/agents/conversation_execution_request.py @@ -0,0 +1,222 @@ +"""The immutable snapshot of ONE chat turn (R04-T01). + +Today a turn's inputs live in a closure plus a 15-key ``ctx`` dict built inside +``ui/chat_panel.py::_start_turn``, and the worker thread reads the widget back +(``self._model``, ``self.title``, ``self.project_id``) while it runs. That is +the mechanism behind the whole class of "I changed the model mid-answer and the +running turn behaved oddly" reports: the turn has no snapshot of its own, so +every later click on the UI is visible to work already in flight. + +:class:`ConversationExecutionRequest` is that missing snapshot. Everything the +runtime needs for one turn is captured once, on the UI thread, at submit time, +and then handed to code that runs on a worker thread. Frozen, so no caller — +widget or service — can retroactively change a decision the turn already acted +on. + +Layer rules (see ``docs/architecture/ADR-001-layered-architecture.md``): this is +the domain layer, so standard library only. No PySide6, no ``requests``, no +filesystem access, and deliberately no import of ``core/*`` — a request only +*describes* a turn; running it is the application layer's job +(``application/conversations/``). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Any, Dict, Optional, Tuple + +# Separator between an instruction prefix (a ``/skill`` block, an ``/agent`` +# persona) and the user's own request. Kept as a constant because the prefix is +# assembled in the presentation layer while the body is only known later on the +# worker thread — both halves must agree on the exact separator or the model +# sees a different prompt shape than it did before this refactor. +PREFIX_SEPARATOR = "\n\n---\n\n" + + +@dataclass(frozen=True) +class ConversationExecutionRequest: + """Everything needed to execute one conversation turn. + + Frozen for the reason above; use :meth:`with_model` / :meth:`with_output_dir` + to derive an adjusted copy rather than mutating one another thread may be + reading. + + Note on depth: ``messages`` is a *shallow* snapshot (a tuple holding the + same message dicts the caller passed). That matches the existing + ``snapshot = list(self.messages)`` semantics in ``_start_turn`` exactly — + the turn is protected from the history list being appended to or replaced, + which is what actually happens between turns. Making it deep would silently + change how ``_finalize_turn`` merges the turn's messages back, so the + stronger guarantee is left to R04-T03 where that merge moves. + """ + + # -- identity ------------------------------------------------------- # + turn_id: str # unique within a session ("t1", "t2", ...) + session_id: str # the conversation this turn belongs to + surface: str = "cowork" # routing/mode key: "cowork" | "co4e" | "ai_edit" + project_id: str = "" # workspace the turn is confined to + title: str = "" # conversation title; also names saved files + + # -- what the user asked -------------------------------------------- # + # The typed request, already stripped of any ``/skill`` or ``/agent`` + # directive (those become ``instruction_prefix``). + prompt: str = "" + instruction_prefix: str = "" # skill rules + agent persona for this turn + # Prepended when the model/agent was switched mid-conversation, asking the + # model to re-check the previous step before continuing. Invisible in the + # chat bubble — it only travels in the payload sent to the provider. + review_note: str = "" + # Attachment PATHS, not their text: extracting a .docx can pip-install a + # parser or shell out to LibreOffice, which must not run on the UI thread. + # The runtime reads them later and passes the result to :meth:`user_content`. + attachments: Tuple[str, ...] = () + # Conversation history as of submit time; the new user message is NOT part + # of it (the runtime appends it once the body is composed). + messages: Tuple[Dict[str, Any], ...] = () + + # -- which model answers -------------------------------------------- # + # Already resolved upstream: an Admin-agent pin, the tab's own picker, or a + # routing override published by ``RoutingApplicationService`` (R03). The + # runtime does not re-decide, so a switch cannot land mid-turn. + provider_id: str = "" + model: str = "" # "" = the provider's configured default + + # -- standing instructions ------------------------------------------ # + project_context: str = "" # Claude-Projects-style shared instructions + session_notes: str = "" # e.g. files this conversation already produced + + # -- tool scope and turn limits -------------------------------------- # + # None = every enabled built-in tool. An explicit (possibly empty) tuple + # restricts the ADVERTISED tools, which is how a "read-only" step is made + # literally unable to write. + allowed_tools: Optional[Tuple[str, ...]] = None + max_steps: int = 30 # interactive cap + completion_max_steps: int = 200 # runaway ceiling for run-to-completion work + run_to_completion: bool = False # Co4E flow steps need the higher ceiling + enforce_rules: bool = True # False for sandboxed Co4E runs + gate_mode: str = "auto" # "confirm" -> ask before run_command/install + agent_role: str = "cowork" # audit-log attribution ("cowork" | "task" | ...) + + # -- where its files go ---------------------------------------------- # + output_dir: Optional[Path] = None # this turn's isolated sandbox + home_output_root: Optional[Path] = None # conversation Output root to promote into + + # -- unattended execution (Schedule Task) ----------------------------- # + unattended: bool = False # no human watching; plan tracking is enforced + timeout_sec: Optional[int] = None # None = no wall-clock limit + + # Escape hatch for surface-specific data a future task needs to thread + # through without another schema change (same role as + # ``ProviderDescriptor.extras``). + extras: Dict[str, Any] = field(default_factory=dict) + + # -- validation / normalisation --------------------------------------- # + def __post_init__(self) -> None: + """Reject unusable requests and freeze the mutable inputs. + + Validation lives here (not at the call site) so a request that exists is + always safe to key by: the audit log, the History autosave and the + per-turn output folder are all named from ``session_id``/``turn_id``. + + Normalisation matters just as much: the caller hands us the composer's + own attachment LIST and the live history LIST, and both get cleared or + appended to for the next turn. Copying them into tuples here is what + actually makes the snapshot a snapshot. ``object.__setattr__`` is the + standard way to do this in a frozen dataclass. + """ + if not (self.turn_id or "").strip(): + raise ValueError("ConversationExecutionRequest.turn_id must not be empty") + if not (self.session_id or "").strip(): + raise ValueError("ConversationExecutionRequest.session_id must not be empty") + + object.__setattr__(self, "attachments", tuple(self.attachments or ())) + object.__setattr__(self, "messages", tuple(self.messages or ())) + # None must survive: it means "no restriction", while an empty tuple + # means "deny every built-in tool" — two very different turns. + if self.allowed_tools is not None: + object.__setattr__(self, "allowed_tools", tuple(self.allowed_tools)) + # Accept str paths so a call site holding a config value does not have to + # wrap it; everything downstream can then assume Path. + for name in ("output_dir", "home_output_root"): + value = getattr(self, name) + if value is not None and not isinstance(value, Path): + object.__setattr__(self, name, Path(value)) + + # -- derived turn policy ---------------------------------------------- # + @property + def has_prompt(self) -> bool: + """Whether the user actually typed something (an attachment-only turn + legitimately has none). Mirrors ``RoutingRequest.has_prompt`` so both + DTOs answer the "is there anything to work with?" question the same way. + """ + return bool((self.prompt or "").strip()) + + @property + def effective_max_steps(self) -> int: + """The tool-use budget for this turn. + + Run-to-completion work (a Co4E flow step whose single instruction may + need many tool calls) gets the higher ceiling; interactive chat keeps the + tight cap. Either way the turn still ends the moment the model stops + calling tools — this is only the runaway limit. + """ + return self.completion_max_steps if self.run_to_completion else self.max_steps + + @property + def requires_permission_gate(self) -> bool: + """Whether ``run_command``/``install_package`` must be approved first. + + Resolved by the caller (per-workspace Auto-run override, else the global + "confirm before running commands" setting) and frozen here, so toggling + the setting mid-turn cannot change the rules the turn started under. + """ + return self.gate_mode == "confirm" + + # -- prompt composition ------------------------------------------------ # + def user_content(self, body: str = "") -> str: + """The exact ``content`` to send as this turn's user message. + + ``body`` is the request text AFTER attachment extraction, which happens + on the worker thread — hence a method taking it as an argument rather + than a stored field. The assembly order reproduces the closure in + ``_start_turn`` byte for byte, because changing what a model receives is + a behaviour change, not a refactor: + + 1. session notes are appended after the body; + 2. the instruction prefix goes in front, behind a fixed separator; + 3. the model-switch review note goes ahead of everything. + """ + content = body or "" + notes = self.session_notes or "" + if notes: + # Guard the empty-body case (attachment-only turn) so the payload + # never opens with a stray blank line. + content = f"{content}\n\n{notes}" if content else notes + prefix = self.instruction_prefix or "" + if prefix: + content = f"{prefix}{PREFIX_SEPARATOR}{content}" + review = self.review_note or "" + if review: + content = f"{review}\n\n{content}" + return content + + # -- derivation --------------------------------------------------------- # + def with_model(self, provider_id: str = "", model: str = "") -> "ConversationExecutionRequest": + """A copy pinned to another provider/model. + + Needed when a decision lands between building the request and running it + (a routing override, an Admin-agent pin). Deriving a new request keeps + the "one turn, one immutable snapshot" rule intact instead of patching a + request another thread may already hold. + """ + return replace(self, provider_id=provider_id or self.provider_id, + model=model or self.model) + + def with_output_dir(self, output_dir) -> "ConversationExecutionRequest": + """A copy writing into a different sandbox — used when the caller only + learns the per-turn folder after the request is assembled.""" + return replace(self, output_dir=output_dir) + + +__all__ = ["PREFIX_SEPARATOR", "ConversationExecutionRequest"] diff --git a/tests/integration/test_agent_event_bridge.py b/tests/integration/test_agent_event_bridge.py new file mode 100644 index 0000000..5914636 --- /dev/null +++ b/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 diff --git a/tests/unit/test_agent_event.py b/tests/unit/test_agent_event.py new file mode 100644 index 0000000..f29a3d4 --- /dev/null +++ b/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"),)) diff --git a/tests/unit/test_agent_result.py b/tests/unit/test_agent_result.py new file mode 100644 index 0000000..609893a --- /dev/null +++ b/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"),) diff --git a/tests/unit/test_conversation_execution_request.py b/tests/unit/test_conversation_execution_request.py new file mode 100644 index 0000000..1ba6be5 --- /dev/null +++ b/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"