From 3665135c38e3bce48f45377f879ec7c1cd0fe0e7 Mon Sep 17 00:00:00 2001 From: Duy Le Huu Date: Sun, 23 Aug 2026 13:14:15 +0900 Subject: [PATCH] feat(R04): run every Cowork turn through ConversationApplicationService MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit R04-T03 — the turn lifecycle, extracted from `core/chat_agent.py::run_cowork` into `application/conversations/`. The 260-line body mixed the lifecycle (step budget, cancel checks, guard -> preview -> gate -> execute ordering, sandbox tidy-up) with the machinery doing each step, and reaching any of it meant standing up a Qt widget and a worker thread. It is now a plain object driven through two Protocols and six callables (`turn_runtime.py`), with the concrete `core/*` wiring confined to `core_runtime_adapter.py` — the same shape R03 used for routing. Faithful port, not an improvement pass: where the original had a quirk (the step-ceiling note only merges into the answer when the last message is the assistant's) the quirk is preserved and commented. R04-T04 — `ui/cowork_tab.py::build_job` no longer calls run_cowork. It captures the widget's state at submit time, builds the request via the new `cowork_turn_request.py` and executes it. `execute(..., messages=...)` hands the widget's own list over because `_reattach_running_turn` replays from it WHILE the worker appends and `_finalize_turn` slices it afterwards — a private list would break both silently. R04-T05 — `core/task_executors.py`'s cowork branch shares the same engine. All five unattended-run behaviours stay put (plan reminder, history_ready, History autosave per assistant message, timeout notice, plan_incomplete_reason), and `_unattended_prompt` now expresses the load-bearing prefix order in one readable call instead of three successive rebindings. Verification: 74 new tests (364 passed, 1 skipped overall; check_imports PASS). The two that matter most: - `test_conversation_service_parity.py` runs the same scripted turn through run_cowork AND the service and compares the event stream, the resulting conversation and the advertised tool list across 7 scenarios; - `test_task_executor_turn.py` was written BEFORE the migration and passed 8/8 against the old code, then unchanged against the new. Known: `ui/cowork_tab.py` (416 -> 455) and `core/task_executors.py` (476 -> 524) stay above the 400-LOC limit. Both were already over it before this change; bringing them under needs the R08 / R07 decompositions. Co-Authored-By: Claude Opus 5 --- .../conversation_application_service.py | 322 +++++++++++++++++ .../conversations/core_runtime_adapter.py | 325 ++++++++++++++++++ .../conversations/cowork_turn_request.py | 77 +++++ application/conversations/turn_runtime.py | 177 ++++++++++ core/task_executors.py | 88 ++++- docs/refactor/Refactoring_Checklist.md | 26 +- tests/fakes/turn_runtime_fakes.py | 141 ++++++++ .../test_conversation_service_parity.py | 207 +++++++++++ tests/integration/test_cowork_tab_turn.py | 220 ++++++++++++ tests/integration/test_task_executor_turn.py | 191 ++++++++++ .../test_conversation_application_service.py | 293 ++++++++++++++++ tests/unit/test_conversation_turn_guards.py | 210 +++++++++++ tests/unit/test_cowork_turn_request.py | 76 ++++ tests/unit/test_task_prompt_assembly.py | 34 ++ tests/unit/test_turn_runtime.py | 36 ++ ui/cowork_tab.py | 77 ++++- 16 files changed, 2452 insertions(+), 48 deletions(-) create mode 100644 application/conversations/conversation_application_service.py create mode 100644 application/conversations/core_runtime_adapter.py create mode 100644 application/conversations/cowork_turn_request.py create mode 100644 application/conversations/turn_runtime.py create mode 100644 tests/fakes/turn_runtime_fakes.py create mode 100644 tests/integration/test_conversation_service_parity.py create mode 100644 tests/integration/test_cowork_tab_turn.py create mode 100644 tests/integration/test_task_executor_turn.py create mode 100644 tests/unit/test_conversation_application_service.py create mode 100644 tests/unit/test_conversation_turn_guards.py create mode 100644 tests/unit/test_cowork_turn_request.py create mode 100644 tests/unit/test_task_prompt_assembly.py create mode 100644 tests/unit/test_turn_runtime.py diff --git a/application/conversations/conversation_application_service.py b/application/conversations/conversation_application_service.py new file mode 100644 index 0000000..fdc9046 --- /dev/null +++ b/application/conversations/conversation_application_service.py @@ -0,0 +1,322 @@ +"""The turn lifecycle, once, in pure Python (R04-T03). + +Extracted from ``core/chat_agent.py::run_cowork``, whose 260-line body mixed the +lifecycle (compose the prompt, call the model, dispatch tools, respect the step +ceiling, tidy the sandbox) with the concrete machinery that does each of those +things. The lifecycle is the part with rules worth testing — and the part that +was untestable, because reaching it meant standing up a Qt widget and a worker +thread. + +Here it is a plain object driven through the seams in :mod:`turn_runtime`, so a +test states a rule ("the guard runs before the model", "a rejected command never +executes") in three lines. ``core/chat_agent.py`` keeps its signature and +delegates, and the presentation layer keeps receiving the same events via the +legacy codec, so nothing downstream had to change with it. + +Behavioural contract: this is a faithful port, not an improvement pass. Where +the original had a quirk (the step-ceiling note only merges into the answer when +the last message is the assistant's), the quirk is preserved and commented — +changing what a user sees belongs in its own change, not smuggled into a move. +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional, Tuple + +from ...domain.agents.agent_event import ( + AssistantMessageCompletedEvent, + ErrorEvent, + OutputsAddedEvent, + OutputsRemovedEvent, + PlanStep, + PlanUpdatedEvent, + ReasoningChunkEvent, + TextChunkEvent, + ToolCallFinishedEvent, + ToolCallStartedEvent, + ToolOutputChunkEvent, +) +from ...domain.agents.agent_result import AgentResult +from ...domain.agents.conversation_execution_request import ConversationExecutionRequest +from .turn_runtime import ( + BUDGET_NOTE_TEMPLATE, + GATED_TOOLS, + PLAN_TOOL, + REASONING_ONLY_NOTE, + REJECTED_OUTPUT, + AttachmentReader, + CancelFn, + CommandGuard, + ContextCompactor, + EventSink, + ModelCallPort, + PermissionRequest, + PromptGuard, + PromptPreparer, + ToolRuntimePort, +) + +logger = logging.getLogger("cowork_local.application.conversations") + + +class ConversationApplicationService: + """Runs one :class:`ConversationExecutionRequest` to completion.""" + + def __init__( + self, + model: ModelCallPort, + tools: ToolRuntimePort, + *, + prepare_prompt: Optional[PromptPreparer] = None, + prompt_guard: Optional[PromptGuard] = None, + command_guard: Optional[CommandGuard] = None, + compact: Optional[ContextCompactor] = None, + permission_request: Optional[PermissionRequest] = None, + attachment_reader: Optional[AttachmentReader] = None, + ) -> None: + self._model = model + self._tools = tools + # Every hook is optional so the service degrades to a plain chat turn. + # That is not only a test convenience: a headless caller legitimately has + # no guards (``security_config=None`` today) and no permission dialog. + self._prepare_prompt = prepare_prompt + self._prompt_guard = prompt_guard + self._command_guard = command_guard + self._compact = compact + self._permission_request = permission_request + self._attachment_reader = attachment_reader + + # -- public API ------------------------------------------------------ # + def execute(self, request: ConversationExecutionRequest, sink: EventSink, + cancel: Optional[CancelFn] = None, + messages: Optional[List[Dict[str, Any]]] = None) -> AgentResult: + """Run the turn, streaming events to ``sink``, and report the outcome. + + ``messages``, when given, is a working list the caller already built — + it MUST already end with this turn's user message, and the service + appends into that very object instead of composing its own. The Cowork + widget needs this: it hands out the same list to + ``_reattach_running_turn``, which replays the steps done so far while the + worker is still appending, and to ``_finalize_turn``, which slices it by + the pre-turn snapshot length. A private list would break both silently. + Passing ``None`` (every headless caller) lets the service compose the + list from the request, which is the mode the rest of this class assumes. + + Raises whatever the runtime raises (a blocked prompt, a dead gateway): + the caller already has a failure path for that — ``AgentWorker.failed`` + in the UI, the artifact writer in Schedule Task — and swallowing the + exception here would silently turn a failed turn into an empty answer. + An :class:`ErrorEvent` is emitted first so subscribers see the failure + on the same stream as everything else. + """ + cancel = cancel or (lambda: False) + + # -- pre-flight. Runs BEFORE the output snapshot, so a turn refused here + # leaves the output folder completely untouched (tidying is not a + # read-only operation — see ToolRuntimePort.finalize). + try: + # The caller's list is used by reference on purpose (see above); only + # the self-composed path may build a fresh one. + working = messages if messages is not None else self._compose_messages(request) + tools = list(self._tools.specs(request.allowed_tools)) + if self._prepare_prompt is not None: + self._prepare_prompt(working, tuple(getattr(t, "name", "") for t in tools)) + if request.enforce_rules and self._prompt_guard is not None: + self._prompt_guard(working) + except Exception as exc: # noqa: BLE001 — reported, then re-raised as-is + sink(ErrorEvent(message=str(exc))) + raise + + before = self._tools.snapshot() + steps_used = 0 + plan_steps: Tuple[PlanStep, ...] = () + completed_naturally = False + try: + for _ in range(request.effective_max_steps): + if cancel(): + break + # Auto-compress when nearing the model's context budget; a no-op + # when off or when the conversation is still short. + if self._compact is not None: + self._compact(working, cancel) + + assistant = self._model.call( + working, tools, + on_text=lambda piece: sink(TextChunkEvent(delta=piece)), + on_reasoning=lambda piece: sink(ReasoningChunkEvent(delta=piece)), + cancel=cancel, + ) + working.append(assistant) + steps_used += 1 + tool_calls = assistant.get("tool_calls") or [] + + if not tool_calls and not (assistant.get("content") or "").strip(): + # Written into the message, not just emitted, so the stored + # conversation never ends on a blank assistant turn. + assistant["content"] = REASONING_ONLY_NOTE + sink(TextChunkEvent(delta=REASONING_ONLY_NOTE)) + sink(AssistantMessageCompletedEvent(content=assistant.get("content", ""))) + + if not tool_calls: + completed_naturally = True + break + + for call in tool_calls: + if cancel(): + break + tool_message, steps = self._dispatch(request, call, sink, cancel) + working.append(tool_message) + if steps is not None: + plan_steps = steps + + if not completed_naturally and not cancel(): + self._announce_budget_exhausted(request, working, sink) + except Exception as exc: # noqa: BLE001 — reported, then re-raised as-is + sink(ErrorEvent(message=str(exc))) + raise + finally: + # Always tidy: the sandbox and generator scripts must not survive a + # turn that stopped abruptly. Runs on success, cancel and failure. + self._finalize_outputs(before, sink, cancelled=cancel()) + + result = AgentResult( + messages=working, steps_used=steps_used, cancelled=cancel(), + budget_exhausted=not completed_naturally and not cancel(), + plan_steps=plan_steps, + ) + sink(result.to_turn_completed_event()) + return result + + # -- internals ------------------------------------------------------- # + def _compose_messages(self, request: ConversationExecutionRequest) -> List[Dict[str, Any]]: + """History snapshot plus this turn's user message. + + The attachment text is read HERE rather than when the request was built, + because extraction is slow enough to freeze the UI thread; the request + deliberately carries paths only. + """ + body = request.prompt + if self._attachment_reader is not None: + body = self._attachment_reader(request.prompt, request.attachments) + messages = [dict(m) for m in request.messages] + messages.append({"role": "user", "content": request.user_content(body)}) + return messages + + def _dispatch(self, request: ConversationExecutionRequest, call: Dict[str, Any], + sink: EventSink, cancel: CancelFn + ) -> Tuple[Dict[str, Any], Optional[Tuple[PlanStep, ...]]]: + """Run one tool call. + + Returns ``(tool_message, plan_steps)`` — the message to append to the + conversation, and the new checklist when this call was the plan tool + (``None`` otherwise, so the caller can tell "no change" from "empty + plan"). + """ + call_id = str(call.get("id", "")) + name = str(call.get("name", "")) + args = call.get("arguments") or {} + + # The plan tool is invisible in the transcript: it updates the Plan panel + # and nothing else, so it skips preview, guard and gate entirely. + if name == PLAN_TOOL: + outcome = self._tools.execute(name, args, on_output=None, cancel=cancel) + steps = tuple(outcome.get("plan_steps") or ()) + sink(PlanUpdatedEvent(steps=steps)) + return self._tool_message(call_id, name, outcome.get("output", "")), steps + + # Announce first: the user sees the code/command about to run before the + # guard or the approval dialog interrupts them, which is the whole point + # of showing the step CLI-style. + preview = self._tools.preview(name, args) + sink(ToolCallStartedEvent(call_id=call_id, name=name, arguments=dict(args), + preview=preview)) + + if request.enforce_rules and self._command_guard is not None: + self._command_guard(name, args) + + if not self._approved(request, name, args, preview, sink, call_id): + return self._tool_message(call_id, name, REJECTED_OUTPUT), None + + outcome = self._tools.execute( + name, args, + on_output=lambda piece: sink(ToolOutputChunkEvent( + call_id=call_id, name=name, delta=piece)), + cancel=cancel, + ) + sink(ToolCallFinishedEvent( + call_id=call_id, name=name, ok=bool(outcome.get("ok", False)), + output=str(outcome.get("output", "")), path=str(outcome.get("path", "") or ""), + produced=outcome.get("produced") or (), + )) + return self._tool_message(call_id, name, outcome.get("output", "")), None + + def _approved(self, request: ConversationExecutionRequest, name: str, + args: Dict[str, Any], preview: Any, sink: EventSink, + call_id: str) -> bool: + """Whether this call may run. + + Only command-shaped tools are gated, and only when the workspace asked + to confirm them: file writes stay inside the turn's own sandbox, so + prompting for those would be noise. A rejection is reported as a failed + tool result — the model needs to read back that it was refused, or it + will simply try the same call again. + """ + if not request.requires_permission_gate or name not in GATED_TOOLS: + return True + if self._permission_request is None: + # Confirm mode with nobody to ask: refusing is the safe direction, + # since auto-running is exactly what confirm mode exists to prevent. + logger.warning("turn: confirm mode without a permission callback — refusing %r", name) + approved = False + else: + approved = bool(self._permission_request({ + "name": name, "args": args, + "preview": preview.to_dict() if preview is not None else {}, + })) + if not approved: + sink(ToolCallFinishedEvent(call_id=call_id, name=name, ok=False, + output=REJECTED_OUTPUT)) + return approved + + @staticmethod + def _tool_message(call_id: str, name: str, output: Any) -> Dict[str, Any]: + """The canonical ``role: tool`` message the model reads back.""" + return {"role": "tool", "tool_call_id": call_id, "name": name, + "content": str(output or "")} + + @staticmethod + def _announce_budget_exhausted(request: ConversationExecutionRequest, + messages: List[Dict[str, Any]], sink: EventSink) -> None: + """Report being cut off by the step ceiling. + + The note always reaches the transcript. It is merged into the stored + answer only when the last message is the assistant's — which, when the + ceiling is hit, it never is (the turn ends on a tool result). The branch + is kept because it is what the current runtime does, and because it is + the correct behaviour the day a caller ends the loop differently. + """ + note = BUDGET_NOTE_TEMPLATE.format(steps=request.effective_max_steps) + sink(TextChunkEvent(delta=note)) + if messages and messages[-1].get("role") == "assistant": + messages[-1]["content"] = (messages[-1].get("content") or "") + note + + def _finalize_outputs(self, before: Any, sink: EventSink, cancelled: bool) -> None: + """Tidy the output folder and report what moved. + + Failures are logged, never raised: this runs in a ``finally``, so an + exception here would replace the turn's real error (or its success) with + a housekeeping one. + """ + try: + removed, added = self._tools.finalize(before, cancelled=cancelled) + except Exception: # noqa: BLE001 + logger.exception("turn: tidying the output folder failed") + return + if removed: + sink(OutputsRemovedEvent(paths=tuple(removed))) + if added: + sink(OutputsAddedEvent(paths=tuple(added))) + + +__all__ = ["ConversationApplicationService"] diff --git a/application/conversations/core_runtime_adapter.py b/application/conversations/core_runtime_adapter.py new file mode 100644 index 0000000..aabf714 --- /dev/null +++ b/application/conversations/core_runtime_adapter.py @@ -0,0 +1,325 @@ +"""Wires :class:`ConversationApplicationService` to the existing runtime (R04-T03). + +The service is written against the narrow seams in :mod:`turn_runtime` so it can +be tested with plain fakes. This module supplies the real implementations — the +provider call with its recovery pass, the tool/sandbox runtime, the security +guards, context compaction — and is therefore the ONLY file in +``application/conversations/`` that knows ``core/*`` exists. Same shape (and +same reason) as ``application/model_routing/core_routing_adapter.py`` in R03. + +Every ``core`` import is deferred into a method body: importing the tool runtime +pulls in ``requests``, ``psutil`` and the sandbox stack, and code that merely +*builds* a service must not pay for that. + +Faithfulness notes — two places where this reproduces a quirk of the current +runtime rather than the behaviour one would design fresh. Both are marked +inline: the MS365 system-prompt paragraph keys off the CONFIGURED extra tools +(not the advertised subset), and the ``tool_result`` path falls back to the +call's own ``path`` argument resolved against the workdir. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple + +from ...domain.agents.agent_event import PlanStep, ToolPreview +from .conversation_application_service import ConversationApplicationService +from .turn_runtime import PLAN_TOOL, EventSink + +# Legacy emit: the dict-based callback every current caller already owns. +LegacyEmit = Callable[[Dict[str, Any]], None] + + +def legacy_event_sink(emit: LegacyEmit) -> EventSink: + """Adapt a typed :class:`EventSink` onto the legacy dict ``emit``. + + This is what lets R04 land without touching the presentation layer: the + service thinks in typed events, ``ui/chat_panel.py::_on_event`` keeps + receiving exactly the dicts it already dispatches on. Deleted in R08 once + the widget consumes events directly. + """ + return lambda event: emit(event.to_legacy_dict()) + + +class CoreModelCall: + """:class:`ModelCallPort` over ``code_agent._call_provider_with_recovery``. + + Not ``provider.chat`` directly: the recovery wrapper adds the one bounded + retry that hides a dropped connection or a momentarily unreachable gateway, + and losing it would be a visible regression on flaky corporate networks. + """ + + def __init__(self, provider: Any) -> None: + self._provider = provider + + def call(self, messages, tools, on_text=None, on_reasoning=None, cancel=None): + from ...core.code_agent import _call_provider_with_recovery + + return _call_provider_with_recovery(self._provider, messages, tools, on_text, + cancel, on_reasoning) + + +class CoreToolRuntime: + """:class:`ToolRuntimePort` over ``core/tools.py`` + Cowork's file tools.""" + + def __init__(self, output_dir: Path, *, title: str = "", + extra_tools: Optional[Sequence[Any]] = None, extra_executor=None, + security_config: Any = None, agent_role: str = "") -> None: + self._output_dir = Path(output_dir) + self._title = title + self._extra_tools = list(extra_tools or ()) + self._extra_names = {getattr(t, "name", "") for t in self._extra_tools} + # The connector executor MCP/REST tools are routed to; None when the + # turn has no connectors enabled. + self._extra_executor = extra_executor + self._security_config = security_config + self._agent_role = agent_role + self._ctx: Any = None # built on first use (see _tool_context) + + # -- the configured extra tools, for the system-prompt hints ---------- # + @property + def extra_names(self) -> frozenset: + return frozenset(self._extra_names) + + def _tool_context(self): + """The sandboxed ``ToolContext`` every built-in tool call runs inside. + + Built once per turn and cached: it carries the resource limits and the + network policy, so re-deriving it mid-turn could let a Settings change + take effect halfway through work already in flight. + """ + if self._ctx is None: + from ...core import agent_security + from ...core.tools import ToolContext + + limits, block_network = agent_security.sandbox_settings(self._security_config) + self._ctx = ToolContext( + self._output_dir, flatten_writes=True, # keep every file in the Output root + resource_limits=limits, block_network=block_network, + allow_url_fetch=agent_security.url_fetch_allowed(self._security_config), + jira=(self._security_config.data.get("jira") if self._security_config else None), + ) + return self._ctx + + # -- ToolRuntimePort -------------------------------------------------- # + def specs(self, allowed_tools: Optional[Sequence[str]] = None) -> List[Any]: + """Advertised tools: Cowork's own two, the enabled built-ins, then MCP. + + ``allowed_tools`` restricts the list so a read-only step literally cannot + write. ``update_plan`` and the connector tools always survive the filter: + the plan tool has no side effects, and connectors are opted into + explicitly rather than governed by the built-in capability scope. + """ + from ...core.chat_agent import SAVE_FILE_SPEC + from ...core.plan import UPDATE_PLAN_SPEC + from ...core.tools import enabled_tool_specs + + specs = ([SAVE_FILE_SPEC, UPDATE_PLAN_SPEC] + + list(enabled_tool_specs(self._security_config)) + + self._extra_tools) + if allowed_tools is None: + return specs + allow = set(allowed_tools) | {PLAN_TOOL} | self._extra_names + return [t for t in specs if getattr(t, "name", "") in allow] + + def preview(self, name: str, args: Dict[str, Any]) -> Optional[ToolPreview]: + """What the user sees before the call runs.""" + # A connector call has no local diff to show, so it renders as the plain + # argument dump the runtime already used. + if name in self._extra_names: + return ToolPreview(kind="info", title=name, text=str(args)) + if name == "save_file": + return self._save_file_preview(args) + from ...core.tools import describe_action + + raw = describe_action(self._tool_context(), name, args) + return ToolPreview.from_dict(raw) + + def _save_file_preview(self, args: Dict[str, Any]) -> ToolPreview: + """A before/after diff for the file the agent is about to write. + + A brand-new file renders all-green (before is empty); an overwrite shows + the real change, so saving a file reads like editing one. + """ + import difflib + + from ...core.chat_agent import _structure_summary, _titled_filename + + fname = _titled_filename(self._title, args.get("filename", "output.txt")) + content = str(args.get("content", "")) + summary = _structure_summary(fname, content) + old = "" + existing = self._output_dir / fname + if existing.exists(): + try: + old = existing.read_text(encoding="utf-8", errors="replace") + except OSError: + pass # unreadable existing file: show it as a fresh write + diff = "".join(difflib.unified_diff( + old.splitlines(keepends=True), content.splitlines(keepends=True), + fromfile=f"a/{fname}", tofile=f"b/{fname}", + )) or content[:4000] + return ToolPreview(kind="diff", title=f"Save {fname}", + text=f"{summary}\n\n{diff[:4000]}") + + def execute(self, name: str, args: Dict[str, Any], on_output=None, + cancel=None) -> Dict[str, Any]: + """Run one tool call and return the runtime's result mapping.""" + if name == PLAN_TOOL: + return self._execute_plan(args) + if name in self._extra_names and self._extra_executor is not None: + # Connector results carry no local file, so no path/produced keys — + # matching what the runtime reports for an MCP call today. + result = self._extra_executor(name, args) or {} + return {"ok": bool(result.get("ok", False)), "output": result.get("output", "")} + if name == "save_file": + from ...core.chat_agent import _do_save_file + + return dict(_do_save_file(self._output_dir, self._title, args)) + + from ...core.tools import execute_tool + + ctx = self._tool_context() + result = dict(execute_tool(ctx, name, args, cancel=cancel, on_output=on_output, + agent_role=self._agent_role)) + # Quirk preserved: a tool that wrote the file named in its OWN arguments + # (write_file/edit_file) does not report a path, so the runtime derives + # one from the argument. Dropping this would empty the Output list. + if not result.get("path") and isinstance(args, dict) and args.get("path"): + result["path"] = str(ctx.workdir / str(args["path"])) + return result + + def _execute_plan(self, args: Dict[str, Any]) -> Dict[str, Any]: + """Apply an ``update_plan`` call: validate the steps and audit them. + + Produces no file and no chat bubble; the service turns the returned + steps into a single plan event. + """ + from ...core import agent_roles, audit_log + from ...core.plan import normalize_plan_steps + + steps = normalize_plan_steps(args.get("steps")) + audit_log.record("tool_call", PLAN_TOOL, True, f"{len(steps)} step(s)", + agent_role=agent_roles.PLANNER) + return {"ok": True, "output": "Plan updated.", + "plan_steps": [PlanStep(title=s["title"], status=s["status"]) for s in steps]} + + def snapshot(self) -> Any: + from ...core.tools import _snapshot + + return _snapshot(self._output_dir) + + def finalize(self, before: Any, cancelled: bool = False + ) -> Tuple[List[str], List[str]]: + """Drop the scratch sandbox and flatten deliverables into the root. + + Returns ``(gone, arrived)``: a file that MOVED counts as both, because + the Output list keys entries by path and must drop the old one. + """ + from ...core.chat_agent import _cleanup_cowork_intermediates + + removed, moved = _cleanup_cowork_intermediates(self._output_dir, before, + cancelled=cancelled) + gone = list(removed) + [old for old, _new in moved] + arrived = [new for _old, new in moved] + return gone, arrived + + +def build_cowork_conversation_service( + provider: Any, + output_dir: Path, + emit: LegacyEmit, + *, + title: str = "", + project_context: str = "", + extra_tools: Optional[Sequence[Any]] = None, + extra_executor=None, + security_config: Any = None, + gate: Any = None, + agent_role: str = "", +) -> ConversationApplicationService: + """A service wired to the real runtime, ready to execute a Cowork turn. + + ``emit`` is the legacy dict callback: the guards and the compactor publish + their own notices through it directly (exactly as they do now), while the + service's typed events reach it via :func:`legacy_event_sink`. + + ``gate`` present means the workspace asked to confirm commands; pass the + request with ``gate_mode="confirm"`` so the two agree. A gate of ``None`` + keeps the pre-existing auto-run behaviour. + """ + from ...core import agent_roles + + tools = CoreToolRuntime( + output_dir, title=title, extra_tools=extra_tools, extra_executor=extra_executor, + security_config=security_config, agent_role=agent_role or agent_roles.COWORK, + ) + + def prepare_prompt(messages: List[Dict[str, Any]], advertised: Tuple[str, ...]) -> None: + """Insert the system prompt, then fold in skills, rules and project text. + + ``advertised`` is unused on purpose: the runtime decides the MS365 + paragraph from the CONFIGURED connector tools, not from the subset a + capability scope left advertised. Changing that changes the prompt the + model sees, so it stays as-is here and belongs to R05's tool-policy work. + """ + from ...core.chat_agent import ( + COWORK_TOOL_PROMPT, + OPENDATALOADER_PDF_PROMPT, + _apply_project_context, + _apply_security_rules, + _apply_skills, + ) + from ...core.deps import _can_pip + from ...core.java_runtime import find_java + from ...core.security_rules import load_rules + from ...core.skills import active_skills_text + + if not messages or messages[0].get("role") != "system": + system = COWORK_TOOL_PROMPT + if any(n.startswith("ms365_") for n in tools.extra_names): + system += ("\nThe user has signed in to Microsoft 365 and enabled some ms365__* " + "tools (Outlook / Teams / OneDrive / SharePoint / meeting transcripts, " + "via the built-in MS365 MCP server). Use them whenever the request " + "involves that data — don't say you can't access it.") + if find_java() is not None and _can_pip(): + # Only advertise the Java-backed PDF extractor when BOTH the JVM + # and pip are available, so the agent is never steered into a + # command that cannot work on this machine. + system += "\n\n" + OPENDATALOADER_PDF_PROMPT + messages.insert(0, {"role": "system", "content": system}) + _apply_skills(messages, active_skills_text()) + _apply_security_rules(messages, load_rules()) + _apply_project_context(messages, project_context) + + def prompt_guard(messages: List[Dict[str, Any]]) -> None: + from ...core import agent_security + + agent_security.enforce_prompt(provider, messages, security_config, emit) + + def command_guard(name: str, args: Dict[str, Any]) -> None: + from ...core import agent_security + + agent_security.enforce_command(provider, name, args, security_config, emit) + + def compact(messages: List[Dict[str, Any]], cancel) -> None: + from ...core import context_budget + + context_budget.maybe_compact(provider, messages, security_config, + emit=emit, cancel=cancel) + + return ConversationApplicationService( + CoreModelCall(provider), tools, + prepare_prompt=prepare_prompt, + prompt_guard=prompt_guard, + command_guard=command_guard, + compact=compact, + permission_request=(gate.request if gate is not None else None), + ) + + +__all__ = [ + "LegacyEmit", "legacy_event_sink", "CoreModelCall", "CoreToolRuntime", + "build_cowork_conversation_service", +] diff --git a/application/conversations/cowork_turn_request.py b/application/conversations/cowork_turn_request.py new file mode 100644 index 0000000..777f1c5 --- /dev/null +++ b/application/conversations/cowork_turn_request.py @@ -0,0 +1,77 @@ +"""Turn the Cowork widget's captured state into a request (R04-T04). + +``ui/cowork_tab.py::build_job`` reads a dozen values off the widget on the UI +thread and has to translate three of them before a turn can run: which message +is this turn's prompt, which messages are its history, and whether the workspace +wants commands confirmed. Those rules lived inline in the widget, where no test +could reach them — and each fails silently when wrong (a duplicated user message, +or a command that quietly stops asking for approval). + +They live here instead, as the mapping step the migration map assigns to the +application layer. The widget keeps only what is genuinely widget-specific: +reading its own state and building the provider. + +Layer rules (``docs/architecture/ADR-001-layered-architecture.md``): pure Python. +Everything arrives as a plain value, so this module never sees a widget. +""" + +from __future__ import annotations + +from typing import Any, Dict, Optional, Sequence + +from ...domain.agents.conversation_execution_request import ConversationExecutionRequest + + +def build_cowork_turn_request( + *, + turn_id: str, + session_id: str, + messages: Sequence[Dict[str, Any]], + surface: str = "cowork", + project_id: str = "", + title: str = "", + provider_id: str = "", + model: str = "", + instructions: str = "", + output_dir: Optional[Any] = None, + home_output_root: Optional[Any] = None, + confirm_commands: bool = False, + agent_role: str = "cowork", +) -> ConversationExecutionRequest: + """Build one Cowork turn's immutable request. + + ``messages`` is the widget's working list, which ALREADY ends with this + turn's user message (the chat panel composes it — prefix, attachments, + session notes — before the job starts). So the prompt is that last message + and the history is everything before it. The request records both; the + service is handed the same working list and appends into it. + + Keyword-only on purpose: a dozen positional strings in a call site is exactly + how a title ends up in the project-id slot. + """ + history = list(messages or ()) + # ``pop`` rather than ``[-1]``/``[:-1]`` so the empty-list case needs no + # special branch: a turn with nothing in it yields an empty prompt instead of + # raising IndexError deep inside a worker thread. + last = history.pop() if history else {} + return ConversationExecutionRequest( + turn_id=turn_id, + session_id=session_id, + surface=surface, + project_id=project_id, + title=title, + prompt=str(last.get("content") or ""), + messages=history, + provider_id=provider_id, + model=model, + project_context=instructions, + output_dir=output_dir, + home_output_root=home_output_root, + # The workspace's Auto-run override (or the global setting) decides + # whether run_command/install_package must be approved first. + gate_mode="confirm" if confirm_commands else "auto", + agent_role=agent_role, + ) + + +__all__ = ["build_cowork_turn_request"] diff --git a/application/conversations/turn_runtime.py b/application/conversations/turn_runtime.py new file mode 100644 index 0000000..5ed6695 --- /dev/null +++ b/application/conversations/turn_runtime.py @@ -0,0 +1,177 @@ +"""The seams :mod:`conversation_application_service` runs a turn through (R04-T03). + +Two Protocols and six callables — chosen deliberately, not by reflex. The +refactor plan forbids giving every class an interface, so a contract exists here +only where there is both a real ``core/*`` implementation AND a test double: + +* :class:`ModelCallPort` — one provider round-trip *including* the app's + existing context-overflow recovery, which is why the raw ``Provider.chat`` + signature is not enough. +* :class:`ToolRuntimePort` — the tool + output-folder runtime, kept as one + cohesive object because every method operates on the same sandbox. + +Everything else is a single function, so it is expressed as a callable type +rather than a class with one method (the same choice R03 made for +``ConfirmationCallback``). All of them are optional: a service built with none +of them still runs a plain chat turn, which is what keeps the unit tests short. + +Layer rules (``docs/architecture/ADR-001-layered-architecture.md``): application +layer — pure Python. Nothing here imports PySide6, ``core.*``, ``providers.*`` +or ``ui.*``; the concrete wiring lives in :mod:`core_runtime_adapter`. +""" + +from __future__ import annotations + +from typing import ( + Any, + Callable, + Dict, + List, + Optional, + Protocol, + Sequence, + Tuple, + runtime_checkable, +) + +from ...domain.agents.agent_event import AgentEvent, ToolPreview + +# The plan tool is special-cased by the loop: it drives the Plan panel and +# produces no chat bubble and no file. Named here so the check is not a bare +# string literal in the middle of the dispatch. +PLAN_TOOL = "update_plan" + +# Tools that need approval before they run when the workspace is in confirm +# mode. R05 replaces this tuple with a real ``ToolPolicyGateway`` keyed on +# ToolCapability; until then it mirrors exactly what the runtime gates today. +GATED_TOOLS = ("run_command", "install_package") + +# Shown when the user (or the workspace policy) rejects a proposed command. The +# exact string also becomes the tool message the model reads back, so it must +# stay stable. +REJECTED_OUTPUT = "Rejected by user." + +# A reasoning model can answer with thinking only. The note is written into the +# assistant message itself, not merely emitted, so an unattended run does not +# read back an empty answer and report "(no output)". +REASONING_ONLY_NOTE = "*(model returned only its reasoning — try rephrasing)*" + +# Emitted when the turn is stopped by its own safety ceiling rather than by the +# model finishing. Never silent: being cut off looks exactly like being done. +BUDGET_NOTE_TEMPLATE = ( + "\n\n⚠️ Reached the {steps}-step safety limit before the task signalled " + "completion — stopping here. Re-run to continue if more work remains." +) + + +def combine_instructions(*blocks: Optional[str]) -> str: + """Join the standing-instruction blocks of a turn, skipping the absent ones. + + A turn's instructions arrive as several independent blocks — the project's + shared context, an Admin agent's persona, a skill's rules, the + "this runs unattended" reminder — and each caller was joining them inline + with its own ``f"{a}\\n\\n{b}" if a else b`` expression. Two call sites now + need the same rule (the Cowork widget in R04-T04 and the task runner in + R04-T05), which is the point at which it stops being an expression. + + Whitespace-only blocks count as absent: they would otherwise open the system + prompt with a stray blank line. + """ + return "\n\n".join(b.strip() for b in blocks if b and b.strip()) + + +# --------------------------------------------------------------------------- # +# Callables. +# --------------------------------------------------------------------------- # +# Receives every typed event the turn produces. The caller decides what that +# means — render it, forward it as a legacy dict, autosave on it. +EventSink = Callable[[AgentEvent], None] + +# True once the user has asked to stop. Polled between steps and between tool +# calls, the same cadence the current runtime uses. +CancelFn = Callable[[], bool] + +# ``(prompt, attachment_paths) -> body``. Runs on the worker thread because +# extracting a .docx may pip-install a parser or call LibreOffice. +AttachmentReader = Callable[[str, Tuple[str, ...]], str] + +# ``(messages, advertised_tool_names) -> None`` — inserts the system prompt and +# folds in skills, security rules and project instructions, in place. It needs +# the tool names because the system prompt gains an MS365 paragraph only when +# ms365 tools are actually present. +PromptPreparer = Callable[[List[Dict[str, Any]], Tuple[str, ...]], None] + +# Reviews the assembled request; raises to refuse the turn outright. +PromptGuard = Callable[[List[Dict[str, Any]]], None] + +# Reviews one proposed tool call; raises to refuse it. +CommandGuard = Callable[[str, Dict[str, Any]], None] + +# ``(messages, cancel) -> None``. Summarises old turns in place when the +# conversation nears the model's context budget; a no-op when compaction is off +# or the conversation is short. It takes the cancel signal because compacting +# calls the model itself, so Stop has to reach it too. +ContextCompactor = Callable[[List[Dict[str, Any]], "CancelFn"], None] + +# ``(action) -> approved``. Blocks the worker thread while a human decides. +PermissionRequest = Callable[[Dict[str, Any]], bool] + + +# --------------------------------------------------------------------------- # +# Ports. +# --------------------------------------------------------------------------- # +@runtime_checkable +class ModelCallPort(Protocol): + """One call to the model, with the app's retry/recovery behaviour applied.""" + + def call(self, messages: List[Dict[str, Any]], tools: Sequence[Any], + on_text: Optional[Callable[[str], None]] = None, + on_reasoning: Optional[Callable[[str], None]] = None, + cancel: Optional[CancelFn] = None) -> Dict[str, Any]: + """Return the canonical assistant message (content plus tool calls).""" + + +@runtime_checkable +class ToolRuntimePort(Protocol): + """The tools a turn may call, and the folder its files land in.""" + + def specs(self, allowed_tools: Optional[Sequence[str]] = None) -> Sequence[Any]: + """Tool specs to advertise to the model, already filtered. + + Returns opaque objects (the provider layer's ``ToolSpec``); the service + only ever reads ``.name`` off them, which is what keeps this layer free + of a provider import. + """ + + def preview(self, name: str, args: Dict[str, Any]) -> Optional[ToolPreview]: + """Human-readable description of a call that is about to run.""" + + def execute(self, name: str, args: Dict[str, Any], + on_output: Optional[Callable[[str], None]] = None, + cancel: Optional[CancelFn] = None) -> Dict[str, Any]: + """Run one tool call. + + Returns the runtime's own result mapping: ``ok``, ``output``, optionally + ``path``/``produced`` for files it created, and ``plan_steps`` for the + plan tool. + """ + + def snapshot(self) -> Any: + """Opaque record of the output folder before the turn started.""" + + def finalize(self, before: Any, cancelled: bool = False + ) -> Tuple[Sequence[str], Sequence[str]]: + """Tidy the output folder; return ``(removed_paths, added_paths)``. + + Not read-only — it deletes the scratch sandbox and flattens sub-folders — + so the service only calls it for a turn that actually started. + """ + + +__all__ = [ + "PLAN_TOOL", "GATED_TOOLS", "REJECTED_OUTPUT", "REASONING_ONLY_NOTE", + "BUDGET_NOTE_TEMPLATE", "combine_instructions", + "EventSink", "CancelFn", "AttachmentReader", "PromptPreparer", "PromptGuard", + "CommandGuard", "ContextCompactor", "PermissionRequest", + "ModelCallPort", "ToolRuntimePort", +] diff --git a/core/task_executors.py b/core/task_executors.py index 3d8bf43..2faf485 100644 --- a/core/task_executors.py +++ b/core/task_executors.py @@ -2,7 +2,9 @@ ``execute_task`` dispatches by ``task_type`` to the app's existing engines: -- ``cowork`` → ``chat_agent.run_cowork`` (documents/answers, real files) +- ``cowork`` → ``ConversationApplicationService`` (documents/answers, real + files) — the same turn engine the interactive Cowork chat + runs on since R04-T05 - ``co4e_code`` → ``code_agent.run_code`` (code agent with file/command tools) - ``script`` → local subprocess with a timeout - ``flow`` → the task's own simple step list, run sequentially, each @@ -162,6 +164,30 @@ _TIMEOUT_NOTICE_TMPL = ( ) +_UNATTENDED_PREFIX = ( + "This runs unattended (Schedule Task) — no one is watching live. Use " + "update_plan to track your steps and keep it accurate: mark a step " + "'error' (not silently skip it) if it genuinely can't be completed." +) + + +def _unattended_prompt(prompt: str, *, skill_text: str = "", + agent_instructions: str = "") -> str: + """Assemble the user message an unattended run sends. + + The order is load-bearing and used to be encoded as three successive + rebindings of ``prompt``, each prepending its own block: the plan reminder + must lead (it is the instruction that keeps a run without a human watching + honest), then the chosen skill's rules, then the Admin agent's persona, and + the task's own words last. Routing it through ``combine_instructions`` keeps + that order in one readable expression and drops the absent blocks instead of + leaving blank lines behind. + """ + from ..application.conversations.turn_runtime import combine_instructions + + return combine_instructions(_UNATTENDED_PREFIX, skill_text, agent_instructions, prompt) + + def _cancel_with_timeout(cancel: CancelFn, timeout_sec: Optional[int]) -> Tuple[CancelFn, Callable[[], bool]]: """Wrap ``cancel`` so it also fires once ``timeout_sec`` of wall-clock time elapses. ``timed_out()`` tells the caller whether THAT is why it stopped @@ -218,36 +244,29 @@ def _run_agent(ctx, task_type: str, prompt: str, out_dir: Path, # default, see state.build_provider_for). A legacy Admin-agent preset # (task.admin_agent_id), if still set on an older task, keeps working and # takes precedence — it pins the provider/model AND prepends instructions. + agent_instructions = "" if admin_agent is not None: from .admin_agents import build_agent_provider provider = build_agent_provider(ctx, admin_agent) agent_instructions = admin_agent.effective_prompt() - if agent_instructions: - prompt = f"{agent_instructions}\n\n{prompt}" elif provider_name or model: # An explicit per-task provider/model override. provider = ctx.build_provider_for(provider_name or None, model or None) else: # Neither overridden → the machine's own Settings default, exactly as before. provider = ctx.build_active_provider() - # A chosen skill's instructions are prepended so this unattended run follows + # A chosen skill's instructions are applied so this unattended run follows # them, mirroring how the interactive chat applies /skill. + skill_text = "" if skill_slug: from .skills import skill_prefix_for skill_text = skill_prefix_for(skill_slug) - if skill_text: - prompt = f"{skill_text}\n\n{prompt}" - # This is an UNATTENDED run (no human watching to catch a half-finished - # job) — push the agent to actually use the Plan checklist so completion - # can be verified afterward, instead of just trusting "no exception". - prompt = ( - "This runs unattended (Schedule Task) — no one is watching live. Use " - "update_plan to track your steps and keep it accurate: mark a step " - "'error' (not silently skip it) if it genuinely can't be completed.\n\n" - f"{prompt}" - ) + # Assemble reminder + skill + persona + the task's own words in one place + # (see _unattended_prompt for why that order matters). + prompt = _unattended_prompt(prompt, skill_text=skill_text, + agent_instructions=agent_instructions) messages = [{"role": "user", "content": prompt}] session_id = new_session_id() project_id = project.project_id if project is not None else "" @@ -273,10 +292,41 @@ def _run_agent(ctx, task_type: str, prompt: str, out_dir: Path, 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) + # R04-T05: the unattended run shares the interactive turn engine + # instead of calling run_cowork itself, so there is exactly one place + # where a turn's lifecycle is defined. Everything unattended-specific + # stays here (the plan reminder above, the History autosave in + # emit_and_autosave, the timeout notice below). + from ..application.conversations.core_runtime_adapter import ( + build_cowork_conversation_service, + legacy_event_sink, + ) + from ..domain.agents.conversation_execution_request import ( + ConversationExecutionRequest, + ) + + # No extra_tools/extra_executor and no permission gate: a scheduled + # run gets no MCP connectors and nobody is there to approve a + # command, which is exactly what run_cowork was called with. + service = build_cowork_conversation_service( + provider, out_dir, emit_and_autosave, title=title, + project_context=project_context, security_config=ctx.config, + agent_role=agent_roles.TASK, + ) + request = ConversationExecutionRequest( + # The artifact folder is named by the run id, which identifies + # this attempt in the audit log. + turn_id=out_dir.name or session_id, session_id=session_id, + surface="task", title=title, project_id=project_id, + prompt=prompt, output_dir=out_dir, + agent_role=agent_roles.TASK, unattended=True, + timeout_sec=timeout_sec, + ) + # ``messages`` is handed over so the History autosave in + # emit_and_autosave (and the final save in the finally block below) + # keep reading the live conversation as it grows. + service.execute(request, legacy_event_sink(emit_and_autosave), + cancel=watched_cancel, messages=messages) 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 c857c66..377809b 100644 --- a/docs/refactor/Refactoring_Checklist.md +++ b/docs/refactor/Refactoring_Checklist.md @@ -135,16 +135,22 @@ * **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-23 00:56` | End: `2026-08-23 01:00`* +- [x] **R04-T02 (Team Duy)**: Chuẩn hóa các sự kiện `AgentEvent` (TextChunk, ToolCallStarted, ToolCallResult, Error) ➔ `domain/agents/agent_event.py` (+ `domain/agents/agent_event_codec.py` — shim dịch legacy dict, tách riêng để giữ LOC < 400 và để xoá gọn sau R08) + *Start: `2026-08-23 01:00` | End: `2026-08-23 01:08`* +- [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` (+ `turn_runtime.py` định nghĩa 2 port/6 callable, `core_runtime_adapter.py` cầu nối sang `core/*`, `domain/agents/agent_result.py`) + *Start: `2026-08-23 01:08` | End: `2026-08-23 07:10`* + Chưa đổi call site nào — `run_cowork` giữ nguyên (Co4E vẫn dùng); việc chuyển call site là T04/T05. Bằng chứng tương đương: `tests/integration/test_conversation_service_parity.py` chạy cùng 1 script provider qua 2 đường và so khớp từng event/message/tool list trên 7 kịch bản. +- [x] **R04-T04 (Team Duy)**: Di chuyển `ui/cowork_tab.py::build_job` sang sử dụng `ConversationExecutionRequest` + *Start: `2026-08-23 07:10` | End: `2026-08-23 07:23`* + `build_job` không còn gọi `run_cowork`: nó chụp state widget tại submit time ➔ `build_cowork_turn_request()` (mới, `application/conversations/cowork_turn_request.py`) ➔ `ConversationApplicationService`. Thêm `combine_instructions()` vào `turn_runtime.py` (project context + admin agent, T05 dùng lại) và tham số `messages=` cho `execute()` để service append vào **đúng list của widget** — `_reattach_running_turn` đọc list đó trong lúc turn đang chạy và `_finalize_turn` slice nó sau đó. Kiểm chứng: `tests/integration/test_cowork_tab_turn.py` gọi thẳng `CoworkTab.build_job` (widget stub, không cần Qt) và chạy turn thật với `FakeProvider`. + ⚠️ `ui/cowork_tab.py` 416 ➔ 455 LOC — file này **vốn đã vượt 400 trước khi sửa**; phân rã thuộc EPIC R08. +- [x] **R04-T05 (Team Duy)**: Di chuyển `core/task_executors.py` sang dùng chung `ConversationApplicationService` + *Start: `2026-08-23 07:23` | End: `2026-08-23 07:31`* + Nhánh `task_type == "cowork"` của `_run_agent` gọi service thay vì `run_cowork`; 5 hành vi riêng của unattended run giữ nguyên (plan reminder, `history_ready`, autosave History mỗi `assistant_done`, timeout notice, `plan_incomplete_reason`). Tách `_unattended_prompt()` dùng `combine_instructions` để thứ tự reminder → skill → persona → prompt nằm ở 1 chỗ đọc được. Lưới an toàn: `tests/integration/test_task_executor_turn.py` viết **trước** khi migrate và pass 8/8 trên code cũ, vẫn pass sau khi migrate. + ⚠️ `core/task_executors.py` 476 ➔ 524 LOC — file này **vốn đã vượt 400 trước khi sửa**; phân rã thuộc EPIC R07 (`application/scheduling/`). + Còn lại gọi `run_cowork`: `core/co4e_runner.py` (×2) và `ui/co4e_tab.py` — phân hệ Co4E của 🟣 Team Nam, R04 không chạm theo luật 1 file 1 team. --- diff --git a/tests/fakes/turn_runtime_fakes.py b/tests/fakes/turn_runtime_fakes.py new file mode 100644 index 0000000..ca1255a --- /dev/null +++ b/tests/fakes/turn_runtime_fakes.py @@ -0,0 +1,141 @@ +"""Offline test doubles for the R04 turn runtime seams. + +Sits beside ``fake_provider.py``/``fake_tool_executor.py`` (R01-T02) and plays +the same role one level up: those fake a *provider*, these fake the ports +``ConversationApplicationService`` is driven through +(``application/conversations/turn_runtime.py``). + +Deliberately dumb — they record what they were asked and return canned answers. +A failing test then points at the service under test rather than at a mock +framework's configuration. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Tuple + +from cowork_local.domain.agents.agent_event import ToolPreview +from cowork_local.domain.agents.conversation_execution_request import ( + ConversationExecutionRequest, +) + + +class FakeSpec: + """An advertised tool. The service only ever reads ``.name`` off a spec.""" + + def __init__(self, name: str) -> None: + self.name = name + + +class FakeReply: + """One programmed provider answer.""" + + def __init__(self, content: str = "", tool_calls=None, chunks=None, reasoning: str = ""): + self.content = content + self.tool_calls = tool_calls or [] + # Default to streaming the whole content as a single chunk, which is what + # a non-streaming gateway effectively does. + self.chunks = chunks if chunks is not None else ([content] if content else []) + self.reasoning = reasoning + + +class FakeModelCall: + """:class:`ModelCallPort` returning programmed replies in order. + + A programmed entry may be an exception instead of a reply, which is how a + test simulates the gateway dying mid-turn. + """ + + def __init__(self, replies: List[Any]) -> None: + self.replies = list(replies) + self.calls: List[Dict[str, Any]] = [] + + def call(self, messages, tools, on_text=None, on_reasoning=None, cancel=None): + # Snapshot the messages: the service keeps mutating its own list, so + # storing it by reference would make every recorded call look identical. + self.calls.append({"messages": [dict(m) for m in messages], + "tool_names": [getattr(t, "name", "") for t in tools]}) + reply = self.replies.pop(0) if self.replies else FakeReply(content="(default)") + if isinstance(reply, BaseException): + raise reply + if reply.reasoning and on_reasoning: + on_reasoning(reply.reasoning) + for chunk in reply.chunks: + if on_text and chunk: + on_text(chunk) + assistant: Dict[str, Any] = {"role": "assistant", "content": reply.content} + if reply.tool_calls: + assistant["tool_calls"] = reply.tool_calls + return assistant + + +class FakeToolRuntime: + """:class:`ToolRuntimePort` over an imaginary output folder.""" + + def __init__(self, specs=("save_file", "run_command", "update_plan"), + results: Optional[Dict[str, Dict[str, Any]]] = None, + removed: Tuple[str, ...] = (), added: Tuple[str, ...] = ()) -> None: + self._specs = [FakeSpec(n) for n in specs] + self._results = results or {} + self._removed, self._added = removed, added + self.executed: List[Tuple[str, Dict[str, Any]]] = [] + self.finalize_calls: List[Dict[str, Any]] = [] + # When set, every executed tool streams this string through ``on_output``. + self.emit_output: Optional[str] = None + + def specs(self, allowed_tools=None): + if allowed_tools is None: + return list(self._specs) + return [s for s in self._specs if s.name in allowed_tools] + + def preview(self, name, args): + return ToolPreview(kind="info", title=name, text=str(args)) + + def execute(self, name, args, on_output=None, cancel=None): + self.executed.append((name, dict(args))) + if self.emit_output and on_output: + on_output(self.emit_output) + return dict(self._results.get(name, {"ok": True, "output": f"{name} ok"})) + + def snapshot(self): + return "before" + + def finalize(self, before, cancelled=False): + self.finalize_calls.append({"before": before, "cancelled": cancelled}) + return list(self._removed), list(self._added) + + +# --------------------------------------------------------------------------- # +# Small helpers shared by the turn tests. +# --------------------------------------------------------------------------- # +def make_request(**overrides) -> ConversationExecutionRequest: + """A minimal valid request; each test overrides only what it exercises.""" + base: Dict[str, Any] = {"turn_id": "t1", "session_id": "s1", "prompt": "do it"} + base.update(overrides) + return ConversationExecutionRequest(**base) + + +def run_turn(service, request=None, cancel=None): + """Execute a turn and return ``(result, events)``.""" + events: List[Any] = [] + result = service.execute(request or make_request(), events.append, cancel=cancel) + return result, events + + +def events_of_type(events, cls): + """Every emitted event of one type, in order.""" + return [e for e in events if isinstance(e, cls)] + + +def tool_turn(tool_name: str = "save_file", args=None, **tool_kwargs): + """A turn that calls one tool and then answers — ``(model, tools)``.""" + calls = [{"id": "c1", "name": tool_name, "arguments": args or {"filename": "a.md"}}] + model = FakeModelCall([FakeReply(content="working", tool_calls=calls), + FakeReply(content="done")]) + return model, FakeToolRuntime(**tool_kwargs) + + +__all__ = [ + "FakeSpec", "FakeReply", "FakeModelCall", "FakeToolRuntime", + "make_request", "run_turn", "events_of_type", "tool_turn", +] diff --git a/tests/integration/test_conversation_service_parity.py b/tests/integration/test_conversation_service_parity.py new file mode 100644 index 0000000..2d54135 --- /dev/null +++ b/tests/integration/test_conversation_service_parity.py @@ -0,0 +1,207 @@ +"""R04-T03 (c) — the service must behave exactly like ``run_cowork``. + +The unit tests prove the loop follows the rules I wrote down. They cannot prove +those rules are the ones the shipped runtime actually follows. This file does: +each test scripts one provider, runs the SAME turn twice — once through +``core/chat_agent.py::run_cowork``, once through +``ConversationApplicationService`` wired by ``core_runtime_adapter`` — and +compares the emitted event stream, the resulting conversation and the tool list +the model was shown. + +Anything the port got wrong (a missing event, a reordered guard, a different +tool set, a changed message) fails here rather than in front of a user. The only +allowed difference is the extra ``turn_completed`` event R04 introduces, which +has no legacy consumer. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from cowork_local.application.conversations.core_runtime_adapter import ( + build_cowork_conversation_service, + legacy_event_sink, +) +from cowork_local.core import chat_agent +from cowork_local.domain.agents.conversation_execution_request import ( + ConversationExecutionRequest, +) +from cowork_local.tests.fakes.fake_provider import FakeProvider + +_USER_TURN = [{"role": "user", "content": "make me a report"}] + + +class _FakeGate: + """Stands in for ``core/permissions.py::PermissionGate``.""" + + def __init__(self, approve: bool) -> None: + self.approve = approve + self.requests: List[Dict[str, Any]] = [] + + def request(self, action: Dict[str, Any]) -> bool: + self.requests.append(action) + return self.approve + + +def _normalise(events: List[Dict[str, Any]], out_dir: Path) -> List[Dict[str, Any]]: + """Replace the run's own output path with a placeholder. + + The two runs write into different temp folders, so absolute paths in + ``tool_result``/``outputs_*`` events differ by construction. Everything else + must match verbatim. + """ + marker, raw = "", str(out_dir) + + def scrub(value: Any) -> Any: + if isinstance(value, str): + return value.replace(raw, marker).replace(raw.replace("\\", "/"), marker) + if isinstance(value, list): + return [scrub(v) for v in value] + if isinstance(value, dict): + return {k: scrub(v) for k, v in value.items()} + return value + + return [scrub(e) for e in events] + + +def _run_legacy(tmp_path: Path, provider: FakeProvider, *, allowed_tools=None, + gate: Optional[_FakeGate] = None, max_steps: int = 30 + ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[str]]: + """Run the turn through the existing ``run_cowork``.""" + out_dir = tmp_path / "legacy" + out_dir.mkdir(parents=True, exist_ok=True) + events: List[Dict[str, Any]] = [] + messages = [dict(m) for m in _USER_TURN] + + chat_agent.run_cowork( + provider, messages, out_dir, events.append, title="Report", + security_config=None, allowed_tools=allowed_tools, gate=gate, max_steps=max_steps, + ) + tool_names = [t.name for t in (provider.last_tools or [])] + return _normalise(events, out_dir), messages, tool_names + + +def _run_service(tmp_path: Path, provider: FakeProvider, *, allowed_tools=None, + gate: Optional[_FakeGate] = None, max_steps: int = 30 + ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[str]]: + """Run the same turn through the application service.""" + out_dir = tmp_path / "service" + out_dir.mkdir(parents=True, exist_ok=True) + events: List[Dict[str, Any]] = [] + + service = build_cowork_conversation_service( + provider, out_dir, events.append, title="Report", security_config=None, gate=gate) + request = ConversationExecutionRequest( + turn_id="t1", session_id="s1", + # run_cowork receives the user message already appended; the request + # carries the history and this turn's prompt separately. + messages=_USER_TURN[:-1], prompt=_USER_TURN[-1]["content"], + output_dir=out_dir, allowed_tools=allowed_tools, max_steps=max_steps, + gate_mode="confirm" if gate is not None else "auto", + ) + result = service.execute(request, legacy_event_sink(events.append)) + + # The end-of-turn event is new in R04 and has no legacy counterpart. + kept = [e for e in events if e.get("type") != "turn_completed"] + tool_names = [t.name for t in (provider.last_tools or [])] + return _normalise(kept, out_dir), list(result.messages), tool_names + + +def _assert_parity(tmp_path: Path, script, *, approve: Optional[bool] = None, **kwargs) -> None: + """Script two identical providers, run both paths, compare everything.""" + legacy_provider, service_provider = FakeProvider(), FakeProvider() + script(legacy_provider) + script(service_provider) + + legacy_gate = _FakeGate(approve) if approve is not None else None + service_gate = _FakeGate(approve) if approve is not None else None + + legacy_events, legacy_messages, legacy_tools = _run_legacy( + tmp_path, legacy_provider, gate=legacy_gate, **kwargs) + service_events, service_messages, service_tools = _run_service( + tmp_path, service_provider, gate=service_gate, **kwargs) + + assert service_events == legacy_events + assert service_messages == legacy_messages + assert service_tools == legacy_tools + if legacy_gate is not None and service_gate is not None: + assert [r["name"] for r in service_gate.requests] == \ + [r["name"] for r in legacy_gate.requests] + + +# --------------------------------------------------------------------------- # +# Scenarios. +# --------------------------------------------------------------------------- # +def test_a_plain_answer_turn_behaves_identically(tmp_path: Path) -> None: + def script(provider: FakeProvider) -> None: + provider.queue_response(content="Here you go.", chunks=["Here ", "you go."]) + + _assert_parity(tmp_path, script) + + +def test_a_save_file_turn_behaves_identically(tmp_path: Path) -> None: + def script(provider: FakeProvider) -> None: + provider.queue_response( + content="Writing it.", + tool_calls=[{"id": "c1", "name": "save_file", + "arguments": {"filename": "report.md", "content": "# Report\n"}}], + ) + provider.queue_response(content="Saved.") + + _assert_parity(tmp_path, script) + + +def test_an_update_plan_turn_behaves_identically(tmp_path: Path) -> None: + def script(provider: FakeProvider) -> None: + provider.queue_response( + content="Planning.", + tool_calls=[{"id": "c1", "name": "update_plan", + "arguments": {"steps": [{"title": "Draft", "status": "running"}, + {"title": "Ship", "status": "pending"}]}}], + ) + provider.queue_response(content="Done.") + + _assert_parity(tmp_path, script) + + +def test_a_reasoning_only_reply_behaves_identically(tmp_path: Path) -> None: + def script(provider: FakeProvider) -> None: + provider.queue_response(content="", reasoning="thinking hard") + + _assert_parity(tmp_path, script) + + +def test_restricting_the_tool_scope_advertises_the_same_tools(tmp_path: Path) -> None: + def script(provider: FakeProvider) -> None: + provider.queue_response(content="ok") + + _assert_parity(tmp_path, script, allowed_tools=["save_file"]) + + +def test_a_rejected_command_behaves_identically(tmp_path: Path) -> None: + # The security-critical path: the gate says no, so the command must never + # run and the model must read back the same refusal in both designs. + def script(provider: FakeProvider) -> None: + provider.queue_response( + content="Running it.", + tool_calls=[{"id": "c1", "name": "run_command", + "arguments": {"command": "echo hi"}}], + ) + provider.queue_response(content="Understood.") + + _assert_parity(tmp_path, script, approve=False) + + +def test_hitting_the_step_ceiling_behaves_identically(tmp_path: Path) -> None: + # The model never stops calling tools, so both paths must stop at the same + # place and say so the same way. + def script(provider: FakeProvider) -> None: + for i in range(4): + provider.queue_response( + content=f"step {i}", + tool_calls=[{"id": f"c{i}", "name": "save_file", + "arguments": {"filename": f"f{i}.md", "content": "x"}}], + ) + + _assert_parity(tmp_path, script, max_steps=2) diff --git a/tests/integration/test_cowork_tab_turn.py b/tests/integration/test_cowork_tab_turn.py new file mode 100644 index 0000000..c95dc72 --- /dev/null +++ b/tests/integration/test_cowork_tab_turn.py @@ -0,0 +1,220 @@ +"""R04-T04 — the migrated Cowork call site, exercised end to end without Qt. + +``CoworkTab.build_job`` only ever *reads attributes* off its widget, so the real +production method can be invoked against a stand-in that supplies those +attributes. That is what happens here: the actual ``build_job`` body runs, builds +a request, wires the service through ``core_runtime_adapter``, and drives a real +turn (real tool execution, real output-folder cleanup) against ``FakeProvider``. + +Why it matters: this is the only automated check that the widget's contract with +the service still holds — that the worker's list is appended to in place (the +transcript re-render and history merge both read it), that events still arrive as +legacy dicts, and that a produced file really lands in the turn's folder. None of +it needs a display server, so it runs in CI like every other test. +""" + +from __future__ import annotations + +import copy +from pathlib import Path +from typing import Any, Dict, List, Optional + +import pytest +from cowork_local.config import DEFAULT_CONFIG, AppConfig +from cowork_local.tests.fakes.fake_provider import FakeProvider + + +class _FakeWorker: + """The parts of ``core/worker.py::AgentWorker`` a job actually touches.""" + + def __init__(self, approve_commands: bool = True) -> None: + self.events: List[Dict[str, Any]] = [] + self.gate: Optional[Any] = None + self._approve = approve_commands + self.cancelled = False + + def emit_event(self, event: Dict[str, Any]) -> None: + self.events.append(event) + + def is_cancelled(self) -> bool: + return self.cancelled + + def new_gate(self, mode: str, agent_role: str = "") -> Any: + # Mirrors AgentWorker.new_gate: the gate is stored on the worker so the + # UI thread can resolve it, and answers request() from the worker thread. + worker = self + + class _Gate: + requests: List[Dict[str, Any]] = [] + + def request(self, action: Dict[str, Any]) -> bool: + self.requests.append(action) + return worker._approve + + self.gate = _Gate() + return self.gate + + +class _FakeCtx: + """The ``AppContext`` surface ``build_job`` uses.""" + + def __init__(self, config: AppConfig, confirm_commands: bool = False) -> None: + self.config = config + self._confirm = confirm_commands + + def project_confirm_commands(self) -> bool: + return self._confirm + + def build_mcp_tools(self): + return [], None + + +class _WidgetStub: + """Stands in for the CoworkTab instance ``build_job`` reads its state from.""" + + kind = "cowork" + + def __init__(self, out_root: Path, ctx: _FakeCtx, provider: FakeProvider) -> None: + self._out_root = out_root + self.ctx = ctx + self._provider = provider + self.title = "Report" + self.session_id = "s1" + self.project_id = "" # the auto-seeded default workspace + self._model = "" + self._routed_provider = None + self._routed_model = None + + def _session_output_dir(self) -> Path: + return self._out_root + + def workspace_dir(self) -> Path: + return self._out_root + + def admin_agent_prompt(self) -> str: + return "" + + def build_provider(self) -> FakeProvider: + return self._provider + + +def _config() -> AppConfig: + """A real AppConfig that never touches ``~/.cowork_local``. + + The AI security guardrails are switched off: they would call the model to + review the prompt, which is a separate feature with its own tests and would + make this one depend on what the fake answers. + """ + data = copy.deepcopy(DEFAULT_CONFIG) + data["agent_security"]["enabled"] = False + return AppConfig(data) + + +def _run_turn(tmp_path: Path, provider: FakeProvider, messages: List[Dict[str, Any]], + *, confirm_commands: bool = False, approve: bool = True): + """Invoke the real ``CoworkTab.build_job`` against the stub and run its job.""" + from cowork_local.ui.cowork_tab import CoworkTab + + out_dir = tmp_path / ".turns" / "t1" + out_dir.mkdir(parents=True, exist_ok=True) + widget = _WidgetStub(tmp_path, _FakeCtx(_config(), confirm_commands), provider) + worker = _FakeWorker(approve_commands=approve) + + job = CoworkTab.build_job(widget, "make me a report", messages, out_dir) + result = job(worker) + return result, worker + + +def test_the_turn_runs_and_reports_its_folder(tmp_path: Path) -> None: + provider = FakeProvider() + provider.queue_response(content="Here you go.", chunks=["Here ", "you go."]) + messages = [{"role": "user", "content": "make me a report"}] + + result, worker = _run_turn(tmp_path, provider, messages) + + assert result["turn_dir"] == str(tmp_path / ".turns" / "t1") + assert [e["type"] for e in worker.events] == [ + "text", "text", "assistant_done", "turn_completed"] + + +def test_the_worker_list_is_appended_to_in_place(tmp_path: Path) -> None: + # _reattach_running_turn replays from this very list while the turn runs, and + # _finalize_turn slices it by the pre-turn length afterwards. + provider = FakeProvider() + provider.queue_response(content="Done.") + user = {"role": "user", "content": "make me a report"} + messages = [user] + + result, _ = _run_turn(tmp_path, provider, messages) + + assert result["messages"] is messages + # Identity, not just equality: _reattach_running_turn locates the turn's user + # message with ``m is ctx["user_msg"]`` to replay the steps after it. + assert any(m is user for m in messages) + # Several system blocks are expected — the tool prompt plus the tagged + # skills/security-rules blocks the runtime refreshes on every turn. + assert [m["role"] for m in messages if m["role"] != "system"] == ["user", "assistant"] + assert messages[-1]["content"] == "Done." + + +def test_a_saved_file_lands_in_the_turn_folder(tmp_path: Path) -> None: + provider = FakeProvider() + provider.queue_response( + content="Writing it.", + tool_calls=[{"id": "c1", "name": "save_file", + "arguments": {"filename": "report.md", "content": "# Report\n"}}], + ) + provider.queue_response(content="Saved.") + messages = [{"role": "user", "content": "make me a report"}] + + _, worker = _run_turn(tmp_path, provider, messages) + + produced = list((tmp_path / ".turns" / "t1").glob("*.md")) + assert len(produced) == 1 + assert produced[0].read_text(encoding="utf-8") == "# Report\n" + results = [e for e in worker.events if e["type"] == "tool_result"] + assert results and results[0]["ok"] is True + + +def test_auto_run_mode_never_creates_a_permission_gate(tmp_path: Path) -> None: + provider = FakeProvider() + provider.queue_response(content="ok") + + _, worker = _run_turn(tmp_path, provider, [{"role": "user", "content": "hi"}], + confirm_commands=False) + + assert worker.gate is None + + +def test_confirm_mode_creates_the_gate_and_a_refusal_stops_the_command(tmp_path: Path) -> None: + provider = FakeProvider() + provider.queue_response( + content="Running it.", + tool_calls=[{"id": "c1", "name": "run_command", + "arguments": {"command": "echo hi"}}], + ) + provider.queue_response(content="Understood.") + + _, worker = _run_turn(tmp_path, provider, [{"role": "user", "content": "run it"}], + confirm_commands=True, approve=False) + + assert worker.gate is not None + refusals = [e for e in worker.events + if e["type"] == "tool_result" and e["output"] == "Rejected by user."] + assert len(refusals) == 1 + + +def test_cancelling_before_the_turn_starts_calls_no_model(tmp_path: Path) -> None: + from cowork_local.ui.cowork_tab import CoworkTab + + provider = FakeProvider() + provider.queue_response(content="never") + out_dir = tmp_path / ".turns" / "t1" + out_dir.mkdir(parents=True) + widget = _WidgetStub(tmp_path, _FakeCtx(_config()), provider) + worker = _FakeWorker() + worker.cancelled = True + + CoworkTab.build_job(widget, "x", [{"role": "user", "content": "x"}], out_dir)(worker) + + assert provider.call_count == 0 diff --git a/tests/integration/test_task_executor_turn.py b/tests/integration/test_task_executor_turn.py new file mode 100644 index 0000000..a79d75e --- /dev/null +++ b/tests/integration/test_task_executor_turn.py @@ -0,0 +1,191 @@ +"""R04-T05 — the Schedule Task runner's cowork branch, pinned before and after. + +Written against the CURRENT ``_run_agent`` first, as the safety net for moving it +onto ``ConversationApplicationService``: an unattended run has five behaviours the +interactive path does not have (the plan reminder prefixed to the prompt, the +session registered in History before the model starts, a re-save after every +assistant message, the timeout notice, and the "did the agent's own checklist +finish?" report), and none of them was covered by a test. + +Everything is isolated from the user's real config: history goes to ``tmp_path`` +via ``history.custom_dir`` and the AI guardrails are off, so no run touches +``~/.cowork_local`` or calls a model to review a prompt. +""" + +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import Any, Dict, List, Optional + +from cowork_local.config import DEFAULT_CONFIG, AppConfig +from cowork_local.core import task_executors +from cowork_local.tests.fakes.fake_provider import FakeProvider + + +class _FakeCtx: + """The ``AppContext`` surface ``_run_agent`` touches.""" + + def __init__(self, config: AppConfig, provider: FakeProvider) -> None: + self.config = config + self._provider = provider + + def build_active_provider(self) -> FakeProvider: + return self._provider + + def build_provider_for(self, name=None, model=None) -> FakeProvider: + return self._provider + + +def _config(tmp_path: Path) -> AppConfig: + data = copy.deepcopy(DEFAULT_CONFIG) + # Keep the run entirely offline and off the real config dir. + data["agent_security"]["enabled"] = False + data["history"]["custom_dir"] = str(tmp_path / "history") + return AppConfig(data) + + +def _run(tmp_path: Path, provider: FakeProvider, *, prompt: str = "write the report", + timeout_sec: Optional[int] = None, admin_agent: Any = None): + """Run one cowork task and return ``(result_tuple, events, config)``.""" + out_dir = tmp_path / "run" + out_dir.mkdir(parents=True, exist_ok=True) + config = _config(tmp_path) + events: List[Dict[str, Any]] = [] + + result = task_executors._run_agent( + _FakeCtx(config, provider), "cowork", prompt, out_dir, + events.append, lambda: False, title="Weekly report", + timeout_sec=timeout_sec, admin_agent=admin_agent, + ) + return result, events, config + + +def _saved_conversation(config: AppConfig) -> Dict[str, Any]: + """The single conversation the run wrote into the isolated history folder.""" + files = list(Path(config.history_dir()).rglob("*.json")) + assert len(files) == 1, f"expected one saved conversation, found {files}" + return json.loads(files[0].read_text(encoding="utf-8")) + + +# --------------------------------------------------------------------------- # +def test_a_cowork_task_returns_the_final_answer(tmp_path: Path) -> None: + provider = FakeProvider() + provider.queue_response(content="Report is ready.") + + (answer, timed_out, incomplete), _events, _config = _run(tmp_path, provider) + + assert answer == "Report is ready." + assert timed_out is False + assert incomplete == "" + + +def test_the_plan_reminder_is_prefixed_to_the_prompt(tmp_path: Path) -> None: + # An unattended run has nobody watching, so the agent is pushed to keep its + # own checklist honest. The reminder must lead the message. + provider = FakeProvider() + provider.queue_response(content="ok") + + _run(tmp_path, provider, prompt="write the report") + + sent = provider.call_history[0][-1]["content"] + assert sent.startswith("This runs unattended (Schedule Task)") + assert sent.endswith("write the report") + + +def test_an_admin_agent_persona_sits_between_the_reminder_and_the_prompt( + tmp_path: Path) -> None: + class _Agent: + # An admin agent may pin its own provider/model; blank means "use the + # machine's Settings default", which is what build_agent_provider reads. + provider = "" + model = "" + + def effective_prompt(self) -> str: + return "You are the reporting agent." + + provider = FakeProvider() + provider.queue_response(content="ok") + + _run(tmp_path, provider, prompt="write the report", admin_agent=_Agent()) + + sent = provider.call_history[0][-1]["content"] + assert sent.index("This runs unattended") < sent.index("You are the reporting agent.") + assert sent.index("You are the reporting agent.") < sent.index("write the report") + + +def test_the_session_is_announced_once_it_exists_on_disk(tmp_path: Path) -> None: + # The scheduler refreshes History on this event, so it must not fire before + # the conversation is really there. + provider = FakeProvider() + provider.queue_response(content="ok") + + _result, events, config = _run(tmp_path, provider) + + ready = [e for e in events if e["type"] == "history_ready"] + assert len(ready) == 1 + assert ready[0]["session_id"] + assert _saved_conversation(config)["session_id"] == ready[0]["session_id"] + + +def test_the_saved_conversation_carries_the_answer_and_the_task_title( + tmp_path: Path) -> None: + provider = FakeProvider() + provider.queue_response(content="Report is ready.") + + _result, _events, config = _run(tmp_path, provider) + + saved = _saved_conversation(config) + assert saved["title"] == "[Task] Weekly report" + assert saved["messages"][-1] == {"role": "assistant", "content": "Report is ready."} + + +def test_an_unfinished_checklist_is_reported_back_to_the_scheduler( + tmp_path: Path) -> None: + # The agent ticked no step to done, so the task must not be called finished + # just because no exception was raised. + provider = FakeProvider() + provider.queue_response( + content="Working on it.", + tool_calls=[{"id": "c1", "name": "update_plan", + "arguments": {"steps": [{"title": "Draft", "status": "running"}]}}], + ) + provider.queue_response(content="Stopping here.") + + (_answer, _timed_out, incomplete), _events, _config = _run(tmp_path, provider) + + assert incomplete + assert "Draft" in incomplete + + +def test_a_finished_checklist_reports_nothing_outstanding(tmp_path: Path) -> None: + provider = FakeProvider() + provider.queue_response( + content="Done.", + tool_calls=[{"id": "c1", "name": "update_plan", + "arguments": {"steps": [{"title": "Draft", "status": "done"}]}}], + ) + provider.queue_response(content="All done.") + + (_answer, _timed_out, incomplete), _events, _config = _run(tmp_path, provider) + + assert incomplete == "" + + +def test_running_out_of_time_appends_the_timeout_notice_to_the_conversation( + tmp_path: Path) -> None: + # A negative timeout puts the deadline in the past, which is the only + # deterministic way to exercise a wall-clock branch in a unit test. + provider = FakeProvider() + provider.queue_response(content="never gets there") + + (answer, timed_out, incomplete), events, config = _run( + tmp_path, provider, timeout_sec=-1) + + assert timed_out is True + assert incomplete == "" # a timeout is not an unfinished checklist + assert "quá thời gian chờ" in answer + assert any(e["type"] == "assistant_done" and "quá thời gian chờ" in e["content"] + for e in events) + assert "quá thời gian chờ" in _saved_conversation(config)["messages"][-1]["content"] diff --git a/tests/unit/test_conversation_application_service.py b/tests/unit/test_conversation_application_service.py new file mode 100644 index 0000000..38fb4a5 --- /dev/null +++ b/tests/unit/test_conversation_application_service.py @@ -0,0 +1,293 @@ +"""R04-T03 (b) — the turn loop: composition, tool dispatch, budget, cancel. + +Behaviour that used to be reachable only by running the real widget. Every +dependency is a fake from ``tests/fakes/turn_runtime_fakes.py``, so the file +runs in milliseconds and each test states one rule of the loop. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Tuple + +from cowork_local.application.conversations.conversation_application_service import ( + ConversationApplicationService, +) +from cowork_local.domain.agents.agent_event import ( + AssistantMessageCompletedEvent, + PlanStep, + PlanUpdatedEvent, + TextChunkEvent, + ToolCallFinishedEvent, + ToolCallStartedEvent, + ToolOutputChunkEvent, + ToolPreview, + TurnCompletedEvent, +) +from cowork_local.tests.fakes.turn_runtime_fakes import ( + FakeModelCall, + FakeReply, + FakeToolRuntime, + events_of_type, + make_request, + run_turn, + tool_turn, +) + + +def _service(model, tools, **overrides) -> ConversationApplicationService: + return ConversationApplicationService(model, tools, **overrides) + + +# --------------------------------------------------------------------------- # +# The happy path. +# --------------------------------------------------------------------------- # +def test_a_plain_answer_streams_text_then_reports_the_message_and_the_turn() -> None: + model = FakeModelCall([FakeReply(content="Hello there", chunks=["Hello ", "there"])]) + + result, events = run_turn(_service(model, FakeToolRuntime())) + + assert [e.delta for e in events_of_type(events, TextChunkEvent)] == ["Hello ", "there"] + assert events_of_type(events, AssistantMessageCompletedEvent) == [ + AssistantMessageCompletedEvent(content="Hello there")] + assert events_of_type(events, TurnCompletedEvent) == [ + TurnCompletedEvent(final_text="Hello there", steps_used=1)] + assert result.final_text == "Hello there" + assert result.ok is True + + +def test_the_composed_user_message_is_appended_before_the_first_call() -> None: + model = FakeModelCall([FakeReply(content="ok")]) + request = make_request(prompt="ship it", instruction_prefix="RULES", + session_notes="earlier: a.md", + messages=[{"role": "user", "content": "previous"}]) + + run_turn(_service(model, FakeToolRuntime()), request) + + sent = model.calls[0]["messages"] + assert sent[-1] == {"role": "user", + "content": "RULES\n\n---\n\nship it\n\nearlier: a.md"} + assert sent[-2] == {"role": "user", "content": "previous"} + + +def test_attachments_are_read_when_the_turn_runs_not_when_it_was_built() -> None: + # Extraction can pip-install a parser or shell out to LibreOffice, so it must + # happen here (worker thread), not while the UI was assembling the request. + seen: List[Tuple[str, Tuple[str, ...]]] = [] + + def reader(prompt: str, attachments: Tuple[str, ...]) -> str: + seen.append((prompt, attachments)) + return f"{prompt}\n\n" + + model = FakeModelCall([FakeReply(content="ok")]) + request = make_request(prompt="summarise", attachments=["a.docx", "b.pdf"]) + + run_turn(_service(model, FakeToolRuntime(), attachment_reader=reader), request) + + assert seen == [("summarise", ("a.docx", "b.pdf"))] + assert "contents of 2 file(s)" in model.calls[0]["messages"][-1]["content"] + + +def test_the_prompt_preparer_is_told_which_tools_the_turn_advertises() -> None: + # The system prompt gains an MS365 paragraph only when ms365__* tools are + # present, so the preparer has to see the real list. + seen: List[Tuple[str, ...]] = [] + model = FakeModelCall([FakeReply(content="ok")]) + tools = FakeToolRuntime(specs=("save_file", "ms365__send_mail")) + + run_turn(_service(model, tools, + prepare_prompt=lambda messages, names: seen.append(names))) + + assert seen == [("save_file", "ms365__send_mail")] + + +def test_only_the_allowed_tools_are_advertised() -> None: + model = FakeModelCall([FakeReply(content="ok")]) + tools = FakeToolRuntime(specs=("save_file", "run_command", "update_plan")) + + run_turn(_service(model, tools), make_request(allowed_tools=("save_file", "update_plan"))) + + assert model.calls[0]["tool_names"] == ["save_file", "update_plan"] + + +# --------------------------------------------------------------------------- # +# Tool dispatch. +# --------------------------------------------------------------------------- # +def tool_turn(tool_name: str = "save_file", args=None, **tool_kwargs): + """A turn that calls one tool, then answers.""" + calls = [{"id": "c1", "name": tool_name, "arguments": args or {"filename": "a.md"}}] + model = FakeModelCall([FakeReply(content="working", tool_calls=calls), + FakeReply(content="done")]) + return model, FakeToolRuntime(**tool_kwargs) + + +def test_a_tool_call_is_announced_executed_and_answered_in_the_message_list() -> None: + model, tools = tool_turn(results={"save_file": {"ok": True, "output": "saved", + "path": "out/a.md"}}) + + result, events = run_turn(_service(model, tools)) + + assert events_of_type(events, ToolCallStartedEvent) == [ToolCallStartedEvent( + call_id="c1", name="save_file", arguments={"filename": "a.md"}, + preview=ToolPreview(kind="info", title="save_file", text="{'filename': 'a.md'}"))] + assert events_of_type(events, ToolCallFinishedEvent) == [ToolCallFinishedEvent( + call_id="c1", name="save_file", ok=True, output="saved", path="out/a.md")] + assert tools.executed == [("save_file", {"filename": "a.md"})] + assert result.messages[-2] == {"role": "tool", "tool_call_id": "c1", + "name": "save_file", "content": "saved"} + + +def test_live_tool_output_is_streamed_while_the_tool_runs() -> None: + model, tools = tool_turn("run_command", {"command": "ls"}) + tools.emit_output = "file-a\n" + + _, events = run_turn(_service(model, tools)) + + assert events_of_type(events, ToolOutputChunkEvent) == [ToolOutputChunkEvent( + call_id="c1", name="run_command", delta="file-a\n")] + + +def test_the_loop_ends_as_soon_as_the_model_stops_calling_tools() -> None: + model, tools = tool_turn() + + result, _ = run_turn(_service(model, tools)) + + assert result.steps_used == 2 + assert result.budget_exhausted is False + + +def test_the_plan_tool_reports_a_plan_update_and_no_tool_bubble() -> None: + calls = [{"id": "c1", "name": "update_plan", + "arguments": {"steps": [{"title": "Draft", "status": "running"}]}}] + model = FakeModelCall([FakeReply(content="planning", tool_calls=calls), FakeReply(content="done")]) + tools = FakeToolRuntime(results={"update_plan": { + "ok": True, "output": "Plan updated.", + "plan_steps": [PlanStep(title="Draft", status="running")]}}) + + result, events = run_turn(_service(model, tools)) + + assert events_of_type(events, PlanUpdatedEvent) == [ + PlanUpdatedEvent(steps=(PlanStep(title="Draft", status="running"),))] + assert events_of_type(events, ToolCallStartedEvent) == [] + assert events_of_type(events, ToolCallFinishedEvent) == [] + assert result.plan_steps == (PlanStep(title="Draft", status="running"),) + + +# --------------------------------------------------------------------------- # +# Budget, cancellation. +# --------------------------------------------------------------------------- # +def test_running_out_of_steps_is_flagged_and_announced() -> None: + # The model keeps calling tools forever; the ceiling must stop it visibly. + forever = [FakeReply(content=f"step {i}", + tool_calls=[{"id": f"c{i}", "name": "save_file", "arguments": {}}]) + for i in range(5)] + model = FakeModelCall(forever) + + result, events = run_turn(_service(model, FakeToolRuntime()), make_request(max_steps=2)) + + assert result.steps_used == 2 + assert result.budget_exhausted is True + assert "2-step safety limit" in events_of_type(events, TextChunkEvent)[-1].delta + # The note reaches the transcript but NOT the stored answer: a turn that hits + # the ceiling always ends on a tool message, and the existing runtime only + # merges the note when the last message is the assistant's. Pinned here so a + # future change to that rule is a deliberate decision, not a silent drift. + assert result.final_text == "step 1" + + +def test_run_to_completion_uses_the_higher_ceiling() -> None: + forever = [FakeReply(content="x", tool_calls=[{"id": "c", "name": "save_file", "arguments": {}}]) + for _ in range(6)] + model = FakeModelCall(forever) + + result, _ = run_turn(_service(model, FakeToolRuntime()), + make_request(max_steps=2, completion_max_steps=5, run_to_completion=True)) + + assert result.steps_used == 5 + + +def test_a_turn_cancelled_before_it_starts_never_calls_the_model() -> None: + model = FakeModelCall([FakeReply(content="never")]) + + result, events = run_turn(_service(model, FakeToolRuntime()), cancel=lambda: True) + + assert model.calls == [] + assert result.cancelled is True + assert result.budget_exhausted is False + assert events_of_type(events, TurnCompletedEvent) == [TurnCompletedEvent(cancelled=True)] + + +def test_cancelling_during_a_turn_stops_dispatching_the_remaining_tool_calls() -> None: + calls = [{"id": "c1", "name": "save_file", "arguments": {}}, + {"id": "c2", "name": "save_file", "arguments": {}}] + model = FakeModelCall([FakeReply(content="two tools", tool_calls=calls)]) + tools = FakeToolRuntime() + stop = {"now": False} + + def cancel() -> bool: + return stop["now"] + + original_execute = tools.execute + + def execute(name, args, on_output=None, cancel=None): + stop["now"] = True # cancel raised while the first tool runs + return original_execute(name, args, on_output=on_output, cancel=cancel) + + tools.execute = execute + + result, _ = run_turn(_service(model, tools), cancel=cancel) + + assert len(tools.executed) == 1 + assert result.cancelled is True + + +# --------------------------------------------------------------------------- # +# Bring-your-own working list. +# +# ``ui/chat_panel.py`` holds the turn's message list in its own turn context and +# reads it WHILE the worker appends (``_reattach_running_turn`` replays the steps +# done so far when the user reopens a running conversation; ``_finalize_turn`` +# slices it by ``snapshot_len``). A service that built its own private list would +# silently break both, so a caller can hand its list over instead. +# --------------------------------------------------------------------------- # +def test_a_caller_supplied_list_is_appended_to_in_place() -> None: + model, tools = tool_turn() + live: List[Dict[str, Any]] = [{"role": "user", "content": "already composed"}] + + result = ConversationApplicationService(model, tools).execute( + make_request(), lambda event: None, messages=live) + + roles = [m["role"] for m in live] + assert roles == ["user", "assistant", "tool", "assistant"] + assert result.messages == tuple(live) + + +def test_a_caller_supplied_list_is_used_as_is_without_recomposing_the_prompt() -> None: + # The widget already applied the skill prefix and the session notes when it + # built its message; composing again would duplicate them. + model = FakeModelCall([FakeReply(content="ok")]) + user = {"role": "user", "content": "already composed"} + live = [user] + + ConversationApplicationService(model, FakeToolRuntime()).execute( + make_request(prompt="typed text", instruction_prefix="RULES", + session_notes="notes"), + lambda event: None, messages=live) + + assert live[0] is user + assert live[0]["content"] == "already composed" + assert [m["role"] for m in live].count("user") == 1 + + +def test_a_caller_supplied_list_skips_the_attachment_reader() -> None: + # Reading the attachments is what produced the caller's message in the first + # place; doing it again would re-parse every file. + model = FakeModelCall([FakeReply(content="ok")]) + calls: List[Any] = [] + + ConversationApplicationService( + model, FakeToolRuntime(), + attachment_reader=lambda prompt, attachments: calls.append(prompt) or prompt, + ).execute(make_request(attachments=["a.docx"]), lambda event: None, + messages=[{"role": "user", "content": "composed"}]) + + assert calls == [] diff --git a/tests/unit/test_conversation_turn_guards.py b/tests/unit/test_conversation_turn_guards.py new file mode 100644 index 0000000..dcdb062 --- /dev/null +++ b/tests/unit/test_conversation_turn_guards.py @@ -0,0 +1,210 @@ +"""R04-T03 (b) — the turn loop: guards, permission gate, compaction, cleanup. + +Split out of ``test_conversation_application_service.py`` to keep each file +inside the 400-LOC limit. Same fakes, same service; this half pins the ORDER of +the safety steps (guard before model, guard before execute, gate before execute) +and the promise that the output sandbox is tidied on the way out. +""" + +from __future__ import annotations + +from typing import Any, Dict, List + +import pytest +from cowork_local.application.conversations.conversation_application_service import ( + ConversationApplicationService, +) +from cowork_local.domain.agents.agent_event import ( + ErrorEvent, + OutputsAddedEvent, + ReasoningChunkEvent, + TextChunkEvent, + ToolCallFinishedEvent, +) +from cowork_local.tests.fakes.turn_runtime_fakes import ( + FakeModelCall, + FakeReply, + FakeToolRuntime, + events_of_type, + make_request, + run_turn, + tool_turn, +) + + +def _service(model, tools, **overrides) -> ConversationApplicationService: + return ConversationApplicationService(model, tools, **overrides) + + +# --------------------------------------------------------------------------- # +# Guards and the permission gate. +# --------------------------------------------------------------------------- # +def test_the_prompt_guard_runs_before_the_model_is_ever_called() -> None: + order: List[str] = [] + model = FakeModelCall([FakeReply(content="ok")]) + model_call = model.call + + def call(*a, **kw): + order.append("model") + return model_call(*a, **kw) + + model.call = call + + run_turn(_service(model, FakeToolRuntime(), prompt_guard=lambda messages: order.append("guard"))) + + assert order == ["guard", "model"] + + +def test_a_blocked_prompt_propagates_before_the_output_folder_is_touched() -> None: + model = FakeModelCall([FakeReply(content="never")]) + tools = FakeToolRuntime() + events: List[Any] = [] + + def guard(messages) -> None: + raise RuntimeError("SecurityBlocked: nope") + + service = _service(model, tools, prompt_guard=guard) + + with pytest.raises(RuntimeError, match="SecurityBlocked"): + service.execute(make_request(), events.append) + + assert model.calls == [] + assert events_of_type(events, ErrorEvent) == [ErrorEvent(message="SecurityBlocked: nope")] + # Cleanup is NOT a read-only operation (it deletes a stale .scratch and every + # empty sub-folder), so a turn rejected before it started must not run it. + assert tools.finalize_calls == [] + + +def test_output_cleanup_still_runs_when_the_turn_fails_mid_loop() -> None: + # Once the turn has started producing files, the sandbox must be tidied on + # the way out no matter how the turn ends. + model = FakeModelCall([RuntimeError("gateway exploded")]) + tools = FakeToolRuntime() + events: List[Any] = [] + + with pytest.raises(RuntimeError, match="gateway exploded"): + _service(model, tools).execute(make_request(), events.append) + + assert tools.finalize_calls == [{"before": "before", "cancelled": False}] + assert events_of_type(events, ErrorEvent) == [ErrorEvent(message="gateway exploded")] + + +def test_the_command_guard_runs_before_the_tool_executes() -> None: + order: List[str] = [] + model, tools = tool_turn("run_command", {"command": "ls"}) + original = tools.execute + + def execute(name, args, on_output=None, cancel=None): + order.append("execute") + return original(name, args, on_output=on_output, cancel=cancel) + + tools.execute = execute + + run_turn(_service(model, tools, + command_guard=lambda name, args: order.append(f"guard:{name}"))) + + assert order == ["guard:run_command", "execute"] + + +def test_disabling_rule_enforcement_skips_both_guards() -> None: + # Co4E flow steps run inside the workspace sandbox and opt out on purpose. + calls: List[str] = [] + model, tools = tool_turn("run_command", {"command": "ls"}) + + run_turn(_service(model, tools, + prompt_guard=lambda messages: calls.append("prompt"), + command_guard=lambda name, args: calls.append("command")), + make_request(enforce_rules=False)) + + assert calls == [] + + +def test_the_permission_gate_is_asked_only_for_command_tools() -> None: + asked: List[str] = [] + model, tools = tool_turn("save_file", {"filename": "a.md"}) + + run_turn(_service(model, tools, + permission_request=lambda action: asked.append(action["name"]) or True), + make_request(gate_mode="confirm")) + + assert asked == [] # save_file writes into the sandbox: never gated + + +def test_a_command_tool_in_confirm_mode_asks_before_running() -> None: + asked: List[Dict[str, Any]] = [] + model, tools = tool_turn("run_command", {"command": "ls"}) + + def approve(action: Dict[str, Any]) -> bool: + asked.append(action) + return True + + run_turn(_service(model, tools, permission_request=approve), make_request(gate_mode="confirm")) + + assert [a["name"] for a in asked] == ["run_command"] + assert tools.executed == [("run_command", {"command": "ls"})] + + +def test_a_rejected_command_is_reported_as_a_failed_tool_and_never_runs() -> None: + model, tools = tool_turn("run_command", {"command": "rm -rf /"}) + + result, events = run_turn(_service(model, tools, permission_request=lambda action: False), + make_request(gate_mode="confirm")) + + assert tools.executed == [] + assert events_of_type(events, ToolCallFinishedEvent) == [ToolCallFinishedEvent( + call_id="c1", name="run_command", ok=False, output="Rejected by user.")] + assert result.messages[-2]["content"] == "Rejected by user." + + +def test_auto_mode_never_asks_even_for_a_command() -> None: + model, tools = tool_turn("run_command", {"command": "ls"}) + + def refuse(action): # would block the turn if it were consulted + raise AssertionError("the gate must not be consulted in auto mode") + + run_turn(_service(model, tools, permission_request=refuse), make_request(gate_mode="auto")) + + assert tools.executed == [("run_command", {"command": "ls"})] + + +# --------------------------------------------------------------------------- # +# Context compaction, reasoning, output cleanup. +# --------------------------------------------------------------------------- # +def test_the_conversation_is_offered_for_compaction_before_every_call() -> None: + compactions: List[int] = [] + model, tools = tool_turn() + + run_turn(_service(model, tools, + compact=lambda messages, cancel: compactions.append(len(messages)))) + + assert len(compactions) == 2 # once per provider call + + +def test_reasoning_is_streamed_as_its_own_event() -> None: + model = FakeModelCall([FakeReply(content="42", reasoning="thinking...")]) + + _, events = run_turn(_service(model, FakeToolRuntime())) + + assert events_of_type(events, ReasoningChunkEvent) == [ReasoningChunkEvent(delta="thinking...")] + + +def test_a_reasoning_only_reply_gets_a_visible_note_in_the_transcript() -> None: + # Otherwise a Schedule Task run reads back an empty answer and writes + # "(no output)" into its report. + model = FakeModelCall([FakeReply(content="", reasoning="thought hard")]) + + result, events = run_turn(_service(model, FakeToolRuntime())) + + assert "only its reasoning" in events_of_type(events, TextChunkEvent)[-1].delta + assert "only its reasoning" in result.final_text + + +def test_promoted_and_discarded_output_files_are_reported_at_the_end() -> None: + model = FakeModelCall([FakeReply(content="ok")]) + tools = FakeToolRuntime(added=("out/report.pptx",)) + + _, events = run_turn(_service(model, tools)) + + assert events_of_type(events, OutputsAddedEvent) == [ + OutputsAddedEvent(paths=("out/report.pptx",))] + assert tools.finalize_calls == [{"before": "before", "cancelled": False}] diff --git a/tests/unit/test_cowork_turn_request.py b/tests/unit/test_cowork_turn_request.py new file mode 100644 index 0000000..6a5d931 --- /dev/null +++ b/tests/unit/test_cowork_turn_request.py @@ -0,0 +1,76 @@ +"""R04-T04 — unit tests for the UI-state -> request mapping. + +Three small rules used to sit inline in ``ui/cowork_tab.py::build_job``, where no +test could reach them: the turn's prompt is the last message in the working list, +the history is everything before it, and the confirm-commands flag becomes a gate +mode. Getting any of them wrong is silent (a duplicated user message, a command +that stops asking for approval), so they are pinned here. +""" + +from __future__ import annotations + +from pathlib import Path + +from cowork_local.application.conversations.cowork_turn_request import ( + build_cowork_turn_request, +) + + +def _build(**overrides): + base = { + "turn_id": "t3", + "session_id": "s1", + "messages": [{"role": "user", "content": "make me a report"}], + } + base.update(overrides) + return build_cowork_turn_request(**base) + + +def test_the_last_message_becomes_the_prompt_and_the_rest_the_history() -> None: + request = _build(messages=[ + {"role": "user", "content": "earlier"}, + {"role": "assistant", "content": "sure"}, + {"role": "user", "content": "now this"}, + ]) + + assert request.prompt == "now this" + assert request.messages == ({"role": "user", "content": "earlier"}, + {"role": "assistant", "content": "sure"}) + + +def test_an_empty_working_list_yields_an_empty_prompt() -> None: + # Defensive: a turn with no message at all must not raise on messages[-1]. + request = _build(messages=[]) + + assert request.prompt == "" + assert request.messages == () + + +def test_confirming_commands_puts_the_turn_in_confirm_gate_mode() -> None: + assert _build(confirm_commands=True).gate_mode == "confirm" + assert _build(confirm_commands=False).gate_mode == "auto" + assert _build().gate_mode == "auto" # auto-run is the default + + +def test_the_captured_widget_state_is_carried_into_the_request() -> None: + request = _build( + surface="cowork", project_id="p7", title="Weekly report", + provider_id="anthropic", model="claude-sonnet-4-6", + instructions="PROJECT RULES", output_dir="out/.turns/t3", + home_output_root="out", agent_role="cowork", + ) + + assert (request.turn_id, request.session_id) == ("t3", "s1") + assert (request.surface, request.project_id, request.title) == \ + ("cowork", "p7", "Weekly report") + assert (request.provider_id, request.model) == ("anthropic", "claude-sonnet-4-6") + assert request.project_context == "PROJECT RULES" + assert request.output_dir == Path("out/.turns/t3") + assert request.home_output_root == Path("out") + assert request.agent_role == "cowork" + + +def test_the_prompt_survives_a_message_whose_content_is_missing() -> None: + request = _build(messages=[{"role": "user"}]) + + assert request.prompt == "" diff --git a/tests/unit/test_task_prompt_assembly.py b/tests/unit/test_task_prompt_assembly.py new file mode 100644 index 0000000..9f260d0 --- /dev/null +++ b/tests/unit/test_task_prompt_assembly.py @@ -0,0 +1,34 @@ +"""R04-T05 — unit tests for the unattended-run prompt assembly. + +``_run_agent`` used to build this by rebinding ``prompt`` three times, each with +its own ``f"{block}\n\n{prompt}"``. The ORDER that produced is load-bearing (the +plan reminder has to lead, the task's own words have to trail) and it was +readable only by replaying the rebindings in your head. +""" + +from __future__ import annotations + +from cowork_local.core.task_executors import _unattended_prompt + + +def test_the_plan_reminder_leads_and_the_task_prompt_trails() -> None: + built = _unattended_prompt("write the report") + + assert built.startswith("This runs unattended (Schedule Task)") + assert built.endswith("write the report") + + +def test_a_skill_block_sits_between_the_reminder_and_the_agent_persona() -> None: + built = _unattended_prompt("write the report", skill_text="SKILL", + agent_instructions="PERSONA") + + assert built.index("This runs unattended") < built.index("SKILL") + assert built.index("SKILL") < built.index("PERSONA") + assert built.index("PERSONA") < built.index("write the report") + + +def test_absent_blocks_leave_no_extra_blank_lines() -> None: + built = _unattended_prompt("do it", skill_text="", agent_instructions=None) + + assert "\n\n\n" not in built + assert built.count("do it") == 1 diff --git a/tests/unit/test_turn_runtime.py b/tests/unit/test_turn_runtime.py new file mode 100644 index 0000000..4737618 --- /dev/null +++ b/tests/unit/test_turn_runtime.py @@ -0,0 +1,36 @@ +"""R04-T04 — unit tests for the shared turn-runtime helpers. + +``combine_instructions`` is the small rule the UI applied inline: a turn's +standing instructions are several independent blocks (project context, an Admin +agent's persona, a skill's rules, an unattended-run reminder) that must be joined +with one blank line, skipping whatever is absent. Two call sites need it (T04's +widget and T05's task runner), which is exactly when a rule stops being an inline +expression. +""" + +from __future__ import annotations + +from cowork_local.application.conversations.turn_runtime import combine_instructions + + +def test_two_blocks_are_joined_by_a_blank_line() -> None: + assert combine_instructions("PROJECT", "AGENT") == "PROJECT\n\nAGENT" + + +def test_an_absent_block_leaves_no_blank_line_behind() -> None: + assert combine_instructions("", "AGENT") == "AGENT" + assert combine_instructions("PROJECT", "") == "PROJECT" + assert combine_instructions("PROJECT", None) == "PROJECT" + + +def test_whitespace_only_blocks_do_not_count_as_instructions() -> None: + assert combine_instructions(" \n ", "AGENT") == "AGENT" + + +def test_nothing_to_say_produces_an_empty_string() -> None: + assert combine_instructions() == "" + assert combine_instructions("", None, " ") == "" + + +def test_more_than_two_blocks_keep_their_order() -> None: + assert combine_instructions("A", "B", "C") == "A\n\nB\n\nC" diff --git a/ui/cowork_tab.py b/ui/cowork_tab.py index 44c620f..ad47cee 100644 --- a/ui/cowork_tab.py +++ b/ui/cowork_tab.py @@ -346,19 +346,45 @@ 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). + """This turn's job: a frozen request run through the conversation service. + + Since R04-T04 the widget no longer drives the turn loop. Every value a + turn depends on is read HERE, on the UI thread at submit time, and packed + into an immutable ``ConversationExecutionRequest`` — so clicking a + different model or switching workspace mid-answer cannot reach work + already in flight. + """ output_dir = out_dir or self._session_output_dir() + # The sandbox folder is named by the turn id ('.turns/t3'); with no + # sandbox the session id identifies the turn well enough for the audit log. + turn_id = out_dir.name if out_dir is not None else self.session_id + session_id = self.session_id title = self.title project_id = self.project_id + home_output_root = self.workspace_dir() # Captured at submit time (UI thread): the Admin-defined agent # preset's instructions, if one is selected in the Agent picker. agent_prompt = self.admin_agent_prompt() + # Per-workspace Auto-run override wins, else the global "confirm before + # running commands" setting. Frozen now, so a Settings change mid-turn + # cannot flip the rules this turn started under. + confirm_commands = self.ctx.project_confirm_commands() + # What the turn is recorded as running on. A routing override (R03) wins + # over the tab's own picker; '' means the provider's configured default. + # Informational only — an Admin-agent preset builds its own provider + # below, so treat these as the record, not the decision. + provider_id = self._routed_provider or self.ctx.config.active_provider + model = self._routed_model or self._model or "" def job(worker: AgentWorker): - from ..core.chat_agent import run_cowork + from ..application.conversations.core_runtime_adapter import ( + build_cowork_conversation_service, + legacy_event_sink, + ) + from ..application.conversations.cowork_turn_request import ( + build_cowork_turn_request, + ) + from ..application.conversations.turn_runtime import combine_instructions from ..core.projects import load_project, project_context_text provider = self.build_provider() # this tab's selected agent/model @@ -367,23 +393,36 @@ class CoworkTab(ChatPanel): # 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 + # Shared project instructions (Claude-Projects style) plus the Admin + # agent's persona, refreshed each turn so edits in the Workspace + # screen apply immediately. + instructions = combine_instructions( + project_context_text(load_project(project_id)), 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). + # matches the pre-existing auto-run behavior. The gate lives on the + # worker because the UI resolves it from the main thread. gate = None - if self.ctx.project_confirm_commands(): + if 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) + + service = build_cowork_conversation_service( + provider, output_dir, worker.emit_event, title=title, + project_context=instructions, extra_tools=extra_tools, + extra_executor=extra_exec, security_config=self.ctx.config, + gate=gate, agent_role=agent_roles.COWORK, + ) + request = build_cowork_turn_request( + turn_id=turn_id, session_id=session_id, surface=self.kind, + project_id=project_id, title=title, messages=messages, + provider_id=provider_id, model=model, instructions=instructions, + output_dir=output_dir, home_output_root=home_output_root, + confirm_commands=gate is not None, agent_role=agent_roles.COWORK, + ) + # Hand the widget's own list over: _reattach_running_turn replays + # from it while the turn is still running, and _finalize_turn slices + # it afterwards, so the service must append into that very object. + service.execute(request, legacy_event_sink(worker.emit_event), + cancel=worker.is_cancelled, messages=messages) return {"messages": messages, "turn_dir": str(output_dir)} return job