amphibious¶

class Context(BaseModel):
    """
    Base class for agent context — fields + an overridable ``summary``.

    Collapsed (per the small-loop redesign) to its essentials:

    * :meth:`_raw_fields` — the most primitive view: every field's raw value
      as ``{name: value}`` (no rendering, no filtering).
    * :meth:`summary` — the **overridable** method the framework injects with
      that raw dict. The default returns the dict unchanged; an override is
      handed the ``fields`` dict and composes whatever it wants (usually a
      ``str``) — without fetching anything itself.

    A bare ``Context`` is free-form cross-turn state (the big-loop half):
    declare fields and override ``summary``, and that is all. Tools are **not**
    a base-context concern — they belong to the OTA loop that actually acts, so
    the tool registry (``tools`` field + :meth:`~OTAContext.tool` /
    :meth:`~OTAContext.add_tool`) lives on :class:`OTAContext`.

    Examples
    --------
    >>> class MyContext(Context):
    ...     goal: str = ""
    ...     def summary(self, fields):   # ``fields`` is auto-injected (the raw dict)
    ...         return f"Goal: {fields['goal']}"
    """
    model_config = ConfigDict(arbitrary_types_allowed=True)

    ############################################################################
    # Initialize Context
    ############################################################################

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)

        # A subclass that overrides ``summary`` uses ``fields`` directly.
        if "summary" in cls.__dict__:
            cls.summary = cls._summary_injecting(cls.__dict__["summary"])

    @staticmethod
    def _summary_injecting(user_summary: Callable) -> Callable:
        """
        Wrap an overridden ``summary`` so its ``fields`` argument is always
        the raw per-field dict (built via :meth:`_raw_fields` when omitted).
        """
        @functools.wraps(user_summary)
        def _wrapped(self, fields: Optional[Dict[str, Any]] = None) -> Any:
            if fields is None:
                fields = self._raw_fields()
            return user_summary(self, fields)
        return _wrapped

    def model_post_init(self, __context: Any) -> None:
        super().model_post_init(__context)

        # Call __post_init__ if defined in subclass (provides dataclass-like API)
        if hasattr(self, '__post_init__') and callable(getattr(self, '__post_init__')):
            self.__post_init__()

    ############################################################################
    # Core Methods
    ############################################################################

    def _raw_fields(self) -> Dict[str, Any]:
        """Every model field's raw value, unprocessed: ``{name: value}``.

        No rendering, no per-field summarising, no filtering — the most
        primitive view of the context. This is the dict the framework hands
        to :meth:`summary`.
        """
        return {name: getattr(self, name) for name in type(self).model_fields}

    def summary(self, fields: Optional[Dict[str, Any]] = None) -> Any:
        """Assemble this context for the prompt (the **overridable** method).

        An overridden ``summary`` is auto-wrapped (see :meth:`__init_subclass__`)
        so ``fields`` is always the raw per-field dict (:meth:`_raw_fields`) — the
        override just uses it (``fields.get(...)``) and composes whatever it wants
        (typically a ``str``), with zero manual fetch. The default (no override)
        returns the raw dict unchanged.

        Parameters
        ----------
        fields : Optional[Dict[str, Any]]
            The raw per-field dict, auto-injected into an overridden ``summary``.

        Returns
        -------
        Any
            The raw dict by default, or whatever an override composes.
        """
        return fields if fields is not None else self._raw_fields()

    def __iter__(self) -> Iterator[Tuple[str, Any]]:
        """Yield ``(field_name, field_value)`` for every model field on this context."""
        for field_name in type(self).model_fields:
            yield field_name, getattr(self, field_name)

    def __str__(self) -> str:
        """Return a formatted, human-readable view of every field."""
        lines = [f"{'=' * 50}", f"  {self.__class__.__name__}", f"{'=' * 50}"]
        for field_name in self.__class__.model_fields:
            value = getattr(self, field_name)
            if value is not None:
                lines.append(f"\n[{field_name}]")
                lines.append(f"  {value}")
        lines.append(f"\n{'=' * 50}")
        return "\n".join(lines)

    def __repr__(self) -> str:
        """Return a concise representation of the context."""
        parts = []
        for field_name in self.__class__.model_fields:
            value = getattr(self, field_name)
            if value is not None:
                parts.append(f"{field_name}={value!r}")
        return f"{self.__class__.__name__}({', '.join(parts)})"

class OTAContext(Context):
    """Small-loop working context: one run's input + its OTA round trace + tools.

    The **framework-owned** half of the two-loop model. During a run the
    automa drives it directly through the per-round result accessors
    (``ota_ctx.obs_result = ...`` / ``.think_result`` / ``.action_result``),
    :meth:`open_record`, and :meth:`add_tool`. Each round is one
    :class:`OTARecord` (observe/think/action results, ``extra="allow"`` so a
    ``before_action`` / ``after_action`` hook can fold custom per-round fields
    like a ``permission_result`` via :meth:`_current_record`).

    Its ``tools`` are **declared on the class** via :meth:`tool` — the registry
    lives here, not on the base :class:`Context`, because tools are an OTA-loop
    concern. The framework no longer merges any tools in; whatever the context
    declares is what the small loop carries.

    Attributes
    ----------
    user_input : Any
        This run's question / objective. Any payload — a plain ``str``, or a
        structured input the agent's own hooks (e.g. ``observation``) parse.
        Framework built-ins only ``str()`` it for default rendering / tracing.
    ota_record : List[OTARecord]
        The observe-think-act round trace (one :class:`OTARecord` per round).

    Examples
    --------
    >>> ota = OTAContext(user_input="Find the bug")
    >>> ota.open_record()                  # framework brackets each OTA cycle
    >>> ota.obs_result = "saw a stack trace"
    >>> ota.think_result = decision
    >>> ota.action_result = tool_output
    """
    _declared_tools: ClassVar[List[ToolSpec]] = []  # Tools declared on this class via ``tool`` (per-subclass, inherits bases).

    user_input: Any = Field(
        default="",
        description=(
            "This run's question / objective. Any payload — the agent's hooks "
            "interpret it; framework built-ins only ``str()`` it (default render)."
        ),
    )
    ota_record: List[OTARecord] = Field(
        default_factory=list,
        description="Observe-think-act round trace (one OTARecord per round)",
    )
    tools: List[ToolSpec] = Field(
        default_factory=list,
        description="Action-phase tool affordances carried by this OTA run",
    )

    ############################################################################
    # Tool registry (action-phase affordances the OTA loop carries)
    ############################################################################
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)

        # A subclass inherits its bases' declared tools and can add more via its own ``tool`` calls.
        seeded: List[ToolSpec] = []
        seen: set = set()
        for base in cls.__bases__:
            for spec in getattr(base, "_declared_tools", []):
                if spec.tool_name in seen:
                    continue
                seen.add(spec.tool_name)
                seeded.append(spec)
        cls._declared_tools = seeded

    @classmethod
    def tool(cls, obj):
        """Declare a tool on this OTA context — usable as a decorator **and** a call.

        Every context now declares the tools it carries; nothing is
        auto-injected by the framework. ``obj`` is normalized by
        :func:`_to_tool_spec` and appended to this class's
        :attr:`_declared_tools`. The original ``obj`` is returned so this works
        transparently as a decorator.

        * ``@MyOTACtx.tool`` on a standalone ``def f(...)`` — registers ``f``,
          returns ``f``.
        * ``MyOTACtx.tool(bash_tool)`` — registers an existing :class:`ToolSpec`.
        * ``MyOTACtx.tool(obj.method)`` — registers a bound method, keeping
          ``obj`` as its ``self`` (see :func:`_to_tool_spec`).

        Parameters
        ----------
        obj : Callable | ToolSpec
            A plain callable, a bound method, or an existing tool spec.

        Returns
        -------
        Callable | ToolSpec
            ``obj`` unchanged, so this may be used as a decorator.
        """
        cls._declared_tools.append(_to_tool_spec(obj))
        return obj

    def add_tool(self, tool: ToolSpec) -> None:
        """Register a tool into this run's action-phase toolset."""
        self.tools.append(tool)

    def model_post_init(self, __context: Any) -> None:
        # Seed this run's toolset from the class's declared tools before the
        # base hook fires, so a subclass ``__post_init__`` can rely on it. An
        # explicit ``tools=`` (e.g. a narrowed delegation set) is preserved.
        if not self.tools:
            self.tools = list(type(self)._declared_tools)
        super().model_post_init(__context)

    ############################################################################
    # Round lifecycle + per-round result accessors
    ############################################################################
    def _current_record(self) -> OTARecord:
        """The in-flight (latest) round; opens one lazily if none exists yet.

        This is the record the result accessors write to, and the fold-point
        hooks attach custom fields onto (e.g.
        ``ota_ctx._current_record().permission_result = verdict``).
        """
        if not self.ota_record:
            self.open_record()
        return self.ota_record[-1]

    def open_record(self) -> None:
        """Explicitly open a new round (append a new record to the trace)."""
        self.ota_record.append(OTARecord())

    @property
    def obs_result(self) -> Any:
        return self.ota_record[-1].observation_result if self.ota_record else None

    @obs_result.setter
    def obs_result(self, value: Any) -> None:
        self._current_record().observation_result = value

    @property
    def think_result(self) -> Any:
        return self.ota_record[-1].think_result if self.ota_record else None

    @think_result.setter
    def think_result(self, value: Any) -> None:
        self._current_record().think_result = value

    @property
    def action_result(self) -> Any:
        return self.ota_record[-1].action_result if self.ota_record else None

    @action_result.setter
    def action_result(self, value: Any) -> None:
        self._current_record().action_result = value

    ############################################################################
    # Prompt rendering (overridable)
    ############################################################################
    def summary(self, fields: Optional[Dict[str, Any]] = None) -> str:
        """Render this run's small-loop state for the prompt.

        Default: the user input + the OTA round trace. ``fields`` is the
        auto-injected raw dict (available to an override that prefers it);
        subclass and override to customise how the run is summarised.

        Returns
        -------
        str
            A prompt-facing rendering of the input + round trace.
        """
        parts: List[str] = [f"User input: {self.user_input}"]
        for i, record in enumerate(self.ota_record):
            parts.append(f"[Round {i}]")
            if record.observation_result is not None:
                parts.append(f"  Observation: {record.observation_result}")
            if record.think_result is not None:
                parts.append(f"  Think: {record.think_result}")
            if record.action_result is not None:
                parts.append(f"  Action: {record.action_result}")
            for key, value in (getattr(record, "model_extra", None) or {}).items():
                parts.append(f"  {key}: {value}")
        return "\n".join(parts)

class CognitiveWorker(GraphAutoma):
    """Cognitive worker — one observe-think-act cycle.

    Observation and action execution are handled by ``AmphibiousAutoma`` as
    shared infrastructure. The worker owns one thing: the **thinking** step.

    :meth:`thinking` is abstract — subclass and implement it to assemble the
    prompt from the two contexts and call ``self._llm`` (chat / tool-select /
    structured output); whatever you return, :meth:`_assemble_decision`
    adapts into the framework's decision. Optional hooks ``observation`` /
    ``before_action`` / ``after_action`` refine the cycle further.

    >>> class MyThink(CognitiveWorker):
    ...     async def thinking(self, ota_context, context=None):
    ...         return await self._llm.aselect_tool(
    ...             messages=build_messages(ota_context),
    ...             tools=[t.to_tool() for t in ota_context.tools],
    ...         )
    """

    def __init__(
        self,
        llm: Optional[BaseLlm] = None,
        verbose: Optional[bool] = None,
    ):
        super().__init__()

        # LLM
        self._llm = llm

        # Log
        self._verbose = verbose

        # Usage stats
        self.spent_tokens = 0
        self.spent_time = 0

    ############################################################################
    # Core methods
    ############################################################################

    @worker(is_start=True, is_output=True)
    async def _thinking(self, ota_context: Optional[OTAContext] = None, context: Optional[Context] = None) -> Any:
        """Framework entry for the thinking phase — orchestrates :meth:`thinking`.

        Both contexts are injected by the dispatcher
        (``arun(ota_context=…, context=…)``): ``ota_context`` is the
        small-loop OTA state, ``context`` the free-form knowledge (``None``
        for a pure-reasoning run). Validates the LLM, calls the overridable
        :meth:`thinking` method to interact with the LLM, then *adapts* its
        result — ``(tool_calls, content)``, a structured ``BaseModel``, or
        text — into the framework's decision shape (see
        :meth:`_assemble_decision`). That decision becomes the ``arun()``
        return value — the driver captures it and runs the act phase.

        The seam: ``thinking`` owns *talking to the model*, the framework
        owns *turning the reply into a decision*.
        """
        if self._llm is None:
            raise RuntimeError(
                "CognitiveWorker has no LLM set — pass llm= when constructing "
                "the worker."
            )
        if ota_context is None:
            ota_context = OTAContext()

        result = await self.thinking(ota_context, context)
        return self._assemble_decision(result)

    def _assemble_decision(self, result: Any) -> ThinkResult:
        """Adapt whatever :meth:`thinking` returned into the act phase's decision.

        The single seam between *any* bridgic LLM protocol and
        ``AmphibiousAutoma``'s dispatch: ``thinking`` calls the model however
        it likes and hands back that call's **natural** result; this maps each
        protocol's real return shape (see ``bridgic.llms.*``) onto a decision:

        Every shape collapses to a flat ``ThinkResult`` (``step_content`` +
        ``tool_calls``); a result with NO ``tool_calls`` IS the finish:

        - ``Response``                 [``achat``]
          -> content-only (``step_content`` = the reply text).
        - ``(tool_calls, content)``    [``aselect_tool``]
          -> tool-calling (``step_content`` = content). NOTE the order —
          ``aselect_tool`` returns ``tool_calls`` first.
        - a pydantic ``BaseModel`` / ``dict``  [``astructured_output``]
          -> content-only; the structured value is serialized into
          ``step_content`` (JSON).
        - a ``str``                    [plain text / an accumulated stream]
          -> content-only.

        Tool-call items may be ``ToolCall`` objects (``.name`` / ``.arguments``)
        or ``{"name": ..., "arguments": {...}}`` dicts.
        """
        # chat — a Response (text in .message.content). Checked before
        # BaseModel because Response *is* a BaseModel.
        if isinstance(result, Response):
            return ThinkResult(step_content=result.message.content or "", tool_calls=[])

        # structured output — a pydantic model or a json-schema dict; the
        # typed value is serialized into ``step_content`` (which is text).
        if isinstance(result, BaseModel):
            return ThinkResult(step_content=result.model_dump_json(), tool_calls=[])
        if isinstance(result, dict):
            return ThinkResult(
                step_content=json.dumps(result, ensure_ascii=False, default=str),
                tool_calls=[],
            )

        # plain text / an accumulated stream.
        if isinstance(result, str):
            return ThinkResult(step_content=result, tool_calls=[])

        # tool-select — (tool_calls, content), aselect_tool's native order.
        if not isinstance(result, (tuple, list)):
            raise TypeError(
                f"thinking() returned an unsupported type {type(result).__name__}; "
                "return a Response, (tool_calls, content), a pydantic BaseModel, "
                "a dict, or str."
            )
        tool_calls, content = result
        tool_calls = tool_calls or []
        tool_calls = [
            StepToolCall(
                call_id=self._tool_call_id(call),
                tool=self._tool_call_name(call),
                tool_arguments=[
                    ToolArgument(name=str(name), value=value)
                    for name, value in self._tool_call_args(call).items()
                ],
            )
            for call in tool_calls
        ]
        return ThinkResult(step_content=content or "", tool_calls=tool_calls)

    ############################################################################
    # Internal helpers
    ############################################################################

    @staticmethod
    def _tool_call_id(call: Any) -> Optional[str]:
        """Read a tool call id from common provider/adapter shapes."""
        if isinstance(call, dict):
            for key in ("id", "call_id", "tool_call_id"):
                value = call.get(key)
                if value:
                    return str(value)
            return None

        for attr in ("id", "call_id", "tool_call_id"):
            value = getattr(call, attr, None)
            if value:
                return str(value)
        return None

    @staticmethod
    def _tool_call_name(call: Any) -> str:
        """Read a tool call's name — accepts an object (``.name``) or a dict."""
        return call["name"] if isinstance(call, dict) else call.name

    @staticmethod
    def _tool_call_args(call: Any) -> Dict[str, Any]:
        """Read a tool call's arguments — accepts an object (``.arguments``) or a dict."""
        args = call.get("arguments") if isinstance(call, dict) else call.arguments
        return args or {}

    def _clone(self) -> "CognitiveWorker":
        """Return a fresh worker with the same configuration.

        The ``BaseLlm`` is *shared* across clones — it is stateless per
        call — so the clone is built with ``llm=None`` and the agent sets
        the LLM at runtime. Only config (verbose) is carried over; runtime
        state (tokens, time, GraphAutoma execution state) starts clean.
        Subclasses with extra ``__init__`` params should override.

        Used by ``ThinkUnitDescriptor._clone_worker`` for state isolation
        at every ``yield ThinkUnit(...)``.
        """
        return type(self)(
            llm=None,
            verbose=self._verbose,
        )

    ############################################################################
    # Template Methods
    ############################################################################

    async def observation(self, ota_context: OTAContext, context: Optional[Context] = None) -> Any:
        """Worker-level observation hook. Override to customize.

        Both forms accepted: coroutine (``return _DELEGATE`` /
        ``return value``) or async-generator (yield side-effect calls,
        then ``yield RETURN(value)``).

        Returning ``_DELEGATE`` (or ``None``) hands off to
        ``AmphibiousAutoma.observation()``. Other values become the
        observation directly.

        >>> async def observation(self, ota_context):
        ...     return f"Current state: {ota_context.user_input}"
        """
        return _DELEGATE

    async def thinking(self, ota_context: OTAContext, context: Optional[Context] = None) -> Any:
        """Assemble context, call the model, return its result. **Override this.**

        The worker's one job and override point: turn the two contexts into a
        prompt and call ``self._llm`` however the model needs, then return that
        call's **natural** result — :meth:`_assemble_decision` adapts it. Just
        return what the bridgic protocol hands you:

        - ``achat`` -> a ``Response`` (content-only, finished).
        - ``aselect_tool`` -> ``(tool_calls, content)``.
        - ``astructured_output`` -> a pydantic ``BaseModel`` or a ``dict``.
        - ``astream`` -> consume the ``MessageChunk`` deltas yourself (forward
          them to a live callback if you want), then return the accumulated
          ``str`` (or a ``Response``).

        Two-loop inputs: ``ota_context`` is the small-loop OTA context (its
        ``user_input`` + ``ota_record`` round trace form the task, its
        ``tools`` list the action affordances); ``context`` is the free-form
        knowledge context (``None`` for a pure-reasoning run).

        >>> async def thinking(self, ota_context, context=None):
        ...     msgs = my_messages(ota_context, context)
        ...     return await self._llm.aselect_tool(messages=msgs, tools=...)
        """
        raise NotImplementedError(
            "CognitiveWorker.thinking must be overridden: assemble the prompt "
            "from the two contexts, call self._llm, and return "
            "(tool_calls, content), a pydantic BaseModel, or text."
        )

    async def before_action(self, ota_context: OTAContext, context: Optional[Context] = None) -> Any:
        """Worker-level before_action hook. Override to intercept the decision.

        No decision argument — the pending decision is already on the current
        OTA round: read it from ``ota_context.think_result``. Return
        ``_DELEGATE`` / ``None`` to chain to the agent-level hook; return any
        other value (or ``yield RETURN(...)``) to override the decision before
        the act phase runs.
        """
        return _DELEGATE

    async def after_action(self, ota_context: OTAContext, context: Optional[Context] = None) -> Any:
        """Worker-level after_action hook. Override for side-effects.

        No result argument — the action result is already on the current OTA
        round: read it from ``ota_context.action_result``. The return value is
        a control signal: ``_DELEGATE`` / ``None`` chains to the agent-level
        hook; any other value suppresses it. Folding extra fields onto
        ``ota_context`` is the framework's one sanctioned in-place seam (the
        round trace is mutated by design); do not rely on mutating
        user-supplied data elsewhere.
        """
        return _DELEGATE

    ############################################################################
    # Entry point
    ############################################################################

    async def arun(
        self,
        *args: Any,
        feedback_data: Optional[Union[InteractionFeedback, List[InteractionFeedback]]] = None,
        **kwargs: Any,
    ) -> Any:
        """Execute the thinking phase. The automa runs the observe step first; this round's observation is on the latest ``OTARecord`` (read via ``.obs_result``)."""
        start_time = time.monotonic()
        result = await super().arun(*args, feedback_data=feedback_data, **kwargs)
        self.spent_time += time.monotonic() - start_time
        return result

class AgentWorker(GraphAutoma):
    """External-agent worker — one delegated cycle.

    Concrete class, peer of ``CognitiveWorker``. Anchored on a
    ``BaseAgent`` (its BASE, exactly as ``CognitiveWorker`` is anchored
    on a ``BaseLlm``): the worker only ever calls ``self._agent.run()``;
    the ``BaseAgent`` owns how the CLI is actually driven.

    Constructor
    -----------
    agent : BaseAgent
        The external coding-agent driver (``ClaudeCodeAgent``, …).
    verbose : Optional[bool]
        Logging override (``None`` = inherit from ``AmphibiousAutoma``).
    verbose_prompt : Optional[bool]
        When truthy, log the assembled message before each delegation.

    No ``goal`` / ``tools`` / ``skills`` / ``history`` knobs — those are
    all carried by the contexts the framework passes in, and read
    straight from them (``thinking()`` reads ``ota_context.user_input``
    etc.).

    Customize by subclassing and overriding the template methods —
    ``thinking`` (assemble the message), ``observation``,
    ``before_action``, ``after_action``. The default ``thinking``
    already produces a sensible message, so ``AgentWorker(agent)`` works
    out of the box with no subclass.

    >>> reviewer = think_agent(AgentWorker(
    ...     ClaudeCodeAgent(allowed_builtin_tools=["Read", "Grep"]),
    ... ))
    >>>
    >>> class StrictReviewer(AgentWorker):
    ...     async def thinking(self, ota_context, context=None):
    ...         base = await super().thinking(ota_context, context)
    ...         return base + "\\n\\nBe extremely thorough."
    """

    def __init__(
        self,
        agent: BaseAgent,
        *,
        verbose: Optional[bool] = None,
        verbose_prompt: Optional[bool] = None,
    ) -> None:
        super().__init__()

        if not isinstance(agent, BaseAgent):
            raise TypeError(
                f"AgentWorker(agent, ...) requires a BaseAgent instance; "
                f"got {type(agent).__name__}. Use ClaudeCodeAgent(...) or "
                "subclass BaseAgent."
            )
        # The BASE — mirrors ``CognitiveWorker._llm``.
        self._agent: BaseAgent = agent

        # Logging runtime (``None`` = inherit / off). ``verbose`` mirrors
        # ``CognitiveWorker.verbose``; ``verbose_prompt`` logs the prompt.
        self._verbose = verbose
        self._verbose_prompt = verbose_prompt

        # Framework-injected per-call wiring (set by
        # ``AmphibiousAutoma._run_think_agent`` before driving). Each
        # CLI tool call is surfaced onto this channel as a decision; the
        # framework's consumer executes it and returns the result. The
        # worker itself never touches ``_run_action_call`` — same spirit
        # as ``CognitiveWorker`` only ever holding a ``_llm``.
        self._decision_channel: Optional["asyncio.Queue"] = None

        # Wall-clock spent across this worker's delegations. (No
        # ``spent_tokens`` — an external CLI agent's token usage is not
        # visible to the framework; its stdout is not captured.)
        self.spent_time = 0.0

    ############################################################################
    # Core dispatch — NOT user-overridable
    ############################################################################

    @worker(is_start=True, is_output=True)
    async def _think(
        self,
        ota_context: OTAContext,
        context: Optional[Context] = None,
    ) -> Any:
        """Think-phase entry (mirrors ``CognitiveWorker._thinking``).

        Both contexts are injected by the dispatcher
        (``arun(ota_context=…, context=…)``): ``ota_context`` is the
        small-loop OTA state, ``context`` the free-form knowledge. Returns
        the ``AgentResult`` from the delegation — the parent dispatcher
        unwraps ``.output`` for the ``yield ThinkAgent`` asend value.
        """
        if not isinstance(ota_context, OTAContext):
            raise TypeError(
                f"Expected OTAContext, got {type(ota_context).__name__}. "
                "AgentWorker is driven with the small-loop OTA context."
            )
        return await self._run_think(ota_context, context)

    async def _run_think(
        self,
        ota_context: OTAContext,
        context: Optional[Context] = None,
    ) -> AgentResult:
        """Organize context, then delegate to ``self._agent.run(...)``.

        Phases:

        1. MCP-ify ``ctx.tools`` — derive bindings, boot the in-process
           FastMCP host (project tools + the ``agent_done`` signal).
        2. Assemble the message via the ``thinking()`` template.
        3. Pack an ``AgentRequest`` (message + cwd + mcp servers +
           allow-list + completion future).
        4. ``await self._agent.run(request)`` — the BaseAgent owns the
           CLI mechanics from here.
        5. Tear the host down; return the ``AgentResult``.

        This method is the AgentWorker's whole job: context organization
        + delegation. It is not user-overridable — override ``thinking``
        / the hooks instead.

        Returns the ``AgentResult`` straight from ``self._agent.run`` —
        no result is stashed on the worker; the outcome flows out
        through the return value, mirroring ``CognitiveWorker.arun``.
        """
        ########################
        # 1. MCP-ify ctx.tools → boot the host
        ########################
        builtin_names = {t.tool_name for t in ALL_BUILTIN_TOOLS}
        bindings = self._build_bindings_from_ctx(ota_context, builtin_names)

        loop = asyncio.get_event_loop()
        agent_done_future: asyncio.Future[str] = loop.create_future()

        async def _on_tool_call(tool_name: str, args: Dict[str, Any]) -> Any:
            return await self._emit_decision(tool_name, args)

        def _on_agent_done(result: str) -> None:
            if not agent_done_future.done():
                agent_done_future.set_result(result)

        host = MCPHost(
            server_name=self._mcp_server_name(),
            bindings=bindings,
            on_tool_call=_on_tool_call,
            on_agent_done=_on_agent_done,
        )
        await host.start()

        ########################
        # 2-5. Assemble request → delegate → teardown
        ########################
        try:
            with tempfile.TemporaryDirectory(prefix="amphi-delegate-") as tmp:
                cwd = Path(tmp)

                message = await self.thinking(ota_context, context)
                if self._verbose_prompt:
                    printer.print(
                        f"[AgentWorker] prompt → {type(self._agent).__name__}",
                        color="cyan",
                    )
                    printer.print(message, color="gray")

                exposed_tool_names = [
                    f"mcp__{host.server_name}__{b.name}" for b in bindings
                ]
                exposed_tool_names.append(f"mcp__{host.server_name}__agent_done")

                request = AgentRequest(
                    message=message,
                    cwd=cwd,
                    mcp_servers={
                        host.server_name: {"type": "http", "url": host.url},
                    },
                    allowed_tools=exposed_tool_names,
                    done_signal=agent_done_future,
                )
                result = await self._agent.run(request)
        finally:
            await host.stop()

        return result

    def _mcp_server_name(self) -> str:
        """MCP server name advertised to the external agent.

        Subclasses may override. The name prefixes every bridged tool's
        external identifier (``mcp__<server>__<tool>``).
        """
        return "amphi-bridge"

    ############################################################################
    # Internal helpers — MCP bindings / decision emit / cloning
    ############################################################################

    def _build_bindings_from_ctx(
        self, ctx: OTAContext, builtin_names: set,
    ) -> List[MCPToolBinding]:
        """Derive MCP tool bindings from ``ctx.tools``.

        Skips framework built-ins (the CLI has its own); any
        ``expose_tools`` filtering already happened upstream where the
        parent dispatcher built the sub-run's OTA context with ``ota.tools``
        narrowed to the whitelisted set before invoking us.
        """
        bindings: List[MCPToolBinding] = []
        for tool in ctx.tools:
            name = tool.tool_name
            if name in builtin_names:
                continue
            bindings.append(MCPToolBinding(
                name=name,
                description=tool.tool_description or name,
                parameters=tool.tool_parameters or {
                    "type": "object",
                    "properties": {},
                    "required": [],
                },
            ))
        return bindings

    async def _emit_decision(
        self, tool_name: str, args: Dict[str, Any],
    ) -> Any:
        """Surface a CLI tool call as a decision for the framework to execute.

        Each tool call the external agent makes over MCP is, in effect,
        a ``think_result`` — "I want to call this tool." The worker does
        NOT execute it. It builds the decision, hands it to the framework
        over ``self._decision_channel``, and relays back whatever the
        framework's consumer fills in.

        This keeps ``AgentWorker`` symmetric with ``CognitiveWorker``:
        both only *produce* decisions; ``AmphibiousAutoma`` is the one
        that *acts*. The worker never references ``_run_action_call``.

        The framework's consumer runs each decision through
        ``_run_action_call`` (worker ``before_action`` / ``after_action``
        hooks fire, the call folds onto the OTA context's current round +
        the ``AgentTrace``), then resolves the future with the resulting
        ``Step`` — which is unwrapped here into the raw tool result the
        external agent expects back.
        """
        if self._decision_channel is None:
            raise RuntimeError(
                "AgentWorker has no decision channel — it must be driven "
                "by AmphibiousAutoma._run_think_agent, which wires the "
                "channel and runs the consumer that executes each decision."
            )
        # Build the decision directly: each external-agent tool call becomes a
        # one-call ``ThinkResult`` for the framework's consumer to execute.
        # Args are carried as ``ToolArgument`` name/value pairs, so a tool
        # parameter named ``description`` / ``tool_name`` poses no collision.
        decision = ThinkResult(
            step_content=f"[think_agent] {tool_name}",
            tool_calls=[StepToolCall(
                tool=tool_name,
                tool_arguments=[
                    ToolArgument(name=k, value=str(v)) for k, v in args.items()
                ],
            )],
        )
        result_future: "asyncio.Future" = (
            asyncio.get_event_loop().create_future()
        )
        await self._decision_channel.put((decision, result_future))
        step = await result_future
        return _extract_tool_result(step)

    def _clone(self) -> "AgentWorker":
        """Return a fresh worker with the same configuration.

        The ``BaseAgent`` is *shared* across clones — it is stateless
        per call (every dynamic input arrives via ``AgentRequest``),
        exactly as a ``BaseLlm`` is shared across ``CognitiveWorker``
        clones. Subclasses with extra ``__init__`` params should
        override.

        Used by ``ThinkAgentDescriptor._clone_worker`` for state
        isolation at every ``yield ThinkAgent(...)``.
        """
        return type(self)(
            self._agent,
            verbose=self._verbose,
            verbose_prompt=self._verbose_prompt,
        )

    ############################################################################
    # Template methods (override by user to customize the behavior)
    ############################################################################

    async def observation(self, ota_context: OTAContext, context: Optional[Context] = None) -> Any:
        """Worker-level observation hook. Override to customize.

        Same contract as ``CognitiveWorker.observation``: returning
        ``_DELEGATE`` (or ``None``) hands off to
        ``AmphibiousAutoma.observation()``. Other values are used as
        the observation directly.
        """
        return _DELEGATE

    async def thinking(
        self,
        ota_context: OTAContext,
        context: Optional[Context] = None,
    ) -> str:
        """Assemble the message handed to the external agent.

        The AgentWorker analog of ``CognitiveWorker.thinking()`` — it
        produces the prompt. The default layout is goal → parent context
        (small-loop task summary + big-loop knowledge + observation) →
        completion contract; override to restructure or inject domain
        instructions.

        Two-loop inputs (both injected by the dispatcher): ``ota_context``
        is the small-loop OTA context — its ``user_input`` is the resolved
        goal for this delegation; ``context`` is the free-form big-loop
        knowledge context (``None`` for a pure-reasoning run). The project
        tools are NOT enumerated here — the agent discovers them through
        the MCP server; the contract just needs to tell it they exist and
        how to finish.
        """
        goal = str(ota_context.user_input or "")
        context_info = _format_context_info(ota_context, context, ota_context.obs_result)

        parts = [
            "You are running as the external agent layer of an AmphibiousAutoma.",
            "",
            "GOAL:",
            goal or "(no goal supplied)",
        ]
        if context_info:
            parts += ["", "PARENT CONTEXT:", context_info]
        parts += [
            "",
            "PROJECT TOOLS:",
            "  Project tools are exposed to you over MCP. Calling them "
            "routes back into the parent automa's hook pipeline.",
            "",
            "COMPLETION CONTRACT:",
            "  When the goal is fully complete, call the `agent_done` "
            "tool with `result` set to the final answer / summary "
            "string, then finish. The parent automa resumes with the "
            "value you pass.",
        ]
        return "\n".join(parts)

    async def before_action(
        self,
        ota_context: OTAContext,
        context: Optional[Context] = None,
    ) -> Any:
        """Worker-level before_action hook. Override to intercept bridged tool calls.

        Same contract as ``CognitiveWorker.before_action``: ``_DELEGATE`` /
        ``None`` chains to the agent-level hook; any other value
        overrides the decision.
        """
        return _DELEGATE

    async def after_action(
        self,
        ota_context: OTAContext,
        context: Optional[Context] = None,
    ) -> Any:
        """Worker-level after_action hook. Override for side-effects.

        Same contract as ``CognitiveWorker.after_action``: ``_DELEGATE`` /
        ``None`` chains to the agent-level hook; any other value
        suppresses it.
        """
        return _DELEGATE

    ############################################################################
    # Entry point
    ############################################################################

    async def arun(
        self,
        *args: Any,
        feedback_data: Optional[Union[InteractionFeedback, List[InteractionFeedback]]] = None,
        **kwargs: Any,
    ) -> Any:
        """Execute the think phase.

        Observation must be pre-set in ``context.observation`` (handled
        by ``AmphibiousAutoma._run_think_agent``). Returns the
        ``AgentResult`` for the delegation (mirrors how
        ``CognitiveWorker.arun`` returns its decision).
        """
        start_time = time.monotonic()
        result = await super().arun(*args, feedback_data=feedback_data, **kwargs)
        self.spent_time += time.monotonic() - start_time
        return result