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