cowork-local/tests/fakes/turn_runtime_fakes.py

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