diff --git a/application/conversations/__init__.py b/application/conversations/__init__.py index d4a4b1c..edf4398 100644 --- a/application/conversations/__init__.py +++ b/application/conversations/__init__.py @@ -1,5 +1,8 @@ """Conversation use case: the lifecycle of one agent turn (EPIC R04).""" -from .conversation_application_service import ConversationApplicationService +from .conversation_application_service import ( + ConversationApplicationService, + TurnResult, +) -__all__ = ["ConversationApplicationService"] +__all__ = ["ConversationApplicationService", "TurnResult"] diff --git a/application/conversations/conversation_application_service.py b/application/conversations/conversation_application_service.py new file mode 100644 index 0000000..4cde359 --- /dev/null +++ b/application/conversations/conversation_application_service.py @@ -0,0 +1,328 @@ +"""ConversationApplicationService - the turn lifecycle, outside the widget (R04-T03). + +What this replaces +------------------ +The lifecycle of one Cowork turn is currently spread across a closure inside +``ui/cowork_tab.py::build_job`` and a second, near-identical assembly inside +``core/task_executors.py::_run_agent``. Both: + +* read live UI/config state from a worker thread, +* build the provider, the MCP tool set and the project context by hand, +* call ``core.chat_agent.run_cowork`` with a dozen positional-ish arguments, +* consume untyped event dicts. + +Two copies means a fix to one path (say, promoting output files on failure) +silently misses the other. This service is the single implementation: it takes +an immutable :class:`ConversationExecutionRequest`, runs the turn, and reports +typed :class:`AgentEvent` objects. + +What it deliberately does NOT do +-------------------------------- +It does not re-implement the agent loop. ``run_cowork`` stays the engine +(strangler fig, ADR-001 section 4) and keeps its characterization tests +(``tests/characterization/test_run_cowork.py``). This layer owns the parts that +were tangled into the UI: assembling the call, translating events, and giving a +turn a well-defined end. + +Pure Python: no Qt import, no config access. Everything it needs arrives through +constructor callbacks, so the same service runs a turn from a chat panel, from +the scheduler, or from a test. +""" +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Tuple + +from cowork_local.domain.agents.agent_event import ( + AgentEvent, + ErrorEvent, + TurnCompletedEvent, + collect_text, + event_from_dict, +) +from cowork_local.domain.agents.conversation_execution_request import ( + ConversationExecutionRequest, +) + +logger = logging.getLogger("cowork_local.conversations") + +# Presentation/scheduler supplies these. Kept as plain callables (not objects) +# so a test can wire the service with three lambdas. +EventCallback = Callable[[AgentEvent], None] +CancelFn = Callable[[], bool] +ProviderFactory = Callable[[str, str], Any] # (provider_id, model) -> Provider +ToolSourceFactory = Callable[[], Tuple[Any, Any]] # () -> (extra_tools, extra_executor) +GateFactory = Callable[[ConversationExecutionRequest], Any] # -> PermissionGate or None + + +@dataclass +class TurnResult: + """What a finished turn produced. + + ``messages`` is the conversation AFTER the turn (system prompt inserted, + assistant and tool messages appended) - the caller persists this as the new + history. ``final_text`` is the visible answer, reasoning excluded. + """ + + request: ConversationExecutionRequest + messages: List[Dict[str, Any]] = field(default_factory=list) + events: List[AgentEvent] = field(default_factory=list) + final_text: str = "" + cancelled: bool = False + error: str = "" + # The original exception, kept alongside its message so a caller that needs + # to preserve legacy failure handling can re-raise the SAME object rather + # than a lookalike (SecurityBlocked, for instance, carries context that a + # re-wrapped RuntimeError would lose). + exception: Optional[BaseException] = None + + @property + def ok(self) -> bool: + """True when the turn completed without an error and without a Stop.""" + return not self.error and not self.cancelled + + def raise_if_failed(self) -> None: + """Re-raise the turn's failure, if any. + + Callers that already have failure handling built around an exception + (the Qt worker turns one into its ``failed`` signal) use this to keep + that path intact while still getting a TurnResult on success.""" + if self.exception is not None: + raise self.exception + + def output_dir(self) -> Optional[Path]: + """This turn's output folder, or None when it could not write files.""" + return Path(self.request.output_dir) if self.request.output_dir else None + + +class ConversationApplicationService: + """Runs one agent turn from an immutable request. + + Args: + provider_factory: ``(provider_id, model) -> Provider``. Production passes + ``AppContext.build_provider_for``; tests pass a lambda returning a + :class:`FakeProvider`. + tool_source: ``() -> (extra_tools, extra_executor)`` for MCP/connector + tools. Optional - a turn with no external tools passes nothing. + gate_factory: ``(request) -> PermissionGate | None``, consulted when the + request asks to confirm commands. Optional for the same reason. + runner: the turn engine. Defaults to ``core.chat_agent.run_cowork``, + imported lazily so this module stays importable (and testable) + without pulling in the whole legacy tool stack. + security_config: the app config the security layers read. ``None`` + disables them, which is what headless callers already rely on. + """ + + def __init__( + self, + provider_factory: ProviderFactory, + *, + tool_source: Optional[ToolSourceFactory] = None, + gate_factory: Optional[GateFactory] = None, + runner: Optional[Callable[..., Any]] = None, + security_config: Any = None, + ) -> None: + self._provider_factory = provider_factory + self._tool_source = tool_source + self._gate_factory = gate_factory + self._runner = runner + self._security_config = security_config + + # -- main entry point -------------------------------------------------- # + def run_turn( + self, + request: ConversationExecutionRequest, + on_event: Optional[EventCallback] = None, + cancel: Optional[CancelFn] = None, + ) -> TurnResult: + """Execute one turn and return everything it produced. + + Never raises: a provider or tool failure becomes an :class:`ErrorEvent` + plus ``TurnResult.error``. Callers run this on a worker thread and have + no good way to handle an exception crossing that boundary - today an + escaped error kills the worker and the UI just stops updating, with no + message shown. + + Exactly one :class:`TurnCompletedEvent` is always emitted last, whether + the turn succeeded, failed or was cancelled. That is the end-of-turn + signal the legacy engine never had. + """ + return self.execute_turn(self.begin_turn(request), on_event=on_event, cancel=cancel) + + def begin_turn(self, request: ConversationExecutionRequest) -> TurnResult: + """Create the (still empty) result a turn will fill in. + + Exposed separately from :meth:`run_turn` because some callers need the + LIVE message list while the turn is running, not only afterwards: the + scheduler re-saves the conversation to History after every assistant + message so a long unattended run shows live progress when reopened. + Handing them ``result.messages`` - the very list the engine appends to - + is what makes that possible without leaking the engine into the caller. + """ + return TurnResult(request=request, messages=request.message_list()) + + def execute_turn( + self, + result: TurnResult, + on_event: Optional[EventCallback] = None, + cancel: Optional[CancelFn] = None, + ) -> TurnResult: + """Run a turn previously created by :meth:`begin_turn`. See + :meth:`run_turn` for the error/cancellation contract.""" + request = result.request + emit = self._make_emitter(result, on_event) + cancel = cancel or (lambda: False) + + try: + self._execute(request, result, emit, cancel) + except Exception as exc: # noqa: BLE001 - see docstring + result.error = str(exc) or exc.__class__.__name__ + result.exception = exc + logger.exception("turn %s failed", request.turn_id) + emit(ErrorEvent(message=result.error, + recoverable=self._is_recoverable(exc))) + + result.cancelled = bool(cancel()) + result.final_text = collect_text(result.events) or self._last_assistant_text(result.messages) + emit(TurnCompletedEvent(content=result.final_text, cancelled=result.cancelled)) + return result + + # -- internals --------------------------------------------------------- # + def _execute(self, request: ConversationExecutionRequest, result: TurnResult, + emit: Callable[[AgentEvent], None], cancel: CancelFn) -> None: + """Assemble the engine call from the request snapshot and run it.""" + provider = self._provider_factory(request.provider, request.model) + extra_tools, extra_executor = self._resolve_tools() + gate = self._resolve_gate(request) + + # The engine speaks untyped dicts; bridge them into typed events at this + # single point rather than at every consumer. + def legacy_emit(payload: Dict[str, Any]) -> None: + event = event_from_dict(payload) + if event is not None: + emit(event) + + run = self._resolve_runner() + run( + provider, + result.messages, # mutated in place by the engine, as before + self._output_dir(request), + legacy_emit, + cancel, + title=request.title, + extra_tools=extra_tools, + extra_executor=extra_executor, + project_context=request.project_context, + security_config=self._security_config, + gate=gate, + allowed_tools=list(request.allowed_tools) if request.allowed_tools is not None else None, + max_steps=request.max_steps, + run_to_completion=request.run_to_completion, + completion_max_steps=request.completion_max_steps, + enforce_rules=request.enforce_rules, + **self._role_kwargs(request), + ) + + @staticmethod + def _make_emitter(result: TurnResult, + on_event: Optional[EventCallback]) -> Callable[[AgentEvent], None]: + """Record every event on the result AND forward it to the caller. + + Recording is unconditional so a headless caller (the scheduler) can read + the full event list afterwards without having to supply a callback just + to collect it - which is exactly what task_executors does today with an + ad-hoc list. + """ + def emit(event: AgentEvent) -> None: + result.events.append(event) + if on_event is None: + return + try: + on_event(event) + except Exception: # noqa: BLE001 + # A consumer that throws (a closing widget, say) must not abort + # the turn that is feeding it. + logger.debug("event consumer raised for %s", event.type, exc_info=True) + return emit + + def _resolve_runner(self) -> Callable[..., Any]: + """The turn engine, imported lazily on first use.""" + if self._runner is None: + from cowork_local.core.chat_agent import run_cowork + + self._runner = run_cowork + return self._runner + + def _resolve_tools(self) -> Tuple[Any, Any]: + """MCP/connector tools for this turn, or ``(None, None)``. + + A failure here degrades to "no external tools" rather than failing the + turn: an MCP server that will not start must not stop the user from + chatting, which is the behaviour the chat panel already relies on. + """ + if self._tool_source is None: + return None, None + try: + return self._tool_source() + except Exception: # noqa: BLE001 + logger.warning("tool source unavailable - running without external tools", + exc_info=True) + return None, None + + def _resolve_gate(self, request: ConversationExecutionRequest) -> Any: + """The permission gate, when this turn asked to confirm commands.""" + if not request.confirm_commands or self._gate_factory is None: + return None + return self._gate_factory(request) + + @staticmethod + def _output_dir(request: ConversationExecutionRequest) -> Path: + """The turn's output folder as a Path. + + The request holds it as a string to stay serialisable; converting at the + single point of use keeps that decision from leaking into every caller. + """ + return Path(request.output_dir) if request.output_dir else Path.cwd() + + @staticmethod + def _role_kwargs(request: ConversationExecutionRequest) -> Dict[str, Any]: + """``agent_role`` only when the request set one. + + Omitted otherwise so the engine applies its own default (the interactive + Cowork role) instead of being handed an empty string, which would land + in the audit log as an unattributed tool call. + """ + return {"agent_role": request.agent_role} if request.agent_role else {} + + @staticmethod + def _last_assistant_text(messages: List[Dict[str, Any]]) -> str: + """Fallback answer text when no text events were seen. + + A turn whose whole answer arrived in one non-streamed message still has + to report a final answer - the scheduler writes it into output.md, and + an empty string there reads as "(no output)". + """ + for message in reversed(messages): + if message.get("role") == "assistant" and (message.get("content") or "").strip(): + return str(message["content"]) + return "" + + @staticmethod + def _is_recoverable(exc: Exception) -> bool: + """Whether the user can act on this failure themselves. + + "Model not found" is the motivating case: the chat panel restores the + typed message into the composer so the user can switch model and resend + instead of retyping it (see providers/base.py::MODEL_NOT_FOUND_HINT). + """ + try: + from cowork_local.providers.base import is_model_not_found_error + + return bool(is_model_not_found_error(str(exc))) + except Exception: # noqa: BLE001 + return False + + +__all__ = ["ConversationApplicationService", "TurnResult"] diff --git a/core/task_executors.py b/core/task_executors.py index 3d8bf43..25cbf8a 100644 --- a/core/task_executors.py +++ b/core/task_executors.py @@ -248,9 +248,34 @@ def _run_agent(ctx, task_type: str, prompt: str, out_dir: Path, "'error' (not silently skip it) if it genuinely can't be completed.\n\n" f"{prompt}" ) - messages = [{"role": "user", "content": prompt}] + # One immutable snapshot of this run, then the shared turn service (R04-T05). + # The Schedule Task path used to assemble the run_cowork call itself, in + # parallel with ui/cowork_tab.py doing the same thing slightly differently - + # so a fix to one path silently missed the other. Both now go through + # ConversationApplicationService. + from ..application.conversations import ConversationApplicationService + from ..domain.agents import ConversationExecutionRequest + session_id = new_session_id() project_id = project.project_id if project is not None else "" + project_context = projects.project_context_text(project) + conversation_service = ConversationApplicationService( + # The provider was already resolved above (admin agent / per-task + # override / machine default), so the factory just hands it back. + lambda _provider_id, _model: provider, + security_config=ctx.config, + ) + turn = conversation_service.begin_turn(ConversationExecutionRequest.create( + prompt, [{"role": "user", "content": prompt}], + output_dir=str(out_dir), session_id=session_id, surface="task", + title=title, project_id=project_id, project_context=project_context, + # Tags every tool call in the audit log as a scheduled task rather than + # as the interactive Cowork tab. + agent_role=agent_roles.TASK, + )) + # The LIVE list the engine appends to - History is re-saved from it after + # every assistant message so a long run shows progress when reopened. + messages = turn.messages _save_history_session(ctx, task_type, title, messages, session_id, project_id) # Tell the scheduler the session now genuinely EXISTS on disk — it # refreshes History on this, not on the earlier "task_started" signal @@ -269,14 +294,21 @@ def _run_agent(ctx, task_type: str, prompt: str, out_dir: Path, elif ev.get("type") == "plan_set": last_plan_steps[:] = ev.get("steps") or [] - project_context = projects.project_context_text(project) watched_cancel, timed_out = _cancel_with_timeout(cancel, timeout_sec) try: if task_type == "cowork": - from .chat_agent import run_cowork - run_cowork(provider, messages, out_dir, emit_and_autosave, watched_cancel, - security_config=ctx.config, agent_role=agent_roles.TASK, - project_context=project_context) + # Typed events are rendered back into the legacy dict shape this + # module's autosave/plan tracking already consumes; it moves to + # AgentEvent directly once the scheduler UI migrates (EPIC R07/R08). + result = conversation_service.execute_turn( + turn, + on_event=lambda event: emit_and_autosave(event.to_dict()), + cancel=watched_cancel, + ) + # This module's callers handle a failed run through an exception + # (execute_task writes error.txt from it), so re-raise the ORIGINAL + # error rather than reporting a silently empty answer. + result.raise_if_failed() else: from .code_agent import run_code limits, block_network = agent_security.sandbox_settings(ctx.config) diff --git a/docs/refactor/Refactoring_Checklist.md b/docs/refactor/Refactoring_Checklist.md index 41ceff8..539da79 100644 --- a/docs/refactor/Refactoring_Checklist.md +++ b/docs/refactor/Refactoring_Checklist.md @@ -81,16 +81,16 @@ * **Team chịu trách nhiệm**: 🔵 **Team Duy** (Chủ trì) * **Mục tiêu**: Đóng gói input turn chat thành `ConversationExecutionRequest` bất biến, điều phối vòng đời qua `ConversationApplicationService` và phát sinh sự kiện `AgentEvent` có định kiểu. -- [ ] **R04-T01 (Team Duy)**: Định nghĩa immutable dataclass `ConversationExecutionRequest` ➔ `domain/agents/conversation_execution_request.py` - *Start: `____-__-__ __:__` | End: `____-__-__ __:__`* -- [ ] **R04-T02 (Team Duy)**: Chuẩn hóa các sự kiện `AgentEvent` (TextChunk, ToolCallStarted, ToolCallResult, Error) ➔ `domain/agents/agent_event.py` - *Start: `____-__-__ __:__` | End: `____-__-__ __:__`* -- [ ] **R04-T03 (Team Duy)**: Xây dựng `ConversationApplicationService` điều phối thực thi từ `core/chat_agent.py` ➔ `application/conversations/conversation_application_service.py` - *Start: `____-__-__ __:__` | End: `____-__-__ __:__`* -- [ ] **R04-T04 (Team Duy)**: Di chuyển `ui/cowork_tab.py::build_job` sang sử dụng `ConversationExecutionRequest` - *Start: `____-__-__ __:__` | End: `____-__-__ __:__`* -- [ ] **R04-T05 (Team Duy)**: Di chuyển `core/task_executors.py` sang dùng chung `ConversationApplicationService` - *Start: `____-__-__ __:__` | End: `____-__-__ __:__`* +- [x] **R04-T01 (Team Duy)**: Định nghĩa immutable dataclass `ConversationExecutionRequest` ➔ `domain/agents/conversation_execution_request.py` + *Start: `2026-08-21 10:23` | End: `2026-08-21 10:25`* +- [x] **R04-T02 (Team Duy)**: Chuẩn hóa các sự kiện `AgentEvent` (TextChunk, ToolCallStarted, ToolCallResult, Error) ➔ `domain/agents/agent_event.py` + *Start: `2026-08-21 10:22` | End: `2026-08-21 10:23`* +- [x] **R04-T03 (Team Duy)**: Xây dựng `ConversationApplicationService` điều phối thực thi từ `core/chat_agent.py` ➔ `application/conversations/conversation_application_service.py` + *Start: `2026-08-21 10:25` | End: `2026-08-21 10:27`* +- [x] **R04-T04 (Team Duy)**: Di chuyển `ui/cowork_tab.py::build_job` sang sử dụng `ConversationExecutionRequest` + *Start: `2026-08-21 10:27` | End: `2026-08-21 10:31`* +- [x] **R04-T05 (Team Duy)**: Di chuyển `core/task_executors.py` sang dùng chung `ConversationApplicationService` + *Start: `2026-08-21 10:28` | End: `2026-08-21 10:30`* --- diff --git a/domain/agents/__init__.py b/domain/agents/__init__.py index dee54b3..4a6ddef 100644 --- a/domain/agents/__init__.py +++ b/domain/agents/__init__.py @@ -1,2 +1,48 @@ -"""Domain entities for one agent turn: the request snapshot and the event +"""Domain entities for one agent turn: the request snapshot and the typed event stream it produces (EPIC R04).""" + +from .agent_event import ( + AgentEvent, + AssistantDoneEvent, + ErrorEvent, + HistoryReadyEvent, + NoticeEvent, + OutputsAddedEvent, + OutputsRemovedEvent, + PlanUpdatedEvent, + ReasoningChunkEvent, + TextChunkEvent, + ToolCallFinishedEvent, + ToolCallStartedEvent, + ToolOutputEvent, + TurnCompletedEvent, + collect_text, + event_from_dict, + tool_calls, +) +from .conversation_execution_request import ( + ConversationExecutionRequest, + new_turn_id, +) + +__all__ = [ + "ConversationExecutionRequest", + "new_turn_id", + "AgentEvent", + "TextChunkEvent", + "ReasoningChunkEvent", + "AssistantDoneEvent", + "PlanUpdatedEvent", + "ToolCallStartedEvent", + "ToolOutputEvent", + "ToolCallFinishedEvent", + "OutputsAddedEvent", + "OutputsRemovedEvent", + "NoticeEvent", + "HistoryReadyEvent", + "TurnCompletedEvent", + "ErrorEvent", + "event_from_dict", + "collect_text", + "tool_calls", +] diff --git a/domain/agents/agent_event.py b/domain/agents/agent_event.py new file mode 100644 index 0000000..96628a6 --- /dev/null +++ b/domain/agents/agent_event.py @@ -0,0 +1,370 @@ +"""AgentEvent - the typed event stream one agent turn produces (R04-T02). + +Today the turn engine talks to its caller through untyped dicts:: + + emit({"type": "tool_result", "id": tc_id, "name": name, + "ok": result.get("ok", False), "output": result.get("output", "")}) + +and every consumer re-discovers the vocabulary by reading the producer. There +are eleven such shapes across ``core/chat_agent.py``, ``core/code_agent.py`` and +``core/task_executors.py``; a consumer that misspells ``"tool_result"`` or reads +``"result"`` instead of ``"output"`` fails silently, at runtime, only for the +tool path that triggers it. + +This module makes the vocabulary explicit. Each event is a frozen dataclass, so: + +* the set of possible events is enumerable (see :data:`EVENT_TYPES`); +* a field name typo is an ``AttributeError`` at the point of use, not a silently + missing chat bubble; +* an event can cross a thread boundary safely - it cannot be mutated after the + producer hands it over, which is exactly what the Qt-signal seam needs. + +Bridging with the legacy dicts is deliberate and two-way: :func:`event_from_dict` +adapts what ``run_cowork`` emits today, and :meth:`AgentEvent.to_dict` renders an +event back into the legacy shape so existing widgets keep working untouched +while the presentation layer migrates screen by screen (EPIC R08). + +Pure domain code: stdlib only, no Qt, no I/O. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple + + +@dataclass(frozen=True) +class AgentEvent: + """Base class for everything a turn can report. + + ``type`` is the legacy string tag, kept as a class attribute so the bridge + functions can round-trip an event without a separate mapping table. + """ + + type: str = field(init=False, default="event") + + def to_dict(self) -> Dict[str, Any]: + """Render into the legacy ``emit()`` dict shape.""" + return {"type": self.type} + + +# --------------------------------------------------------------------------- # +# Assistant output +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class TextChunkEvent(AgentEvent): + """One fragment of the visible answer, as it streams in.""" + + delta: str + type: str = field(init=False, default="text") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "delta": self.delta} + + +@dataclass(frozen=True) +class ReasoningChunkEvent(AgentEvent): + """One fragment of the model's PRIVATE reasoning. + + Drives the "Thinking" indicator only. Consumers must never append this to + the answer or persist it into conversation history - keeping it a distinct + type is what makes that mistake hard to make by accident. + """ + + delta: str + type: str = field(init=False, default="reasoning") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "delta": self.delta} + + +@dataclass(frozen=True) +class AssistantDoneEvent(AgentEvent): + """One assistant message finished. A turn with tool calls emits this once + per step, not once per turn - see :class:`TurnCompletedEvent`.""" + + content: str = "" + type: str = field(init=False, default="assistant_done") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "content": self.content} + + +# --------------------------------------------------------------------------- # +# Planning +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class PlanUpdatedEvent(AgentEvent): + """The agent rewrote its plan (the ``update_plan`` tool).""" + + steps: Tuple[Dict[str, Any], ...] = () + type: str = field(init=False, default="plan_set") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "steps": [dict(s) for s in self.steps]} + + +# --------------------------------------------------------------------------- # +# Tool lifecycle +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class ToolCallStartedEvent(AgentEvent): + """A tool call is about to run, with the preview shown to the user. + + Maps the legacy ``tool_proposed`` event. "Proposed" was a misnomer: by the + time it is emitted the call is already going to run unless a permission gate + rejects it, and the gate reports that as a finished call with ``ok=False``. + """ + + call_id: str + name: str + args: Dict[str, Any] = field(default_factory=dict) + preview: Optional[Dict[str, Any]] = None + type: str = field(init=False, default="tool_proposed") + + def to_dict(self) -> Dict[str, Any]: + out: Dict[str, Any] = {"type": self.type, "id": self.call_id, + "name": self.name, "args": dict(self.args)} + if self.preview is not None: + out["preview"] = dict(self.preview) + return out + + +@dataclass(frozen=True) +class ToolOutputEvent(AgentEvent): + """A line of live output from a running tool (command stdout, for example).""" + + call_id: str + name: str + delta: str + type: str = field(init=False, default="tool_output") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "id": self.call_id, "name": self.name, + "delta": self.delta} + + +@dataclass(frozen=True) +class ToolCallFinishedEvent(AgentEvent): + """A tool call ended, successfully or not. + + ``ok=False`` covers every failure mode alike - the tool raised, the sandbox + blocked it, or the user rejected it at the permission gate - because the + consumer's job is the same in all three: show the failure and let the model + react to it. + """ + + call_id: str + name: str + ok: bool = False + output: str = "" + path: str = "" # file the tool wrote, when it wrote one + produced: Tuple[str, ...] = () # extra artefacts (e.g. a generator's outputs) + type: str = field(init=False, default="tool_result") + + def to_dict(self) -> Dict[str, Any]: + out: Dict[str, Any] = {"type": self.type, "id": self.call_id, "name": self.name, + "ok": self.ok, "output": self.output} + if self.path: + out["path"] = self.path + if self.produced: + out["produced"] = list(self.produced) + return out + + +# --------------------------------------------------------------------------- # +# Output folder +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class OutputsAddedEvent(AgentEvent): + """Files appeared in the turn's output folder.""" + + paths: Tuple[str, ...] = () + type: str = field(init=False, default="outputs_added") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "paths": list(self.paths)} + + +@dataclass(frozen=True) +class OutputsRemovedEvent(AgentEvent): + """Files were cleaned up from the turn's output folder (intermediates).""" + + paths: Tuple[str, ...] = () + type: str = field(init=False, default="outputs_removed") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "paths": list(self.paths)} + + +@dataclass(frozen=True) +class NoticeEvent(AgentEvent): + """A UI-visible aside that is not part of the model's answer. + + Three producers today, all reachable from a normal turn: + ``core/agent_security.py`` (a request or command blocked by the security + layer), ``core/context_budget.py`` (the conversation was auto-compressed) + and the attachment readers (a file that could not be processed, plus live + "reading page X/Y" progress). + + ``level`` selects how the UI renders it: ``"progress"`` updates the thinking + indicator in place, anything else becomes a warning bubble. Dropping these + would silently hide security warnings from the user, which is why the type + exists rather than being folded into TextChunkEvent. + """ + + text: str + level: str = "info" + type: str = field(init=False, default="notice") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "level": self.level, "text": self.text} + + +@dataclass(frozen=True) +class HistoryReadyEvent(AgentEvent): + """A history session exists for this run and can be opened.""" + + session_id: str + type: str = field(init=False, default="history_ready") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "session_id": self.session_id} + + +# --------------------------------------------------------------------------- # +# Turn lifecycle - emitted by the application service, not by the legacy engine +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class TurnCompletedEvent(AgentEvent): + """The whole turn finished: no more events will follow. + + New in R04. The legacy engine has no end-of-turn signal at all, so every + consumer infers "done" from the worker thread finishing - which is why a + cancelled turn and a failed turn look identical to the UI today. + """ + + content: str = "" + cancelled: bool = False + type: str = field(init=False, default="turn_completed") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "content": self.content, "cancelled": self.cancelled} + + +@dataclass(frozen=True) +class ErrorEvent(AgentEvent): + """The turn failed. ``recoverable`` marks errors the user can act on + (pick another model, shorten the prompt) rather than a hard outage.""" + + message: str + recoverable: bool = False + type: str = field(init=False, default="error") + + def to_dict(self) -> Dict[str, Any]: + return {"type": self.type, "message": self.message, + "recoverable": self.recoverable} + + +# The legacy tag -> event class map. Also the authoritative list of what a turn +# can emit, which is what makes an exhaustive consumer possible for the first time. +EVENT_TYPES: Dict[str, type] = { + "text": TextChunkEvent, + "reasoning": ReasoningChunkEvent, + "assistant_done": AssistantDoneEvent, + "plan_set": PlanUpdatedEvent, + "tool_proposed": ToolCallStartedEvent, + "tool_start": ToolCallStartedEvent, + "tool_output": ToolOutputEvent, + "tool_result": ToolCallFinishedEvent, + "outputs_added": OutputsAddedEvent, + "outputs_removed": OutputsRemovedEvent, + "notice": NoticeEvent, + "history_ready": HistoryReadyEvent, + "turn_completed": TurnCompletedEvent, + "error": ErrorEvent, +} + + +def event_from_dict(payload: Mapping[str, Any]) -> Optional[AgentEvent]: + """Adapt one legacy ``emit()`` dict into a typed event. + + Returns ``None`` for an unknown tag instead of raising: the legacy engine is + still being refactored and may grow an event before this module knows about + it. Dropping an unrecognised event degrades the UI by one missing bubble; + raising here would abort a turn that had otherwise succeeded. + """ + kind = str(payload.get("type", "")) + cls = EVENT_TYPES.get(kind) + if cls is None: + return None + + if cls is TextChunkEvent or cls is ReasoningChunkEvent: + return cls(delta=str(payload.get("delta", ""))) + if cls is AssistantDoneEvent: + return AssistantDoneEvent(content=str(payload.get("content", ""))) + if cls is PlanUpdatedEvent: + return PlanUpdatedEvent(steps=tuple(payload.get("steps") or ())) + if cls is ToolCallStartedEvent: + return ToolCallStartedEvent( + call_id=str(payload.get("id", "")), name=str(payload.get("name", "")), + args=dict(payload.get("args") or {}), preview=payload.get("preview"), + ) + if cls is ToolOutputEvent: + return ToolOutputEvent(call_id=str(payload.get("id", "")), + name=str(payload.get("name", "")), + delta=str(payload.get("delta", ""))) + if cls is ToolCallFinishedEvent: + return ToolCallFinishedEvent( + call_id=str(payload.get("id", "")), name=str(payload.get("name", "")), + ok=bool(payload.get("ok", False)), output=str(payload.get("output", "")), + path=str(payload.get("path", "") or ""), + produced=tuple(payload.get("produced") or ()), + ) + if cls is OutputsAddedEvent or cls is OutputsRemovedEvent: + return cls(paths=tuple(str(p) for p in (payload.get("paths") or ()))) + if cls is NoticeEvent: + return NoticeEvent(text=str(payload.get("text", "")), + level=str(payload.get("level", "info"))) + if cls is HistoryReadyEvent: + return HistoryReadyEvent(session_id=str(payload.get("session_id", ""))) + if cls is TurnCompletedEvent: + return TurnCompletedEvent(content=str(payload.get("content", "")), + cancelled=bool(payload.get("cancelled", False))) + return ErrorEvent(message=str(payload.get("message", "")), + recoverable=bool(payload.get("recoverable", False))) + + +def collect_text(events: Sequence[AgentEvent]) -> str: + """Join every :class:`TextChunkEvent` - the visible answer, reasoning excluded. + + Provided here so no consumer has to re-derive "which events are the answer", + the question the untyped dicts made easy to get wrong. + """ + return "".join(e.delta for e in events if isinstance(e, TextChunkEvent)) + + +def tool_calls(events: Sequence[AgentEvent]) -> List[ToolCallFinishedEvent]: + """Every finished tool call, in order - for audit views and assertions.""" + return [e for e in events if isinstance(e, ToolCallFinishedEvent)] + + +__all__ = [ + "AgentEvent", + "TextChunkEvent", + "ReasoningChunkEvent", + "AssistantDoneEvent", + "PlanUpdatedEvent", + "ToolCallStartedEvent", + "ToolOutputEvent", + "ToolCallFinishedEvent", + "OutputsAddedEvent", + "OutputsRemovedEvent", + "NoticeEvent", + "HistoryReadyEvent", + "TurnCompletedEvent", + "ErrorEvent", + "EVENT_TYPES", + "event_from_dict", + "collect_text", + "tool_calls", +] diff --git a/domain/agents/conversation_execution_request.py b/domain/agents/conversation_execution_request.py new file mode 100644 index 0000000..f587633 --- /dev/null +++ b/domain/agents/conversation_execution_request.py @@ -0,0 +1,192 @@ +"""ConversationExecutionRequest - an immutable snapshot of one turn (R04-T01). + +``ui/cowork_tab.py::build_job`` currently builds a closure that reads widget +state from inside the worker thread:: + + def job(worker): + provider = self.build_provider() # reads combo boxes + extra_tools, extra_exec = self.ctx.build_mcp_tools() + proj_ctx = project_context_text(load_project(project_id)) + ... + +Everything that closure touches can change while the turn is running: the user +can pick another model, switch workspace, or edit the project instructions. The +turn then runs on a mixture of old and new state, and which mixture depends on +thread timing - the class of bug that reproduces once a week and never in a test. + +This value object is the fix: the presentation layer captures everything a turn +needs ON THE UI THREAD, at submit time, into one frozen object. Whatever happens +to the widgets afterwards, the turn keeps running on the state the user actually +submitted. + +Pure domain code: stdlib only, no Qt, no filesystem access. Paths are held as +strings, not ``Path`` objects, so the snapshot stays trivially serialisable - +which is what will let a turn be queued, replayed or logged later. +""" +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field, replace +from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple + +# Default tool-use budget for an interactive turn, and the higher ceiling a +# run-to-completion step (a Co4E flow step) is allowed. Same numbers +# ``core.chat_agent.run_cowork`` defaults to - kept here so the policy is +# visible in the request rather than buried in a function signature. +DEFAULT_MAX_STEPS = 30 +DEFAULT_COMPLETION_MAX_STEPS = 200 + + +def new_turn_id() -> str: + """A fresh turn id. Short and random: it only has to be unique within a + session's lifetime, and it shows up in log lines humans read.""" + return uuid.uuid4().hex[:12] + + +@dataclass(frozen=True) +class ConversationExecutionRequest: + """Everything one agent turn needs, captured at submit time. + + Attributes: + prompt: the user's message for this turn (already assembled, including + any attachment text the UI inlined). + messages: the full conversation to send, oldest first. Held as a tuple + so the snapshot cannot be mutated after capture; use + :meth:`message_list` to get the mutable copy the engine expects. + output_dir: this turn's OWN folder. Each turn writes into an isolated + directory so parallel turns cannot clobber each other's files. + session_id: the conversation this turn belongs to. + turn_id: unique per turn, for logs and for matching events to a turn. + surface: which screen submitted it ("cowork", "co4e", "ai_edit", "task"). + provider / model: what to run on, already resolved (routing included). + Empty ``model`` means "the provider's configured default". + title: conversation title, used to name generated files. + project_id / project_context: the workspace and its shared instructions, + snapshotted so a mid-turn workspace switch cannot change them. + agent_role: audit-log attribution for every tool call this turn makes. + allowed_tools: permission scope. ``None`` means "all enabled tools"; + a list restricts the ADVERTISED catalogue, so a read-only step + literally cannot be offered a writing tool. + max_steps / run_to_completion / completion_max_steps: tool-use budget. + enforce_rules: run the security rulebase. Co4E sandboxed runs disable it. + confirm_commands: ask before run_command/install_package (permission gate). + metadata: free-form extras a caller wants carried along (never + interpreted here) - e.g. a scheduled task's id. + """ + + prompt: str + messages: Tuple[Mapping[str, Any], ...] = () + output_dir: str = "" + session_id: str = "" + turn_id: str = field(default_factory=new_turn_id) + surface: str = "cowork" + provider: str = "" + model: str = "" + title: str = "" + project_id: str = "" + project_context: str = "" + agent_role: str = "" + allowed_tools: Optional[Tuple[str, ...]] = None + max_steps: int = DEFAULT_MAX_STEPS + run_to_completion: bool = False + completion_max_steps: int = DEFAULT_COMPLETION_MAX_STEPS + enforce_rules: bool = True + confirm_commands: bool = False + metadata: Mapping[str, Any] = field(default_factory=dict) + + # -- construction helpers ------------------------------------------- # + @classmethod + def create(cls, prompt: str, messages: Optional[Sequence[Mapping[str, Any]]] = None, + **kwargs: Any) -> "ConversationExecutionRequest": + """Build a request from ordinary mutable inputs. + + The messages list is copied element by element, so a later append by the + caller (the chat panel keeps appending to its own list) cannot reach + inside a request that is already running. + """ + snapshot = tuple(dict(m) for m in (messages or ())) + allowed = kwargs.pop("allowed_tools", None) + return cls(prompt=prompt, messages=snapshot, + allowed_tools=tuple(allowed) if allowed is not None else None, + **kwargs) + + def with_messages(self, messages: Sequence[Mapping[str, Any]] + ) -> "ConversationExecutionRequest": + """A copy carrying a different message list, everything else unchanged. + + Used when a caller assembles the system prompt or trims history after + building the request - it must produce a NEW snapshot rather than mutate + the one a turn may already be running on. + """ + return replace(self, messages=tuple(dict(m) for m in messages)) + + def with_model(self, provider: str, model: str) -> "ConversationExecutionRequest": + """A copy pinned to another provider/model - how a routing switch is + applied without touching the user's saved settings.""" + return replace(self, provider=provider, model=model) + + # -- accessors ------------------------------------------------------ # + def message_list(self) -> List[Dict[str, Any]]: + """A fresh mutable copy of the messages, for the engine to append to. + + The legacy engine mutates the list it is given (it inserts the system + prompt and appends assistant/tool messages). Handing it a copy is what + keeps this snapshot immutable in practice and not just by declaration. + """ + return [dict(m) for m in self.messages] + + @property + def effective_max_steps(self) -> int: + """The tool-use ceiling actually in force for this turn.""" + return self.completion_max_steps if self.run_to_completion else self.max_steps + + @property + def has_output_dir(self) -> bool: + """True when this turn may write files.""" + return bool(self.output_dir) + + def allows_tool(self, name: str) -> bool: + """Whether ``name`` is inside this turn's permission scope. + + ``update_plan`` is always allowed: it has no side effects and drives the + Plan panel, so scoping it out would silently break the UI rather than + restrict a capability. + """ + if self.allowed_tools is None: + return True + return name == "update_plan" or name in self.allowed_tools + + def describe(self) -> str: + """Compact one-line identity for log lines.""" + target = f"{self.provider}/{self.model}" if self.model else self.provider or "default" + return f"turn={self.turn_id} surface={self.surface} model={target}" + + def to_dict(self) -> Dict[str, Any]: + """JSON-safe projection, for logging a turn or persisting it for replay.""" + return { + "turn_id": self.turn_id, + "session_id": self.session_id, + "surface": self.surface, + "prompt": self.prompt, + "message_count": len(self.messages), + "output_dir": self.output_dir, + "provider": self.provider, + "model": self.model, + "title": self.title, + "project_id": self.project_id, + "agent_role": self.agent_role, + "allowed_tools": list(self.allowed_tools) if self.allowed_tools is not None else None, + "max_steps": self.effective_max_steps, + "run_to_completion": self.run_to_completion, + "enforce_rules": self.enforce_rules, + "confirm_commands": self.confirm_commands, + "metadata": dict(self.metadata), + } + + +__all__ = [ + "ConversationExecutionRequest", + "new_turn_id", + "DEFAULT_MAX_STEPS", + "DEFAULT_COMPLETION_MAX_STEPS", +] diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..5545e76 --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1,8 @@ +"""Integration tests: real widgets, real services, no network (R10-T01 layout). + +These build actual Qt widgets offscreen (``QT_QPA_PLATFORM=offscreen``) and run +a turn end to end with a scripted :class:`FakeProvider`. They are slower than +the unit suite - a QApplication has to exist - and are what proves the seams +introduced by R03/R04 are actually wired into the screens, not just correct in +isolation. +""" diff --git a/tests/integration/test_cowork_turn_flow.py b/tests/integration/test_cowork_turn_flow.py new file mode 100644 index 0000000..f9a11df --- /dev/null +++ b/tests/integration/test_cowork_turn_flow.py @@ -0,0 +1,201 @@ +"""End-to-end check that the Cowork screen really runs turns through the +application layer (R04-T04). + +The unit tests prove ``ConversationApplicationService`` behaves correctly; this +one proves ``ui/cowork_tab.py::build_job`` actually goes through it, on a real +(offscreen) widget, with a scripted provider instead of a network call. + +It also pins the property that motivated R04-T01: the turn runs on the state +captured at SUBMIT time, so a user editing the conversation while a turn is in +flight cannot change what that turn sends. +""" +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Dict, List + +import pytest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +from cowork_local.config import AppConfig # noqa: E402 +from cowork_local.core import chat_agent # noqa: E402 +from cowork_local.state import AppContext # noqa: E402 +from tests.fakes import FakeProvider, ScriptedTurn # noqa: E402 + +pytest.importorskip("PySide6", reason="Qt is required for the integration suite") + + +@pytest.fixture(scope="module") +def qt_app(): + """One QApplication for the module - Qt allows only a single instance.""" + from PySide6.QtWidgets import QApplication + + return QApplication.instance() or QApplication([]) + + +@pytest.fixture +def cowork_tab(qt_app, tmp_path: Path, monkeypatch): + """A real CoworkTab on a throwaway config, with ambient inputs neutralised.""" + monkeypatch.setattr(chat_agent, "active_skills_text", lambda: "") + monkeypatch.setattr(chat_agent, "load_rules", lambda: "") + from cowork_local.core import audit_log + + monkeypatch.setattr(audit_log, "AUDIT_DIR", tmp_path / "audit") + + from cowork_local.ui.cowork_tab import CoworkTab + + ctx = AppContext(AppConfig.load(tmp_path / "config.json")) + # Agent Security's prompt validation is ON by default and spends an EXTRA + # provider call reviewing the request before the agent loop starts (see + # core/agent_security.py::enforce_prompt). That is real behaviour - pinned + # by its own test below - but it would make every other test here script a + # turn that has nothing to do with what it is checking. + ctx.config.agent_security["enabled"] = False + return CoworkTab(ctx) + + +class _StubWorker: + """The slice of ``core.worker.AgentWorker`` a job actually touches.""" + + def __init__(self) -> None: + self.events: List[Dict[str, Any]] = [] + self.gates_requested = 0 + self._cancelled = False + + def emit_event(self, payload: Dict[str, Any]) -> None: + self.events.append(payload) + + def is_cancelled(self) -> bool: + return self._cancelled + + def new_gate(self, _mode: str, **_kwargs) -> Any: + self.gates_requested += 1 + return None + + def cancel(self) -> None: + self._cancelled = True + + +def _run_job(tab, worker, provider, text="hello", messages=None, out_dir=None): + """Build the tab's job with ``provider`` pinned, then run it like the worker + thread would.""" + tab.build_provider = lambda: provider # what routing/agent selection resolves to + job = tab.build_job(text, messages if messages is not None + else [{"role": "user", "content": text}], out_dir) + return job(worker) + + +def test_a_turn_runs_through_the_service_and_returns_history(cowork_tab, tmp_path): + provider = FakeProvider([ScriptedTurn(text="Hello from the fake.")]) + worker = _StubWorker() + + result = _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") + + assert provider.call_count == 1 + # Same return contract as before the refactor - _cleanup_turn reads both keys. + assert set(result) == {"messages", "turn_dir"} + assert [m["role"] for m in result["messages"]] == ["system", "user", "assistant"] + assert result["messages"][-1]["content"] == "Hello from the fake." + + +def test_the_widget_still_receives_the_legacy_event_dicts(cowork_tab, tmp_path): + """The chat widgets consume dicts and are not migrated until EPIC R08, so + the typed events must render back into exactly what they already handle - + plus the new end-of-turn signal, which the if/elif dispatch ignores.""" + provider = FakeProvider([ScriptedTurn(text="Hi")]) + worker = _StubWorker() + + _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") + + assert [e["type"] for e in worker.events] == ["text", "assistant_done", "turn_completed"] + assert worker.events[0] == {"type": "text", "delta": "Hi"} + + +def test_the_turn_ignores_messages_added_after_it_was_submitted(cowork_tab, tmp_path): + """The bug ConversationExecutionRequest exists to prevent: the panel keeps + appending to its own list while a turn is in flight.""" + provider = FakeProvider([ScriptedTurn(text="ok")]) + worker = _StubWorker() + live_messages = [{"role": "user", "content": "first question"}] + + job_result = _run_job(cowork_tab, worker, provider, + messages=live_messages, out_dir=tmp_path / "turn") + + # Simulate the user typing a second message DURING the turn by mutating the + # list the panel handed over. The already-sent conversation must not include it. + live_messages.append({"role": "user", "content": "typed while running"}) + + sent = provider.calls[0].messages + assert [m["content"] for m in sent if m["role"] == "user"] == ["first question"] + assert "typed while running" not in str(job_result["messages"]) + + +def test_a_failing_turn_still_raises_so_the_worker_reports_it(cowork_tab, tmp_path): + """core/worker.py turns an exception into the `failed` signal the chat panel + already handles; swallowing it here would show a successful turn with no + answer instead of an error.""" + provider = FakeProvider([ScriptedTurn(error="gateway down"), + ScriptedTurn(error="gateway down")]) + worker = _StubWorker() + + with pytest.raises(Exception) as excinfo: + _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") + + assert "gateway down" in str(excinfo.value) + # The error was still reported as an event before being re-raised. + assert any(e["type"] == "error" for e in worker.events) + + +def test_a_permission_gate_is_only_requested_when_the_workspace_asks_for_it( + cowork_tab, tmp_path, monkeypatch): + provider = FakeProvider([ScriptedTurn(text="ok"), ScriptedTurn(text="ok")]) + worker = _StubWorker() + + monkeypatch.setattr(cowork_tab.ctx, "project_confirm_commands", lambda: False) + _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "a") + assert worker.gates_requested == 0 + + monkeypatch.setattr(cowork_tab.ctx, "project_confirm_commands", lambda: True) + _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "b") + assert worker.gates_requested == 1 + + +def test_a_tool_turn_writes_into_this_turns_own_output_folder(cowork_tab, tmp_path): + """Turn isolation: each turn writes into its own directory so parallel turns + cannot clobber each other's files.""" + provider = FakeProvider([ + ScriptedTurn(tool_calls=[("save_file", {"filename": "n.md", "content": "x"})]), + ScriptedTurn(text="Saved."), + ]) + worker = _StubWorker() + turn_dir = tmp_path / "turn-1" + + result = _run_job(cowork_tab, worker, provider, out_dir=turn_dir) + + assert result["turn_dir"] == str(turn_dir) + assert [p.name for p in turn_dir.iterdir()] and turn_dir.exists() + assert any(e["type"] == "tool_result" and e["ok"] for e in worker.events) + + +def test_agent_security_still_reviews_the_request_before_the_turn_runs( + cowork_tab, tmp_path): + """Characterisation, not a new behaviour: with Agent Security enabled (the + shipped default) a turn costs an EXTRA provider call, because the request is + reviewed against the rulebase before the agent loop starts. + + Pinned here because it is invisible from the call site and easy to break - + routing a turn through the application layer must not skip the review. + """ + cowork_tab.ctx.config.agent_security["enabled"] = True + provider = FakeProvider([ + ScriptedTurn(text="ALLOW"), # the security pre-flight review + ScriptedTurn(text="the answer"), # the turn itself + ]) + worker = _StubWorker() + + result = _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") + + assert provider.call_count == 2 + assert result["messages"][-1]["content"] == "the answer" diff --git a/tests/unit/test_conversation_service.py b/tests/unit/test_conversation_service.py new file mode 100644 index 0000000..5847fb8 --- /dev/null +++ b/tests/unit/test_conversation_service.py @@ -0,0 +1,393 @@ +"""Unit tests for EPIC R04: the turn snapshot, the typed events and the service. + +The service tests run against the REAL engine (``core.chat_agent.run_cowork``) +driven by :class:`FakeProvider`, not against a stubbed runner. That is +deliberate: the whole point of R04 is that the service produces the same turn +the widget used to produce, and only an end-to-end path through the real engine +can show that. It still costs milliseconds - no Qt, no network, no disk beyond +a tmp folder. +""" +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, List + +import pytest + +from cowork_local.application.conversations import ConversationApplicationService +from cowork_local.core import chat_agent +from cowork_local.domain.agents import ( + AssistantDoneEvent, + ConversationExecutionRequest, + ErrorEvent, + ReasoningChunkEvent, + TextChunkEvent, + ToolCallFinishedEvent, + ToolCallStartedEvent, + TurnCompletedEvent, + collect_text, + event_from_dict, +) +from tests.fakes import FakeProvider, FakeToolExecutor, ScriptedTurn + + +# --------------------------------------------------------------------------- # +# R04-T01 - the immutable request snapshot +# --------------------------------------------------------------------------- # +def test_the_snapshot_cannot_be_changed_by_the_caller_afterwards(): + """The motivating bug: the chat panel keeps appending to its own message + list while a turn runs, and the turn must not see those later messages.""" + live_messages = [{"role": "user", "content": "first"}] + request = ConversationExecutionRequest.create("first", live_messages) + + live_messages.append({"role": "user", "content": "typed while running"}) + live_messages[0]["content"] = "edited" + + assert len(request.messages) == 1 + assert request.messages[0]["content"] == "first" + + +def test_message_list_hands_out_a_fresh_mutable_copy(): + """The engine appends assistant/tool messages to the list it is given, so a + copy is what keeps the snapshot immutable in practice, not just by + declaration.""" + request = ConversationExecutionRequest.create("hi", [{"role": "user", "content": "hi"}]) + + first = request.message_list() + first.append({"role": "assistant", "content": "reply"}) + + assert len(request.message_list()) == 1 + assert first is not request.message_list() + + +def test_with_model_produces_a_new_pinned_snapshot(): + """A routing switch must not mutate a request a turn may already be running.""" + original = ConversationExecutionRequest.create("hi", provider="openai_compat", model="a") + + routed = original.with_model("anthropic", "claude") + + assert (original.provider, original.model) == ("openai_compat", "a") + assert (routed.provider, routed.model) == ("anthropic", "claude") + assert routed.turn_id == original.turn_id # same turn, different target + + +def test_every_turn_gets_its_own_id(): + a = ConversationExecutionRequest.create("x") + b = ConversationExecutionRequest.create("x") + + assert a.turn_id and b.turn_id and a.turn_id != b.turn_id + + +def test_run_to_completion_raises_the_step_ceiling(): + interactive = ConversationExecutionRequest.create("x") + flow_step = ConversationExecutionRequest.create("x", run_to_completion=True) + + assert interactive.effective_max_steps == 30 + assert flow_step.effective_max_steps == 200 + + +def test_permission_scope_always_keeps_update_plan(): + """update_plan has no side effects and drives the Plan panel; scoping it out + would break the UI rather than restrict a capability.""" + request = ConversationExecutionRequest.create("x", allowed_tools=["read_file"]) + + assert request.allows_tool("read_file") is True + assert request.allows_tool("update_plan") is True + assert request.allows_tool("save_file") is False + # No scope at all means every enabled tool is allowed. + assert ConversationExecutionRequest.create("x").allows_tool("save_file") is True + + +# --------------------------------------------------------------------------- # +# R04-T02 - typed events and the legacy bridge +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("payload,expected", [ + ({"type": "text", "delta": "hi"}, TextChunkEvent), + ({"type": "reasoning", "delta": "hmm"}, ReasoningChunkEvent), + ({"type": "assistant_done", "content": "done"}, AssistantDoneEvent), + ({"type": "tool_proposed", "id": "1", "name": "save_file"}, ToolCallStartedEvent), + ({"type": "tool_result", "id": "1", "name": "save_file", "ok": True}, ToolCallFinishedEvent), +]) +def test_legacy_emit_dicts_map_onto_typed_events(payload, expected): + assert isinstance(event_from_dict(payload), expected) + + +def test_an_unknown_event_tag_is_dropped_rather_than_raising(): + """The engine is still being refactored and may grow an event first. Losing + one bubble is survivable; aborting a turn that had succeeded is not.""" + assert event_from_dict({"type": "something_new_in_r08"}) is None + + +@pytest.mark.parametrize("payload", [ + {"type": "text", "delta": "hi"}, + {"type": "tool_result", "id": "1", "name": "save_file", "ok": False, "output": "boom"}, + {"type": "plan_set", "steps": [{"title": "a"}]}, + {"type": "outputs_added", "paths": ["a.md"]}, +]) +def test_events_round_trip_back_into_the_legacy_shape(payload): + """Existing widgets still consume dicts; an event must render back into + exactly what they already handle (EPIC R08 migrates them).""" + event = event_from_dict(payload) + + rendered = event.to_dict() + + assert rendered["type"] == payload["type"] + for key, value in payload.items(): + assert rendered[key] == value + + +def test_events_are_immutable(): + """They cross a thread boundary; a consumer must not be able to edit one + out from under another consumer.""" + event = TextChunkEvent("hi") + + with pytest.raises(Exception): + event.delta = "changed" # type: ignore[misc] + + +def test_collect_text_returns_the_answer_without_the_reasoning(): + events = [TextChunkEvent("Hel"), ReasoningChunkEvent("secret"), TextChunkEvent("lo")] + + assert collect_text(events) == "Hello" + + +# --------------------------------------------------------------------------- # +# R04-T03 - the service, running the real engine +# --------------------------------------------------------------------------- # +@pytest.fixture +def isolated(monkeypatch, tmp_path: Path): + """Same ambient isolation the characterization suite uses.""" + monkeypatch.setattr(chat_agent, "active_skills_text", lambda: "") + monkeypatch.setattr(chat_agent, "load_rules", lambda: "") + from cowork_local.core import audit_log + + monkeypatch.setattr(audit_log, "AUDIT_DIR", tmp_path / "audit") + return tmp_path + + +def _service(provider, **kwargs) -> ConversationApplicationService: + return ConversationApplicationService(lambda _p, _m: provider, **kwargs) + + +def _request(tmp_path: Path, prompt: str = "hi", **kwargs) -> ConversationExecutionRequest: + return ConversationExecutionRequest.create( + prompt, [{"role": "user", "content": prompt}], + output_dir=str(tmp_path / "out"), **kwargs) + + +def test_a_plain_turn_reports_text_and_a_final_answer(isolated): + provider = FakeProvider([ScriptedTurn(text="Hello there.")]) + seen: List[Any] = [] + + result = _service(provider).run_turn(_request(isolated), on_event=seen.append) + + assert result.ok is True + assert result.final_text == "Hello there." + assert [e.type for e in seen] == ["text", "assistant_done", "turn_completed"] + # The conversation coming back is what the caller persists as new history. + assert [m["role"] for m in result.messages] == ["system", "user", "assistant"] + + +def test_a_turn_always_ends_with_exactly_one_completion_event(isolated): + """The end-of-turn signal the legacy engine never had: without it a + cancelled turn and a failed turn look identical to a consumer.""" + provider = FakeProvider([ScriptedTurn(text="ok")]) + seen: List[Any] = [] + + _service(provider).run_turn(_request(isolated), on_event=seen.append) + + completions = [e for e in seen if isinstance(e, TurnCompletedEvent)] + assert len(completions) == 1 + assert seen[-1] is completions[0] + + +def test_a_provider_failure_becomes_an_error_event_not_an_exception(isolated): + """Callers run this on a worker thread; an escaped exception kills the + worker and the UI simply stops updating with nothing shown. + + Two turns are scripted because the engine makes ONE silent recovery attempt + before giving up (core/code_agent.py::_call_provider_with_recovery) - the + service must report the failure only after that retry is also exhausted. + """ + provider = FakeProvider([ScriptedTurn(error="gateway exploded"), + ScriptedTurn(error="gateway exploded")]) + seen: List[Any] = [] + + result = _service(provider).run_turn(_request(isolated), on_event=seen.append) + + assert provider.call_count == 2 # original + one silent retry + assert result.ok is False + assert "gateway exploded" in result.error + assert any(isinstance(e, ErrorEvent) for e in seen) + assert isinstance(seen[-1], TurnCompletedEvent) # still a clean end + + +def test_a_transient_provider_failure_is_recovered_without_surfacing(isolated): + """The engine's single retry must stay invisible: a turn that succeeds on + the second attempt reports no error at all.""" + provider = FakeProvider([ScriptedTurn(error="connection reset"), + ScriptedTurn(text="recovered answer")]) + + result = _service(provider).run_turn(_request(isolated)) + + assert result.ok is True + assert result.final_text == "recovered answer" + assert not [e for e in result.events if isinstance(e, ErrorEvent)] + + +def test_a_cancelled_turn_is_reported_as_cancelled_not_failed(isolated): + provider = FakeProvider([], strict=True) + + result = _service(provider).run_turn(_request(isolated), cancel=lambda: True) + + assert result.cancelled is True + assert result.error == "" + assert provider.call_count == 0 + assert result.events[-1].cancelled is True + + +def test_a_tool_turn_reports_the_full_lifecycle_and_writes_the_file(isolated): + provider = FakeProvider([ + ScriptedTurn(tool_calls=[("save_file", {"filename": "note.md", "content": "# hi"})]), + ScriptedTurn(text="Saved."), + ]) + + result = _service(provider).run_turn(_request(isolated, "make a note")) + + assert [e.type for e in result.events] == [ + "assistant_done", "tool_proposed", "tool_result", + "text", "assistant_done", "turn_completed", + ] + finished = [e for e in result.events if isinstance(e, ToolCallFinishedEvent)] + assert finished[0].ok is True and finished[0].name == "save_file" + written = list((isolated / "out").iterdir()) + assert len(written) == 1 and written[0].read_text(encoding="utf-8") == "# hi" + + +def test_external_tools_are_supplied_through_the_injected_tool_source(isolated): + executor = FakeToolExecutor(results={"ms365_send_mail": {"output": "sent"}}) + provider = FakeProvider([ + ScriptedTurn(tool_calls=[("ms365_send_mail", {"to": "a@b.c"})]), + ScriptedTurn(text="Mail sent."), + ]) + service = _service(provider, tool_source=lambda: (executor.specs(), executor)) + + result = service.run_turn(_request(isolated, "mail them")) + + assert executor.call_names == ["ms365_send_mail"] + assert result.ok is True + + +def test_a_broken_tool_source_degrades_to_no_external_tools(isolated): + """An MCP server that will not start must not stop the user from chatting - + the behaviour the chat panel already relies on today.""" + def exploding_tool_source(): + raise RuntimeError("mcp server did not start") + + provider = FakeProvider([ScriptedTurn(text="still works")]) + service = _service(provider, tool_source=exploding_tool_source) + + result = service.run_turn(_request(isolated)) + + assert result.ok is True + assert result.final_text == "still works" + + +def test_a_consumer_that_raises_does_not_abort_the_turn(isolated): + """A widget being torn down mid-turn must not take the turn with it.""" + provider = FakeProvider([ScriptedTurn(text="answer")]) + + def bad_consumer(_event): + raise RuntimeError("widget already deleted") + + result = _service(provider).run_turn(_request(isolated), on_event=bad_consumer) + + assert result.ok is True + assert result.final_text == "answer" + + +def test_events_are_recorded_even_without_a_callback(isolated): + """Headless callers (the scheduler) read the event list afterwards instead + of supplying a callback purely to collect it.""" + provider = FakeProvider([ScriptedTurn(text="ok")]) + + result = _service(provider).run_turn(_request(isolated)) + + assert [e.type for e in result.events] == ["text", "assistant_done", "turn_completed"] + + +def test_the_request_permission_scope_reaches_the_engine(isolated): + """A read-only step must literally not be offered a writing tool - the scope + has to survive the trip through the service or the restriction is silently + dropped.""" + provider = FakeProvider([ScriptedTurn(text="ok")]) + + _service(provider).run_turn(_request(isolated, allowed_tools=["read_file"])) + + advertised = set(provider.calls[0].tool_names) + assert "save_file" not in advertised + assert "update_plan" in advertised + + +def test_the_permission_gate_is_only_built_when_the_request_asks_for_it(isolated): + built: List[Any] = [] + provider = FakeProvider([ScriptedTurn(text="ok"), ScriptedTurn(text="ok")]) + service = _service(provider, gate_factory=lambda req: built.append(req) or object()) + + service.run_turn(_request(isolated)) + assert built == [] + + service.run_turn(_request(isolated, confirm_commands=True)) + assert len(built) == 1 + + +def test_a_non_streamed_answer_still_produces_a_final_text(isolated): + """A turn whose answer arrived without text events must still report an + answer - the scheduler writes it into output.md, and an empty string there + reads to the user as "(no output)".""" + provider = FakeProvider([ScriptedTurn(text="")]) + service = _service(provider) + request = _request(isolated) + + result = service.run_turn(request) + + # run_cowork substitutes a placeholder for a reasoning-only reply; the + # service must surface that rather than an empty answer. + assert result.final_text != "" + + +# --------------------------------------------------------------------------- # +# Bridge completeness - the failure mode that motivated this test +# --------------------------------------------------------------------------- # +def test_every_event_the_engine_emits_has_a_typed_counterpart(): + """Scan the engine sources for ``emit({"type": "..."})`` tags and assert the + bridge knows all of them. + + Written after a real miss: the first version of the bridge had no + ``notice`` event, so routing turns through the service would have silently + swallowed Agent Security warnings and auto-compaction notices - the user + would simply never see that a request had been blocked. An unknown tag is + dropped by design (see event_from_dict), which is safe for a NEW event but + hides a forgotten one; this test is what turns that silence into a failure. + """ + import re + from pathlib import Path + + from cowork_local.domain.agents.agent_event import EVENT_TYPES + + repo = Path(__file__).resolve().parents[2] + sources = ["core/chat_agent.py", "core/code_agent.py", "core/agent_security.py", + "core/context_budget.py", "core/task_executors.py"] + emitted = set() + for rel in sources: + text = (repo / rel).read_text(encoding="utf-8") + # Only tags inside an emit(...) call; a bare {"type": "object"} in a + # JSON-Schema tool definition is not an event. + for match in re.finditer(r'emit(?:_and_autosave)?\(\s*\{\s*"type":\s*"([a-z_]+)"', text): + emitted.add(match.group(1)) + + missing = sorted(emitted - set(EVENT_TYPES)) + assert not missing, ( + f"engine emits {missing} but domain/agents/agent_event.py has no typed " + "counterpart - those events would be silently dropped by event_from_dict" + ) diff --git a/ui/cowork_tab.py b/ui/cowork_tab.py index 44c620f..a122922 100644 --- a/ui/cowork_tab.py +++ b/ui/cowork_tab.py @@ -346,45 +346,82 @@ class CoworkTab(ChatPanel): self._apply_output_folder_label() # picks up edits made via Settings too def build_job(self, text: str, messages, out_dir): - # Each turn writes into its OWN isolated folder (out_dir) and works on its - # OWN message list, so several turns can run in parallel without clobbering - # each other's files or history. Deliverables are moved up to the session - # Output root when the turn finishes (see _cleanup_turn). + """Build the worker job for one Cowork turn (R04-T04). + + Every input the turn needs is captured HERE, on the UI thread, into an + immutable ``ConversationExecutionRequest``. Previously the job closure + read widget state (selected model, active workspace, project + instructions) from inside the worker thread, so a turn could run on a + mixture of the state at submit time and the state the user changed while + it was running - and which mixture you got depended on thread timing. + + Each turn still writes into its OWN isolated folder (out_dir) and works + on its OWN message list, so several turns can run in parallel without + clobbering each other's files or history. Deliverables are moved up to + the session Output root when the turn finishes (see _cleanup_turn). + """ + from ..core.projects import load_project, project_context_text + from ..domain.agents import ConversationExecutionRequest + output_dir = out_dir or self._session_output_dir() - title = self.title - project_id = self.project_id - # Captured at submit time (UI thread): the Admin-defined agent - # preset's instructions, if one is selected in the Agent picker. + # Shared project instructions (Claude-Projects style), read now so edits + # made in the Workspace screen mid-turn cannot change this turn's prompt. + project_context = project_context_text(load_project(self.project_id)) + # The Admin-defined agent preset's instructions, if one is selected. agent_prompt = self.admin_agent_prompt() + if agent_prompt: + project_context = (f"{project_context}\n\n{agent_prompt}" + if project_context else agent_prompt) + + request = ConversationExecutionRequest.create( + text, messages, + output_dir=str(output_dir), + session_id=self.session_id, + surface=self.kind, + title=self.title, + project_id=self.project_id, + project_context=project_context, + agent_role=agent_roles.COWORK, + # Permission Management (Sandbox Security Layer): off by default - + # matches the pre-existing auto-run behavior. Resolved PER WORKSPACE: + # this project's Auto-run override wins, else the global setting. + confirm_commands=bool(self.ctx.project_confirm_commands()), + ) + # Built on the UI thread with everything else: it already reflects this + # turn's routing decision and the selected agent/model. + provider = self.build_provider() def job(worker: AgentWorker): - from ..core.chat_agent import run_cowork - from ..core.projects import load_project, project_context_text + from ..application.conversations import ConversationApplicationService - provider = self.build_provider() # this tab's selected agent/model - # 🔌 MCP Layer: every tool source flows through MCP now — the - # external servers configured in Settings AND Microsoft 365 (a - # built-in MCP server auto-registered while signed in, see - # AppContext._ms365_builtin_connection / mcp_servers/ms365_server.py). - extra_tools, extra_exec = self.ctx.build_mcp_tools() - # Shared project instructions (Claude-Projects style) — refreshed - # each turn so edits in the Workspace screen apply immediately. - proj_ctx = project_context_text(load_project(project_id)) - if agent_prompt: - proj_ctx = f"{proj_ctx}\n\n{agent_prompt}" if proj_ctx else agent_prompt - # Permission Management (Sandbox Security Layer): off by default — - # matches the pre-existing auto-run behavior. Now resolved PER - # WORKSPACE: this project's Auto-run override wins, else the global - # "confirm before running commands" setting (project_confirm_commands). - gate = None - if self.ctx.project_confirm_commands(): - gate = worker.new_gate("confirm", agent_role=agent_roles.COWORK) - run_cowork(provider, messages, output_dir, worker.emit_event, - worker.is_cancelled, title=title, - extra_tools=extra_tools, extra_executor=extra_exec, - project_context=proj_ctx, security_config=self.ctx.config, - gate=gate) - return {"messages": messages, "turn_dir": str(output_dir)} + service = ConversationApplicationService( + # The provider is part of the snapshot, so the factory ignores + # the request's provider/model rather than re-resolving them + # from live config inside the worker thread. + lambda _provider_id, _model: provider, + # 🔌 MCP Layer: every tool source flows through MCP now - the + # external servers configured in Settings AND Microsoft 365 (a + # built-in MCP server auto-registered while signed in, see + # AppContext._ms365_builtin_connection / mcp_servers/ms365_server.py). + tool_source=self.ctx.build_mcp_tools, + gate_factory=lambda req: worker.new_gate("confirm", + agent_role=agent_roles.COWORK), + security_config=self.ctx.config, + ) + result = service.run_turn( + request, + # The typed events are rendered back into the legacy dict shape + # the chat widgets already consume; they migrate to AgentEvent + # directly in EPIC R08. + on_event=lambda event: worker.emit_event(event.to_dict()), + cancel=worker.is_cancelled, + ) + # The service reports a failure instead of raising, but this worker's + # contract is exception-based (core/worker.py turns one into the + # `failed` signal that _on_failed already handles), so re-raise the + # ORIGINAL exception to keep that path byte-for-byte unchanged. + result.raise_if_failed() + return {"messages": result.messages, "turn_dir": str(output_dir)} return job