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