Model backends

A backend is how an agent talks to a model. It is a configuration choice: the graph and the science do not change with it. See Use a different model or backend for choosing one, and Installation for what each needs.

The adapters below are exported for direct use and for subclassing; a normal run selects one by name from config.yaml and never touches these classes.

adda.ClaudeAdapter

Wraps claude-agent-sdk; runs one agent turn and returns assistant text.

The SDK handles its own tool-execution loop (Bash, Read, Write, Edit). This adapter converts a list of LangChain-style message dicts to the SDK format, runs the query, and assembles the final text response.

Parameters:

Name	Type	Description	Default
model	str	Claude model identifier.	required
system_prompt	str	System prompt for the agent.	required
study_dir	Path or None	Working directory passed to the SDK as cwd.	None
native_tools	list[str]	Tool names to enable (e.g. ["Bash", "Read", "Write"]).	None
closure_tools	dict[str, callable] or None	Extra Python callables exposed to the model as MCP tools.	None

Source code in src/adda/_src/backends/claude.py

class ClaudeAdapter:
    """Wraps claude-agent-sdk; runs one agent turn and returns assistant text.

    The SDK handles its own tool-execution loop (Bash, Read, Write, Edit).
    This adapter converts a list of LangChain-style message dicts to the SDK
    format, runs the query, and assembles the final text response.

    Parameters
    ----------
    model : str
        Claude model identifier.
    system_prompt : str
        System prompt for the agent.
    study_dir : Path or None
        Working directory passed to the SDK as ``cwd``.
    native_tools : list[str]
        Tool names to enable (e.g. ``["Bash", "Read", "Write"]``).
    closure_tools : dict[str, callable] or None
        Extra Python callables exposed to the model as MCP tools.
    """

    # CLI tools the Claude SDK executes natively. A node's other declared tools
    # are injected as Python closures (MCP), not passed here.
    #: This backend has its own default system prompt that a node may keep.
    HAS_BASE_PROMPT = True
    base_prompt: str | None = None
    NATIVE_TOOLS = frozenset({
        "Bash", "Edit", "Read", "Write", "Glob", "Grep",
        # Bash's own SDK companions: poll a backgrounded shell / kill it. These
        # are SDK built-ins we previously omitted, so an agent that got a
        # backgroundTaskId (from auto-background on timeout) had no tool to act
        # on it. Granting them by declaration closes that awareness gap.
        "BashOutput", "KillShell",
        "Task", "WebFetch", "WebSearch",
    })

    @classmethod
    def select_native_tools(cls, agent_tools) -> list[str]:
        """Pick which of an agent's declared tools are native SDK CLI tools.

        Mirror of OpenAICompatibleAdapter.select_native_tools so the runtime
        can choose native tools generically for any backend (forward-compatible
        dispatch)."""
        picked = [t for t in agent_tools if t in cls.NATIVE_TOOLS]
        if DEFAULT_TOOLS in agent_tools:
            picked.append(DEFAULT_TOOLS)
        return picked

    def __init__(
        self,
        model: str,
        system_prompt: str,
        study_dir: Path | None = None,
        native_tools: list[str] | None = None,
        closure_tools: dict[str, Any] | None = None,
        extra_mcp_servers: dict | None = None,
        extra_allowed_tools: list[str] | None = None,
        persistent: bool = False,
        max_history_pairs: int = 5,
    ) -> None:
        self.model = model
        self.system_prompt = system_prompt
        self.study_dir = Path(study_dir) if study_dir else None
        # "Default" is a marker, not a tool name: the node takes the CLI's
        # whole default built-in set (see node_tools.py).
        self.use_default_tools: bool = DEFAULT_TOOLS in (native_tools or [])
        self.native_tools = [t for t in (native_tools or []) if t != DEFAULT_TOOLS]
        # Called once per init record with the tool names the CLI really
        # loaded; set by the runtime, never required.
        self.on_init_tools: Any = None
        self.closure_tools = dict(closure_tools or {})
        self.extra_mcp_servers: dict = dict(extra_mcp_servers or {})
        self.extra_allowed_tools: list[str] = list(extra_allowed_tools or [])
        # persistent and max_history_pairs kept for backward compatibility with
        # Agent subclasses and tests that read these attributes; not used
        # in the core invocation path (history is demand-driven via DelegationLog).
        self.persistent: bool = persistent
        self.max_history_pairs: int = max_history_pairs
        # Serialises calls on THIS adapter object; each delegation runs on its own copy().
        self._lock: threading.Lock = threading.Lock()
        # Set by an orchestrating node; when truthy, the generator is closed after
        # the next AssistantMessage so the session ends on a routing decision.
        self.route_watcher: Any = None
        # Populated after each ainvoke() with token counts from ResultMessage.
        self.last_usage: dict = {}
        self._attempt_usages: list[dict] | None = None
        # Populated after each ainvoke() with the CLI session id (spec 12
        # item 3: session-resumption plumbing, capture-only for now) --
        # ResultMessage's when the turn completed normally, else whatever
        # the last AssistantMessage carried.
        self.last_session_id: str | None = None
        self._background_watch: Any = None

    def _system_prompt_option(self):
        """What the CLI receives: the text alone (it REPLACES Claude Code's
        prompt), or, for a node with ``base_prompt: Default``, a preset that
        keeps Claude Code's prompt and appends the text."""
        text = self._render_system_prompt()
        if self.base_prompt == DEFAULT_PROMPT:
            return {"type": "preset", "preset": "claude_code", "append": text}
        return text

    def _render_system_prompt(self) -> str:
        """The system prompt exactly as the model sees it: base prompt plus
        the ``<tools>`` catalog, with every tool the prose names rewritten to
        the qualified name the SDK exposes (catalog and prose must agree)."""
        from ..prompts.tool_catalog import (
            qualify_tool_mentions,
            system_prompt_with_catalog,
        )
        qualified = _qualify_closure_names(self.closure_tools)
        rendered = system_prompt_with_catalog(
            self.system_prompt, qualified, builtins_held=self.use_default_tools)
        return qualify_tool_mentions(rendered, {
            bare: f"mcp__{_CLOSURE_MCP_SERVER}__{bare}"
            for bare in self.closure_tools
        })

    def _compute_allowed_tools(self, qualified_mcp_tools) -> list[str]:
        """All allowed tool names, ALWAYS as a list (never None).

        The SDK does ``list(options.allowed_tools)`` when building its command,
        which raises ``TypeError`` on ``None`` — so a tool-less agent (e.g. the
        one-shot problem-statement reviewer) must still get ``[]`` here, not
        ``None``. An empty list correctly means "no tools allowed".
        """
        return (
            list(qualified_mcp_tools)
            + list(self.native_tools)
            + list(self.extra_allowed_tools)
        )

    def copy(self) -> ClaudeAdapter:
        """An independent adapter for ONE delegation.

        Shares configuration; owns everything a delegation mutates: its own
        ``closure_tools`` (dispatch binds ReportEvals / Write / FollowUp to
        that delegation's id), its own ``_lock`` and its own per-call
        ``last_*`` state. Same-role delegations therefore run concurrently
        instead of queueing behind one shared lock.
        """
        import copy as _copy
        twin = _copy.copy(self)
        twin.closure_tools = dict(self.closure_tools)
        twin.native_tools = list(self.native_tools)
        twin.extra_allowed_tools = list(self.extra_allowed_tools)
        twin.extra_mcp_servers = dict(self.extra_mcp_servers)
        twin._lock = threading.Lock()
        twin.last_usage = {}
        twin._attempt_usages = None
        twin.last_session_id = None
        return twin

    async def ainvoke(
        self, messages: list[dict], *, idle_timeout: float | None = None,
        resume: str | None = None,
    ) -> str:
        """Run one agent turn asynchronously; return assembled text.

        ``idle_timeout`` overrides the run-wide ``llm_stream_idle_timeout`` for
        THIS call only — used by short advisory side-calls (e.g. the verdict
        validator) that must not inherit a real agent turn's generous window.

        ``resume`` (spec 12 item 3): a CLI session id to resume rather than
        starting fresh -- ``messages`` then carries only the NEW turn (the
        prior conversation is loaded from the resumed session itself, not
        replayed here). ``fork_session=False`` always, so this continues the
        SAME session rather than branching a copy of it.
        """
        _require_sdk()
        from claude_agent_sdk import (
            AssistantMessage,
            ClaudeAgentOptions,
            ResultMessage,
            SdkMcpTool,
            StreamEvent,
            SystemMessage,
            TextBlock,
            ToolUseBlock,
            UserMessage,
            create_sdk_mcp_server,
            query,
        )

        from ..prompts.tool_catalog import tool_summary

        # Build MCP server from closure_tools if any
        mcp_servers: dict = {}
        qualified_mcp_tools: list[str] = []
        if self.closure_tools:
            server_name = _CLOSURE_MCP_SERVER
            sdk_tools: list[Any] = []
            for tool_name, fn in self.closure_tools.items():
                schema = _infer_schema_from_callable(fn)

                async def _handler(args: dict, bound_fn: Any = fn) -> dict:
                    try:
                        result = bound_fn(**args)
                    except Exception as exc:
                        return {
                            "content": [
                                {"type": "text", "text": f"ERROR: {exc}"}
                            ],
                            "is_error": True,
                        }
                    return {
                        "content": [
                            {
                                "type": "text",
                                "text": (
                                    str(result) if result is not None else ""
                                ),
                            }
                        ]
                    }

                sdk_tools.append(
                    SdkMcpTool(
                        name=tool_name,
                        description=tool_summary(fn, tool_name),
                        input_schema=schema,
                        handler=_handler,
                    )
                )

            mcp_cfg = create_sdk_mcp_server(
                name=server_name, tools=sdk_tools or None
            )
            mcp_servers = {server_name: mcp_cfg}
            qualified_mcp_tools = list(_qualify_closure_names(self.closure_tools))

        # Merge external stdio MCP servers declared by the Agent subclass.
        if self.extra_mcp_servers:
            mcp_servers.update(self.extra_mcp_servers)

        _base_disallowed = ["WebSearch", "WebFetch", "Task", "ExitPlanMode"]
        # Under permission_mode="bypassPermissions" the allowed_tools allowlist
        # is NOT enforced — disallowed_tools is the only thing that binds. So a
        # native tool the agent never declared (e.g. Bash/Write for a read-only
        # reviewer) would otherwise be silently usable. Disallow every native
        # tool this agent did not declare, making its declared toolset binding.
        _ungranted_native = [
            t for t in self.NATIVE_TOOLS if t not in self.native_tools
        ]
        if self.use_default_tools:
            # Default has no floor. Only computed additions stay: a built-in
            # that shares its bare name with a closure this node declares
            # (the sandboxed Write replaces the native one) must not run.
            _base_disallowed, _ungranted_native = [], list(self.closure_tools)
        _effective_disallowed = [
            t for t in dict.fromkeys([*_base_disallowed, *_ungranted_native])
            if t not in self.extra_allowed_tools
        ]

        # Non-blocking raw-oracle nudge: a PostToolUse hook that injects a
        # reminder (capped per delegation = per ainvoke) when a Bash/Write
        # call reaches the oracle directly instead of via get_evaluator().
        # Best-effort — if the SDK hook API is unavailable, run without it.
        _hooks = None
        try:
            from claude_agent_sdk import HookMatcher

            from ..runtime import features
            from .base import (
                OracleNudgeBudget,
                get_delegation_id,
                get_run_config_path,
                oracle_registered,
            )
            # Silent until an oracle is registered: pre-registration work (the
            # datagenerator wrapping/validating its raw source) has no
            # get_evaluator() to use, so nudging it is a false positive.
            # The nudge is a monitor intervention; the store notices the same
            # hook carries are store integrity and stay on in every arm.
            _nudge = OracleNudgeBudget(
                enabled=oracle_registered()
                and features.enabled("science_monitor"))
            # Expose on the adapter so the runtime can drain + log its
            # firings as direct evidence (see _record_intervention).
            _nudge.run_config_path = get_run_config_path()
            self._oracle_nudge = _nudge

            # The hook may run on another thread, so bind the delegation and
            # the run's debug dir now, while this thread still holds them.
            from ..infra import pending_notices as _pn
            _rc = get_run_config_path()
            _pn_dir = Path(_rc).parent if _rc else None
            _pn_did = get_delegation_id()

            async def _oracle_hook(input_data, tool_use_id, context):
                msg = _pn.post_tool_context(
                    _nudge,
                    input_data.get("tool_name", ""),
                    input_data.get("tool_input") or {},
                    _pn_dir, _pn_did,
                )
                if not msg:
                    return {}
                return {
                    "hookSpecificOutput": {
                        "hookEventName": "PostToolUse",
                        "additionalContext": msg,
                    }
                }

            _hooks = {"PostToolUse": [HookMatcher(hooks=[_oracle_hook])]}
        except Exception:  # noqa: BLE001 — nudge is best-effort, never fatal
            _hooks = None

        # Per-session env: the SDK MERGES this over the inherited environment
        # (PATH etc. preserved), so bare extra keys are safe. See
        # _build_session_env for what is injected and why.
        _sess_env: dict = _build_session_env()

        from ..runtime.settings import get_float
        _max_buf_mb = get_float("llm_max_buffer_mb", 30.0)
        _max_buf = int(_max_buf_mb * 1024 * 1024)

        # The SDK spawns the CLI with cwd=self.study_dir; if that directory
        # doesn't exist the subprocess dies with a cryptic CLIConnectionError
        # ("Working directory does not exist") mid-delegation. Create it
        # defensively so a missing worker workspace can never abort a run.
        if self.study_dir:
            try:
                self.study_dir.mkdir(parents=True, exist_ok=True)
            except Exception:  # noqa: BLE001 — best-effort; spawn surfaces real errors
                pass

        options = ClaudeAgentOptions(
            system_prompt=self._system_prompt_option(),
            model=self.model,
            cwd=str(self.study_dir) if self.study_dir else None,
            tools=({"type": "preset", "preset": "claude_code"}
                   if self.use_default_tools else self.native_tools or []),
            mcp_servers=mcp_servers if mcp_servers else {},
            allowed_tools=self._compute_allowed_tools(qualified_mcp_tools),
            disallowed_tools=_effective_disallowed,
            permission_mode="bypassPermissions",
            strict_mcp_config=bool(mcp_servers) or bool(self.extra_mcp_servers),
            # Hermetic session: load NO filesystem settings, so worker/critic
            # subprocesses don't inherit the developer's global ~/.claude hooks
            # (e.g. cbm-code-discovery-gate, which blocked legitimate Read calls
            # for workers AND the critic). Our own hooks are passed
            # programmatically via options.hooks below (audit/#1: fresh hooks).
            setting_sources=[],
            env=_sess_env,
            # Stream-message buffer ceiling. Default 1MB is far too small for a
            # literature reviewer whose tools return full PDFs — a single >1MB
            # MCP tool result overflowed it and FATALLY (non-retried) killed the
            # whole delegation (D003). 30MB clears realistic PDFs; non-PDF
            # results never approach it. A result still exceeding this is caught
            # gracefully below (turn cut short + marker), not a fatal crash.
            # Tune via F3DASM_LLM_MAX_BUFFER_MB.
            max_buffer_size=_max_buf,
            # Partial streaming → a fine-grained heartbeat: the stream emits a
            # StreamEvent sub-second while genuinely generating, so total
            # silence becomes a reliable stall signal and the
            # idle timeout can be bounded without false-positiving a
            # slow-but-working generation.
            include_partial_messages=True,
            **_thinking_options(self.model),
            **({"hooks": _hooks} if _hooks else {}),
            **(
                {"resume": resume, "fork_session": False}
                if resume is not None else {}
            ),
        )

        prompt_str = _format_messages_as_prompt(messages)

        last_assistant = None
        last_result: Any = None
        _buffer_overflowed = False
        _deliberate_break = False
        _msg_usage: dict[str, dict] = {}
        _cur_msg_id: list[str | None] = [None]
        raise_if_stopped("model call")
        gen = query(prompt=prompt_str, options=options)
        # Idle-stream timeout — turns a silent stream into a retryable
        # TimeoutError. Resets on EVERY stream message. Scoped to model
        # generation: while a tool runs the window stands down (see _phase).
        # MEASURED (the stream_evt instrumentation settled this): the bundled
        # CLI forwards NO ping events — only message lifecycle — so during a
        # mid-generation pause we get ZERO liveness signal. Two supercompressible
        # (Sonnet) runs captured LEGITIMATE (recovered) active-generation gaps of
        # 53.6s AND 252-256s (one mid-tool_use-block composition, one
        # tool_result→next message_start delay). So earlier 60s/180s windows
        # would guillotine legitimate slow generations. The observed ~256s legit
        # gap left only ~44s of headroom under the prior 300s default, and with
        # NO ping liveness signal a longer legit generation is indistinguishable
        # from a stall — so the default is 600s: ample margin over the measured
        # legit gaps while still catching a truly dead (silent-forever) stream in
        # ~10 min. A genuine runaway is the delegation watchdog's concern, not
        # this window's. Knobs (config.yaml runtime block; env overrides):
        # llm_stream_idle_timeout (0 disables); llm_tool_idle_timeout caps tool
        # execution (0 = uncapped).
        from ..runtime.settings import get_float as _get_float
        _idle = (idle_timeout if idle_timeout is not None
                 else _get_float("llm_stream_idle_timeout", 600.0))
        _tool_idle = _get_float("llm_tool_idle_timeout", 0.0)

        def _phase(msg: Any):
            # True: a tool is now executing → suspend the tight idle window.
            # False: generation active / tool result returned → tight window.
            # None: leave phase unchanged.
            if isinstance(msg, AssistantMessage):
                return any(
                    isinstance(b, ToolUseBlock) for b in msg.content
                )
            if isinstance(msg, StreamEvent):
                # Only message_start opens a new generation. The SDK emits
                # content_block_stop / message_delta / message_stop AFTER the
                # AssistantMessage that carries a ToolUseBlock, while the tool
                # is still running; treating those as "generation resumed"
                # re-armed the stall window mid-tool and killed a 91-minute
                # delegation that was blocked in a long tool call.
                ev = getattr(msg, "event", None)
                if isinstance(ev, dict) and ev.get("type") == "message_start":
                    return False
                return None
            if isinstance(msg, UserMessage):
                return False
            return None

        from .base import append_transcript, debug_enabled

        def _record(msg: Any):
            # Full reasoning + tool-calls + tool-results; StreamEvent partials
            # are skipped (the assembled AssistantMessage carries the text).
            if isinstance(msg, AssistantMessage):
                texts, tools, thinking = [], [], []
                thinking_omitted = 0
                for b in msg.content:
                    if isinstance(b, TextBlock):
                        texts.append(b.text)
                    elif isinstance(b, ToolUseBlock):
                        tools.append({"name": b.name, "input": b.input,
                                      "tool_use_id": getattr(b, "id", None)})
                    else:
                        t = (getattr(b, "thinking", None)
                             or getattr(b, "text", None))
                        if t:
                            thinking.append(t)
                        elif hasattr(b, "thinking"):
                            # A thinking block that arrived with no text
                            # (display "omitted"): counted, so "the model
                            # thought but it was hidden" is distinguishable
                            # from "the model did not think".
                            thinking_omitted += 1
                return {"type": "assistant", "text": "".join(texts),
                        "tools": tools, "thinking": thinking,
                        "thinking_omitted": thinking_omitted,
                        "message_id": getattr(msg, "message_id", None)}
            if isinstance(msg, UserMessage):
                results = []
                for b in (getattr(msg, "content", None) or []):
                    results.append({
                        "tool_use_id": getattr(b, "tool_use_id", None),
                        "content": getattr(b, "content", b),
                    })
                return {"type": "tool_result", "results": results}
            if isinstance(msg, ResultMessage):
                return {"type": "result",
                        "usage": getattr(msg, "usage", None),
                        "model_usage": getattr(msg, "model_usage", None),
                        "cost_usd": getattr(msg, "total_cost_usd", None)}
            if isinstance(msg, SystemMessage):
                # SystemMessage used to be invisible here entirely (this
                # function returned None for anything it didn't recognize) —
                # so a real compact_boundary (the SDK's own context-
                # compaction event) left NO trace in debug/transcripts/,
                # the viewer, or any post-run analysis. Confirmed empirically
                # (a short forced-window test): a single short session
                # produced 50-119 SystemMessages, the overwhelming majority
                # subtype "thinking_tokens" — a per-token streaming heartbeat,
                # pure noise at transcript granularity (stream_evt already
                # covers liveness). Record everything ELSE verbatim,
                # including (especially) compact_boundary's own metadata.
                if msg.subtype in _SYSTEM_MESSAGE_NOISE_SUBTYPES:
                    return None
                return {"type": "system", "subtype": msg.subtype,
                        "data": msg.data}
            return None

        _capture = debug_enabled()
        # Partial-stream checkpointing: an agent that never completes a message
        # (infinite thinking / an unresolved turn) emits only StreamEvents and
        # would otherwise disclose nothing. Buffer the deltas and flush a
        # "partial" record every N events (and on any stream teardown) so the
        # transcript reveals what a stuck turn is doing in near-real-time.
        _PARTIAL_FLUSH_EVERY = 25
        _pbuf: list[str] = []
        _pcount = [0]

        def _extract_delta(ev: Any) -> str:
            if not isinstance(ev, dict):
                return ""
            d = ev.get("delta") or {}
            return (d.get("text") or d.get("thinking")
                    or d.get("partial_json") or "")

        def _flush_partial() -> None:
            if _pbuf:
                append_transcript({"type": "partial", "text": "".join(_pbuf),
                                   "events": _pcount[0]})
                _pbuf.clear()

        try:
            _stream = (
                _stream_with_idle_timeout(
                    gen, _idle,
                    tool_timeout=_tool_idle, classify=_phase,
                )
                if _idle > 0 else gen
            )
            # Measurement clock: the first event's gap below = time-to-first
            # stream event (≈ prefill latency).
            _last_evt = [time.monotonic()]
            async for msg in _stream:
                raise_if_stopped("model call")
                if isinstance(msg, StreamEvent):
                    _track_message_usage(
                        getattr(msg, "event", None), _msg_usage, _cur_msg_id)
                elif isinstance(msg, AssistantMessage) \
                        and getattr(msg, "usage", None) \
                        and getattr(msg, "message_id", None):
                    _msg_usage.setdefault(msg.message_id, dict(msg.usage))
                if _capture:
                    if isinstance(msg, StreamEvent):
                        _pcount[0] += 1
                        _ev = getattr(msg, "event", {}) or {}
                        _et = _ev.get("type", "?") if isinstance(
                            _ev, dict) else "?"
                        _now = time.monotonic()
                        _gap = _now - _last_evt[0]
                        _last_evt[0] = _now
                        # Record every NON-delta event (ping, message_start/
                        # stop, content_block_start/stop, message_delta) and any
                        # delta after a >2s pause — so we can see whether the
                        # stream stays alive (pings) during silent/prefill
                        # phases and the true inter-event gap distribution.
                        # This is what settles whether a 60s silence is a dead
                        # stream or a legitimately-slow first token.
                        if _et != "content_block_delta" or _gap > 2.0:
                            append_transcript({
                                "type": "stream_evt", "evt": _et,
                                "gap_s": round(_gap, 2),
                                **_usage_fields(_ev)})
                        _d = _extract_delta(_ev)
                        if _d:
                            _pbuf.append(_d)
                        if _pcount[0] % _PARTIAL_FLUSH_EVERY == 0:
                            _flush_partial()
                    else:
                        # A complete message: flush any buffered partial first,
                        # then the structured record.
                        _flush_partial()
                        _rec = _record(msg)
                        if _rec is not None:
                            append_transcript(_rec)
                if (isinstance(msg, SystemMessage) and msg.subtype == "init"
                        and self.on_init_tools is not None):
                    # Unconditional like the compaction record below: which
                    # tools the CLI really loaded is a run-level fact.
                    try:
                        self.on_init_tools(list((msg.data or {}).get("tools") or []))
                    except Exception:  # noqa: BLE001 — a notice never fails a turn
                        pass
                if isinstance(msg, SystemMessage) and msg.subtype == "compact_boundary":
                    # Unconditional (not gated on _capture/debug mode): a
                    # compaction is a run-level fact an analyst should never
                    # have to enable debug transcripts to discover.
                    record_stream_diagnostic(
                        "CONTEXT_COMPACTED",
                        "The SDK compacted this session's context "
                        "mid-turn (compact_boundary).",
                        compaction_data=msg.data,
                    )
                if isinstance(msg, AssistantMessage):
                    last_assistant = msg
                    if self.route_watcher and self.route_watcher():
                        _deliberate_break = True
                        break
                elif isinstance(msg, ResultMessage):
                    last_result = msg
                    break
        except Exception as exc:  # noqa: BLE001
            _m = str(exc).lower()
            if "maximum buffer size" in _m or "exceeded maximum" in _m:
                # Graceful contour: a single tool result (e.g. a huge PDF)
                # overflowed the stream buffer. The old behavior was a fatal,
                # NON-retried crash that killed the whole delegation (D003).
                # Instead, end the turn with what we have + a marker so the
                # agent can retry the fetch smaller — the delegation survives.
                _buffer_overflowed = True
            else:
                raise
        finally:
            if _capture:
                _flush_partial()  # disclose a stuck/torn-down turn's tail
            self._settle_usage(last_result, _msg_usage, last_assistant)
            _watch = getattr(self, "_background_watch", None)
            if _watch is not None:
                # Depth 2: the CLI itself (depth 1) is not a background job.
                _watch.end(min_depth=2)
            aclose = getattr(gen, "aclose", None)
            if aclose:
                try:
                    await aclose()
                except Exception:
                    pass

        # A stream that ends with neither a ResultMessage NOR a deliberate
        # route_watcher break, and wasn't already explained by a buffer
        # overflow, is abnormal: the CLI session ended (or the SDK's async
        # generator was exhausted) without ever completing its turn — report
        # 7 (run 20260830T004106, Oscar): the last AssistantMessage carried
        # only a tool call (e.g. Bash/TaskOutput), no result ever arrived,
        # and the near-empty `text` this then returns reads as a malformed
        # report to _invoke_with_report_retry, which silently re-invokes
        # with the ORIGINAL task — restarting the delegation from scratch,
        # orphaning whatever the first attempt launched, with nothing
        # logged anywhere. This does not change that return behaviour (a
        # future decision, pending Elvis) — it only makes the fact visible.
        if last_result is None and not _deliberate_break and not _buffer_overflowed:
            _last_tool = None
            if last_assistant is not None:
                for _b in last_assistant.content:
                    if isinstance(_b, ToolUseBlock):
                        _last_tool = _b.name
            record_stream_diagnostic(
                "STREAM_ENDED_WITHOUT_RESULT",
                "CLI stream ended without a ResultMessage or a deliberate "
                "route break — the turn may not have completed; its "
                "returned text can look like a malformed report and "
                "trigger a silent report-retry.",
                last_tool_in_flight=_last_tool,
                cli_session_id=getattr(last_assistant, "session_id", None),
            )

        text = ""
        if last_assistant is not None:
            for block in last_assistant.content:
                if isinstance(block, TextBlock):
                    text += block.text
        if _buffer_overflowed:
            _note = (
                f"[STREAM NOTE: a tool returned more than {_max_buf_mb:.0f} MB "
                "in a single result and overflowed the message buffer; that "
                "result was dropped and this turn was cut short (the delegation "
                "did NOT crash). Re-run the tool with a smaller/narrower request "
                "— fewer items, or a summary/extract instead of full text.]"
            )
            text = (text + "\n\n" + _note) if text else _note
        return text

    def _settle_usage(self, last_result: Any, _msg_usage: dict,
                      last_assistant: Any) -> None:
        """Record this attempt's usage and session id. Runs from ainvoke's
        ``finally`` so a stream that RAISES (idle TimeoutError, API error)
        still reports the usage it had streamed; ``invoke`` sums the
        attempts."""
        # Capture token usage from ResultMessage for run-level accounting.
        if last_result is not None:
            self.last_usage = {
                **(last_result.usage or {}),
                **_cache_split(last_result.usage or {}),
                "total_cost_usd": last_result.total_cost_usd,
            }
        elif _msg_usage:
            # route_watcher broke the stream on the AssistantMessage that
            # closes the run (Done()) before the SDK's ResultMessage -- the
            # only carrier of the session's cumulative usage and cost -- ever
            # arrived. The strategizer's whole run is ONE long stream, so
            # taking just the last message's usage recorded ~1 output token
            # for the entire run (run 20260928T141126). Sum the per-API-call
            # usage the stream itself reported instead. total_cost_usd stays
            # None: cost is a session-level rollup no single message carries,
            # the same "unknown, not zero" convention openai_compatible uses.
            self.last_usage = {
                **_sum_message_usage(_msg_usage.values()),
                "total_cost_usd": None,
            }
        elif last_assistant is not None and getattr(last_assistant, "usage", None):
            # A backend/stream that never sent message_start/delta events or a
            # message id: the lone message's own (possibly stale) usage.
            self.last_usage = {
                **last_assistant.usage, "total_cost_usd": None,
            }
        else:
            self.last_usage = {}

        # Claude reports input_tokens WITHOUT cache tokens, so each count maps
        # one to one onto the telemetry schema.
        u = self.last_usage
        self.last_usage = {**u, **normalized_usage(
            fresh_input=u.get("input_tokens"),
            cache_read=u.get("cache_read_input_tokens"),
            cache_write=u.get("cache_creation_input_tokens"),
            output=u.get("output_tokens"))}
        if _msg_usage:
            # Claude's input_tokens excludes cache, so a call's whole prompt
            # is the three input counts together.
            self.last_usage.update(call_shape([
                (m.get("input_tokens") or 0)
                + (m.get("cache_read_input_tokens") or 0)
                + (m.get("cache_creation_input_tokens") or 0)
                for m in _msg_usage.values()]))

        self.last_session_id = (
            getattr(last_result, "session_id", None)
            or getattr(last_assistant, "session_id", None)
        )
        if self._attempt_usages is not None:
            self._attempt_usages.append(dict(self.last_usage))

    def invoke(
        self, messages: list[dict], *,
        idle_timeout: float | None = None, retry_max: int | None = None,
        resume: str | None = None,
        on_session_start: Any = None,
        on_session_end: Any = None,
        background_watch: Any = None,
    ) -> str:
        """Synchronous wrapper around :meth:`ainvoke`.

        Acquires _lock to serialize concurrent callers of this adapter object
        (each delegation has its own copy(), so they do not contend). Transient API/network
        failures are retried with exponential backoff (see retry_on_transient).

        ``idle_timeout`` / ``retry_max`` override the run-wide stream-idle and
        retry budgets for THIS call only. A short advisory side-call (verdict
        validator) passes a tight idle + ``retry_max=1`` so a hung CLI stream
        aborts in ~that window instead of inheriting a real turn's
        5×600s budget (which once froze a whole run for ~89 min).

        ``resume``: see :meth:`ainvoke`.

        ``on_session_start``: called with no arguments the instant ``_lock``
        is actually acquired -- i.e. when this call's real work begins, not
        when it was merely requested. This is the point at which the call's real work begins,
        distinct from when it was asked to start.
        Best-effort: swallows any exception so a broken callback never
        breaks the real turn.

        ``on_session_end``: called as ``on_session_end(session_id, usage)``
        with THIS call's own session id and token usage while ``_lock`` is
        still held, once the turn is over (also when it raised, with
        whatever it produced: ``None`` / ``{}`` if nothing). Both are
        per-call output; a caller that needs ITS OWN values takes them from
        here rather than from ``last_session_id`` / ``last_usage``. Best-effort
        like ``on_session_start``.

        ``background_watch``: an ``infra.background_jobs.BackgroundJobWatch``.
        It takes its baseline here and its end snapshot inside ``ainvoke``
        just before the CLI is closed, while the CLI's children still live.
        """
        from .base import retry_on_transient
        with self._lock:
            self._background_watch = background_watch
            if background_watch is not None:
                background_watch.start()
            if on_session_start is not None:
                try:
                    on_session_start()
                except Exception:  # noqa: BLE001
                    pass
            # Cleared so a turn that raises before ainvoke() sets them cannot
            # hand its caller the PREVIOUS call's values.
            self.last_session_id = None
            self.last_usage = {}
            self._attempt_usages = []
            try:
                return retry_on_transient(
                    lambda: _run_async_safe(
                        self.ainvoke(
                            messages, idle_timeout=idle_timeout,
                            resume=resume)),
                    max_attempts=retry_max,
                    on_retry=getattr(self, "on_retry", None),
                )
            finally:
                self.last_usage = _combine_attempt_usage(self._attempt_usages)
                self._attempt_usages = None
                if on_session_end is not None:
                    try:
                        on_session_end(
                            self.last_session_id, dict(self.last_usage or {}))
                    except Exception:  # noqa: BLE001
                        pass

HAS_BASE_PROMPT = True class-attribute instance-attribute

base_prompt: str | None = None class-attribute instance-attribute

NATIVE_TOOLS = frozenset({'Bash', 'Edit', 'Read', 'Write', 'Glob', 'Grep', 'BashOutput', 'KillShell', 'Task', 'WebFetch', 'WebSearch'}) class-attribute instance-attribute

model = model instance-attribute

system_prompt = system_prompt instance-attribute

study_dir = Path(study_dir) if study_dir else None instance-attribute

use_default_tools: bool = DEFAULT_TOOLS in (native_tools or []) instance-attribute

native_tools = [t for t in native_tools or [] if t != DEFAULT_TOOLS] instance-attribute

on_init_tools: Any = None instance-attribute

closure_tools = dict(closure_tools or {}) instance-attribute

extra_mcp_servers: dict = dict(extra_mcp_servers or {}) instance-attribute

extra_allowed_tools: list[str] = list(extra_allowed_tools or []) instance-attribute

persistent: bool = persistent instance-attribute

max_history_pairs: int = max_history_pairs instance-attribute

_lock: threading.Lock = threading.Lock() instance-attribute

route_watcher: Any = None instance-attribute

last_usage: dict = {} instance-attribute

_attempt_usages: list[dict] | None = None instance-attribute

last_session_id: str | None = None instance-attribute

_background_watch: Any = None instance-attribute

select_native_tools(agent_tools) -> list[str] classmethod

Pick which of an agent's declared tools are native SDK CLI tools.

Mirror of OpenAICompatibleAdapter.select_native_tools so the runtime can choose native tools generically for any backend (forward-compatible dispatch).

Source code in src/adda/_src/backends/claude.py

@classmethod
def select_native_tools(cls, agent_tools) -> list[str]:
    """Pick which of an agent's declared tools are native SDK CLI tools.

    Mirror of OpenAICompatibleAdapter.select_native_tools so the runtime
    can choose native tools generically for any backend (forward-compatible
    dispatch)."""
    picked = [t for t in agent_tools if t in cls.NATIVE_TOOLS]
    if DEFAULT_TOOLS in agent_tools:
        picked.append(DEFAULT_TOOLS)
    return picked

_system_prompt_option()

What the CLI receives: the text alone (it REPLACES Claude Code's prompt), or, for a node with base_prompt: Default, a preset that keeps Claude Code's prompt and appends the text.

Source code in src/adda/_src/backends/claude.py

def _system_prompt_option(self):
    """What the CLI receives: the text alone (it REPLACES Claude Code's
    prompt), or, for a node with ``base_prompt: Default``, a preset that
    keeps Claude Code's prompt and appends the text."""
    text = self._render_system_prompt()
    if self.base_prompt == DEFAULT_PROMPT:
        return {"type": "preset", "preset": "claude_code", "append": text}
    return text

_render_system_prompt() -> str

The system prompt exactly as the model sees it: base prompt plus the <tools> catalog, with every tool the prose names rewritten to the qualified name the SDK exposes (catalog and prose must agree).

Source code in src/adda/_src/backends/claude.py

def _render_system_prompt(self) -> str:
    """The system prompt exactly as the model sees it: base prompt plus
    the ``<tools>`` catalog, with every tool the prose names rewritten to
    the qualified name the SDK exposes (catalog and prose must agree)."""
    from ..prompts.tool_catalog import (
        qualify_tool_mentions,
        system_prompt_with_catalog,
    )
    qualified = _qualify_closure_names(self.closure_tools)
    rendered = system_prompt_with_catalog(
        self.system_prompt, qualified, builtins_held=self.use_default_tools)
    return qualify_tool_mentions(rendered, {
        bare: f"mcp__{_CLOSURE_MCP_SERVER}__{bare}"
        for bare in self.closure_tools
    })

_compute_allowed_tools(qualified_mcp_tools) -> list[str]

All allowed tool names, ALWAYS as a list (never None).

The SDK does list(options.allowed_tools) when building its command, which raises TypeError on None — so a tool-less agent (e.g. the one-shot problem-statement reviewer) must still get [] here, not None. An empty list correctly means "no tools allowed".

Source code in src/adda/_src/backends/claude.py

def _compute_allowed_tools(self, qualified_mcp_tools) -> list[str]:
    """All allowed tool names, ALWAYS as a list (never None).

    The SDK does ``list(options.allowed_tools)`` when building its command,
    which raises ``TypeError`` on ``None`` — so a tool-less agent (e.g. the
    one-shot problem-statement reviewer) must still get ``[]`` here, not
    ``None``. An empty list correctly means "no tools allowed".
    """
    return (
        list(qualified_mcp_tools)
        + list(self.native_tools)
        + list(self.extra_allowed_tools)
    )

copy() -> ClaudeAdapter

An independent adapter for ONE delegation.

Shares configuration; owns everything a delegation mutates: its own closure_tools (dispatch binds ReportEvals / Write / FollowUp to that delegation's id), its own _lock and its own per-call last_* state. Same-role delegations therefore run concurrently instead of queueing behind one shared lock.

Source code in src/adda/_src/backends/claude.py

def copy(self) -> ClaudeAdapter:
    """An independent adapter for ONE delegation.

    Shares configuration; owns everything a delegation mutates: its own
    ``closure_tools`` (dispatch binds ReportEvals / Write / FollowUp to
    that delegation's id), its own ``_lock`` and its own per-call
    ``last_*`` state. Same-role delegations therefore run concurrently
    instead of queueing behind one shared lock.
    """
    import copy as _copy
    twin = _copy.copy(self)
    twin.closure_tools = dict(self.closure_tools)
    twin.native_tools = list(self.native_tools)
    twin.extra_allowed_tools = list(self.extra_allowed_tools)
    twin.extra_mcp_servers = dict(self.extra_mcp_servers)
    twin._lock = threading.Lock()
    twin.last_usage = {}
    twin._attempt_usages = None
    twin.last_session_id = None
    return twin

ainvoke(messages: list[dict], *, idle_timeout: float | None = None, resume: str | None = None) -> str async

Run one agent turn asynchronously; return assembled text.

idle_timeout overrides the run-wide llm_stream_idle_timeout for THIS call only — used by short advisory side-calls (e.g. the verdict validator) that must not inherit a real agent turn's generous window.

resume (spec 12 item 3): a CLI session id to resume rather than starting fresh -- messages then carries only the NEW turn (the prior conversation is loaded from the resumed session itself, not replayed here). fork_session=False always, so this continues the SAME session rather than branching a copy of it.

Source code in src/adda/_src/backends/claude.py

async def ainvoke(
    self, messages: list[dict], *, idle_timeout: float | None = None,
    resume: str | None = None,
) -> str:
    """Run one agent turn asynchronously; return assembled text.

    ``idle_timeout`` overrides the run-wide ``llm_stream_idle_timeout`` for
    THIS call only — used by short advisory side-calls (e.g. the verdict
    validator) that must not inherit a real agent turn's generous window.

    ``resume`` (spec 12 item 3): a CLI session id to resume rather than
    starting fresh -- ``messages`` then carries only the NEW turn (the
    prior conversation is loaded from the resumed session itself, not
    replayed here). ``fork_session=False`` always, so this continues the
    SAME session rather than branching a copy of it.
    """
    _require_sdk()
    from claude_agent_sdk import (
        AssistantMessage,
        ClaudeAgentOptions,
        ResultMessage,
        SdkMcpTool,
        StreamEvent,
        SystemMessage,
        TextBlock,
        ToolUseBlock,
        UserMessage,
        create_sdk_mcp_server,
        query,
    )

    from ..prompts.tool_catalog import tool_summary

    # Build MCP server from closure_tools if any
    mcp_servers: dict = {}
    qualified_mcp_tools: list[str] = []
    if self.closure_tools:
        server_name = _CLOSURE_MCP_SERVER
        sdk_tools: list[Any] = []
        for tool_name, fn in self.closure_tools.items():
            schema = _infer_schema_from_callable(fn)

            async def _handler(args: dict, bound_fn: Any = fn) -> dict:
                try:
                    result = bound_fn(**args)
                except Exception as exc:
                    return {
                        "content": [
                            {"type": "text", "text": f"ERROR: {exc}"}
                        ],
                        "is_error": True,
                    }
                return {
                    "content": [
                        {
                            "type": "text",
                            "text": (
                                str(result) if result is not None else ""
                            ),
                        }
                    ]
                }

            sdk_tools.append(
                SdkMcpTool(
                    name=tool_name,
                    description=tool_summary(fn, tool_name),
                    input_schema=schema,
                    handler=_handler,
                )
            )

        mcp_cfg = create_sdk_mcp_server(
            name=server_name, tools=sdk_tools or None
        )
        mcp_servers = {server_name: mcp_cfg}
        qualified_mcp_tools = list(_qualify_closure_names(self.closure_tools))

    # Merge external stdio MCP servers declared by the Agent subclass.
    if self.extra_mcp_servers:
        mcp_servers.update(self.extra_mcp_servers)

    _base_disallowed = ["WebSearch", "WebFetch", "Task", "ExitPlanMode"]
    # Under permission_mode="bypassPermissions" the allowed_tools allowlist
    # is NOT enforced — disallowed_tools is the only thing that binds. So a
    # native tool the agent never declared (e.g. Bash/Write for a read-only
    # reviewer) would otherwise be silently usable. Disallow every native
    # tool this agent did not declare, making its declared toolset binding.
    _ungranted_native = [
        t for t in self.NATIVE_TOOLS if t not in self.native_tools
    ]
    if self.use_default_tools:
        # Default has no floor. Only computed additions stay: a built-in
        # that shares its bare name with a closure this node declares
        # (the sandboxed Write replaces the native one) must not run.
        _base_disallowed, _ungranted_native = [], list(self.closure_tools)
    _effective_disallowed = [
        t for t in dict.fromkeys([*_base_disallowed, *_ungranted_native])
        if t not in self.extra_allowed_tools
    ]

    # Non-blocking raw-oracle nudge: a PostToolUse hook that injects a
    # reminder (capped per delegation = per ainvoke) when a Bash/Write
    # call reaches the oracle directly instead of via get_evaluator().
    # Best-effort — if the SDK hook API is unavailable, run without it.
    _hooks = None
    try:
        from claude_agent_sdk import HookMatcher

        from ..runtime import features
        from .base import (
            OracleNudgeBudget,
            get_delegation_id,
            get_run_config_path,
            oracle_registered,
        )
        # Silent until an oracle is registered: pre-registration work (the
        # datagenerator wrapping/validating its raw source) has no
        # get_evaluator() to use, so nudging it is a false positive.
        # The nudge is a monitor intervention; the store notices the same
        # hook carries are store integrity and stay on in every arm.
        _nudge = OracleNudgeBudget(
            enabled=oracle_registered()
            and features.enabled("science_monitor"))
        # Expose on the adapter so the runtime can drain + log its
        # firings as direct evidence (see _record_intervention).
        _nudge.run_config_path = get_run_config_path()
        self._oracle_nudge = _nudge

        # The hook may run on another thread, so bind the delegation and
        # the run's debug dir now, while this thread still holds them.
        from ..infra import pending_notices as _pn
        _rc = get_run_config_path()
        _pn_dir = Path(_rc).parent if _rc else None
        _pn_did = get_delegation_id()

        async def _oracle_hook(input_data, tool_use_id, context):
            msg = _pn.post_tool_context(
                _nudge,
                input_data.get("tool_name", ""),
                input_data.get("tool_input") or {},
                _pn_dir, _pn_did,
            )
            if not msg:
                return {}
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PostToolUse",
                    "additionalContext": msg,
                }
            }

        _hooks = {"PostToolUse": [HookMatcher(hooks=[_oracle_hook])]}
    except Exception:  # noqa: BLE001 — nudge is best-effort, never fatal
        _hooks = None

    # Per-session env: the SDK MERGES this over the inherited environment
    # (PATH etc. preserved), so bare extra keys are safe. See
    # _build_session_env for what is injected and why.
    _sess_env: dict = _build_session_env()

    from ..runtime.settings import get_float
    _max_buf_mb = get_float("llm_max_buffer_mb", 30.0)
    _max_buf = int(_max_buf_mb * 1024 * 1024)

    # The SDK spawns the CLI with cwd=self.study_dir; if that directory
    # doesn't exist the subprocess dies with a cryptic CLIConnectionError
    # ("Working directory does not exist") mid-delegation. Create it
    # defensively so a missing worker workspace can never abort a run.
    if self.study_dir:
        try:
            self.study_dir.mkdir(parents=True, exist_ok=True)
        except Exception:  # noqa: BLE001 — best-effort; spawn surfaces real errors
            pass

    options = ClaudeAgentOptions(
        system_prompt=self._system_prompt_option(),
        model=self.model,
        cwd=str(self.study_dir) if self.study_dir else None,
        tools=({"type": "preset", "preset": "claude_code"}
               if self.use_default_tools else self.native_tools or []),
        mcp_servers=mcp_servers if mcp_servers else {},
        allowed_tools=self._compute_allowed_tools(qualified_mcp_tools),
        disallowed_tools=_effective_disallowed,
        permission_mode="bypassPermissions",
        strict_mcp_config=bool(mcp_servers) or bool(self.extra_mcp_servers),
        # Hermetic session: load NO filesystem settings, so worker/critic
        # subprocesses don't inherit the developer's global ~/.claude hooks
        # (e.g. cbm-code-discovery-gate, which blocked legitimate Read calls
        # for workers AND the critic). Our own hooks are passed
        # programmatically via options.hooks below (audit/#1: fresh hooks).
        setting_sources=[],
        env=_sess_env,
        # Stream-message buffer ceiling. Default 1MB is far too small for a
        # literature reviewer whose tools return full PDFs — a single >1MB
        # MCP tool result overflowed it and FATALLY (non-retried) killed the
        # whole delegation (D003). 30MB clears realistic PDFs; non-PDF
        # results never approach it. A result still exceeding this is caught
        # gracefully below (turn cut short + marker), not a fatal crash.
        # Tune via F3DASM_LLM_MAX_BUFFER_MB.
        max_buffer_size=_max_buf,
        # Partial streaming → a fine-grained heartbeat: the stream emits a
        # StreamEvent sub-second while genuinely generating, so total
        # silence becomes a reliable stall signal and the
        # idle timeout can be bounded without false-positiving a
        # slow-but-working generation.
        include_partial_messages=True,
        **_thinking_options(self.model),
        **({"hooks": _hooks} if _hooks else {}),
        **(
            {"resume": resume, "fork_session": False}
            if resume is not None else {}
        ),
    )

    prompt_str = _format_messages_as_prompt(messages)

    last_assistant = None
    last_result: Any = None
    _buffer_overflowed = False
    _deliberate_break = False
    _msg_usage: dict[str, dict] = {}
    _cur_msg_id: list[str | None] = [None]
    raise_if_stopped("model call")
    gen = query(prompt=prompt_str, options=options)
    # Idle-stream timeout — turns a silent stream into a retryable
    # TimeoutError. Resets on EVERY stream message. Scoped to model
    # generation: while a tool runs the window stands down (see _phase).
    # MEASURED (the stream_evt instrumentation settled this): the bundled
    # CLI forwards NO ping events — only message lifecycle — so during a
    # mid-generation pause we get ZERO liveness signal. Two supercompressible
    # (Sonnet) runs captured LEGITIMATE (recovered) active-generation gaps of
    # 53.6s AND 252-256s (one mid-tool_use-block composition, one
    # tool_result→next message_start delay). So earlier 60s/180s windows
    # would guillotine legitimate slow generations. The observed ~256s legit
    # gap left only ~44s of headroom under the prior 300s default, and with
    # NO ping liveness signal a longer legit generation is indistinguishable
    # from a stall — so the default is 600s: ample margin over the measured
    # legit gaps while still catching a truly dead (silent-forever) stream in
    # ~10 min. A genuine runaway is the delegation watchdog's concern, not
    # this window's. Knobs (config.yaml runtime block; env overrides):
    # llm_stream_idle_timeout (0 disables); llm_tool_idle_timeout caps tool
    # execution (0 = uncapped).
    from ..runtime.settings import get_float as _get_float
    _idle = (idle_timeout if idle_timeout is not None
             else _get_float("llm_stream_idle_timeout", 600.0))
    _tool_idle = _get_float("llm_tool_idle_timeout", 0.0)

    def _phase(msg: Any):
        # True: a tool is now executing → suspend the tight idle window.
        # False: generation active / tool result returned → tight window.
        # None: leave phase unchanged.
        if isinstance(msg, AssistantMessage):
            return any(
                isinstance(b, ToolUseBlock) for b in msg.content
            )
        if isinstance(msg, StreamEvent):
            # Only message_start opens a new generation. The SDK emits
            # content_block_stop / message_delta / message_stop AFTER the
            # AssistantMessage that carries a ToolUseBlock, while the tool
            # is still running; treating those as "generation resumed"
            # re-armed the stall window mid-tool and killed a 91-minute
            # delegation that was blocked in a long tool call.
            ev = getattr(msg, "event", None)
            if isinstance(ev, dict) and ev.get("type") == "message_start":
                return False
            return None
        if isinstance(msg, UserMessage):
            return False
        return None

    from .base import append_transcript, debug_enabled

    def _record(msg: Any):
        # Full reasoning + tool-calls + tool-results; StreamEvent partials
        # are skipped (the assembled AssistantMessage carries the text).
        if isinstance(msg, AssistantMessage):
            texts, tools, thinking = [], [], []
            thinking_omitted = 0
            for b in msg.content:
                if isinstance(b, TextBlock):
                    texts.append(b.text)
                elif isinstance(b, ToolUseBlock):
                    tools.append({"name": b.name, "input": b.input,
                                  "tool_use_id": getattr(b, "id", None)})
                else:
                    t = (getattr(b, "thinking", None)
                         or getattr(b, "text", None))
                    if t:
                        thinking.append(t)
                    elif hasattr(b, "thinking"):
                        # A thinking block that arrived with no text
                        # (display "omitted"): counted, so "the model
                        # thought but it was hidden" is distinguishable
                        # from "the model did not think".
                        thinking_omitted += 1
            return {"type": "assistant", "text": "".join(texts),
                    "tools": tools, "thinking": thinking,
                    "thinking_omitted": thinking_omitted,
                    "message_id": getattr(msg, "message_id", None)}
        if isinstance(msg, UserMessage):
            results = []
            for b in (getattr(msg, "content", None) or []):
                results.append({
                    "tool_use_id": getattr(b, "tool_use_id", None),
                    "content": getattr(b, "content", b),
                })
            return {"type": "tool_result", "results": results}
        if isinstance(msg, ResultMessage):
            return {"type": "result",
                    "usage": getattr(msg, "usage", None),
                    "model_usage": getattr(msg, "model_usage", None),
                    "cost_usd": getattr(msg, "total_cost_usd", None)}
        if isinstance(msg, SystemMessage):
            # SystemMessage used to be invisible here entirely (this
            # function returned None for anything it didn't recognize) —
            # so a real compact_boundary (the SDK's own context-
            # compaction event) left NO trace in debug/transcripts/,
            # the viewer, or any post-run analysis. Confirmed empirically
            # (a short forced-window test): a single short session
            # produced 50-119 SystemMessages, the overwhelming majority
            # subtype "thinking_tokens" — a per-token streaming heartbeat,
            # pure noise at transcript granularity (stream_evt already
            # covers liveness). Record everything ELSE verbatim,
            # including (especially) compact_boundary's own metadata.
            if msg.subtype in _SYSTEM_MESSAGE_NOISE_SUBTYPES:
                return None
            return {"type": "system", "subtype": msg.subtype,
                    "data": msg.data}
        return None

    _capture = debug_enabled()
    # Partial-stream checkpointing: an agent that never completes a message
    # (infinite thinking / an unresolved turn) emits only StreamEvents and
    # would otherwise disclose nothing. Buffer the deltas and flush a
    # "partial" record every N events (and on any stream teardown) so the
    # transcript reveals what a stuck turn is doing in near-real-time.
    _PARTIAL_FLUSH_EVERY = 25
    _pbuf: list[str] = []
    _pcount = [0]

    def _extract_delta(ev: Any) -> str:
        if not isinstance(ev, dict):
            return ""
        d = ev.get("delta") or {}
        return (d.get("text") or d.get("thinking")
                or d.get("partial_json") or "")

    def _flush_partial() -> None:
        if _pbuf:
            append_transcript({"type": "partial", "text": "".join(_pbuf),
                               "events": _pcount[0]})
            _pbuf.clear()

    try:
        _stream = (
            _stream_with_idle_timeout(
                gen, _idle,
                tool_timeout=_tool_idle, classify=_phase,
            )
            if _idle > 0 else gen
        )
        # Measurement clock: the first event's gap below = time-to-first
        # stream event (≈ prefill latency).
        _last_evt = [time.monotonic()]
        async for msg in _stream:
            raise_if_stopped("model call")
            if isinstance(msg, StreamEvent):
                _track_message_usage(
                    getattr(msg, "event", None), _msg_usage, _cur_msg_id)
            elif isinstance(msg, AssistantMessage) \
                    and getattr(msg, "usage", None) \
                    and getattr(msg, "message_id", None):
                _msg_usage.setdefault(msg.message_id, dict(msg.usage))
            if _capture:
                if isinstance(msg, StreamEvent):
                    _pcount[0] += 1
                    _ev = getattr(msg, "event", {}) or {}
                    _et = _ev.get("type", "?") if isinstance(
                        _ev, dict) else "?"
                    _now = time.monotonic()
                    _gap = _now - _last_evt[0]
                    _last_evt[0] = _now
                    # Record every NON-delta event (ping, message_start/
                    # stop, content_block_start/stop, message_delta) and any
                    # delta after a >2s pause — so we can see whether the
                    # stream stays alive (pings) during silent/prefill
                    # phases and the true inter-event gap distribution.
                    # This is what settles whether a 60s silence is a dead
                    # stream or a legitimately-slow first token.
                    if _et != "content_block_delta" or _gap > 2.0:
                        append_transcript({
                            "type": "stream_evt", "evt": _et,
                            "gap_s": round(_gap, 2),
                            **_usage_fields(_ev)})
                    _d = _extract_delta(_ev)
                    if _d:
                        _pbuf.append(_d)
                    if _pcount[0] % _PARTIAL_FLUSH_EVERY == 0:
                        _flush_partial()
                else:
                    # A complete message: flush any buffered partial first,
                    # then the structured record.
                    _flush_partial()
                    _rec = _record(msg)
                    if _rec is not None:
                        append_transcript(_rec)
            if (isinstance(msg, SystemMessage) and msg.subtype == "init"
                    and self.on_init_tools is not None):
                # Unconditional like the compaction record below: which
                # tools the CLI really loaded is a run-level fact.
                try:
                    self.on_init_tools(list((msg.data or {}).get("tools") or []))
                except Exception:  # noqa: BLE001 — a notice never fails a turn
                    pass
            if isinstance(msg, SystemMessage) and msg.subtype == "compact_boundary":
                # Unconditional (not gated on _capture/debug mode): a
                # compaction is a run-level fact an analyst should never
                # have to enable debug transcripts to discover.
                record_stream_diagnostic(
                    "CONTEXT_COMPACTED",
                    "The SDK compacted this session's context "
                    "mid-turn (compact_boundary).",
                    compaction_data=msg.data,
                )
            if isinstance(msg, AssistantMessage):
                last_assistant = msg
                if self.route_watcher and self.route_watcher():
                    _deliberate_break = True
                    break
            elif isinstance(msg, ResultMessage):
                last_result = msg
                break
    except Exception as exc:  # noqa: BLE001
        _m = str(exc).lower()
        if "maximum buffer size" in _m or "exceeded maximum" in _m:
            # Graceful contour: a single tool result (e.g. a huge PDF)
            # overflowed the stream buffer. The old behavior was a fatal,
            # NON-retried crash that killed the whole delegation (D003).
            # Instead, end the turn with what we have + a marker so the
            # agent can retry the fetch smaller — the delegation survives.
            _buffer_overflowed = True
        else:
            raise
    finally:
        if _capture:
            _flush_partial()  # disclose a stuck/torn-down turn's tail
        self._settle_usage(last_result, _msg_usage, last_assistant)
        _watch = getattr(self, "_background_watch", None)
        if _watch is not None:
            # Depth 2: the CLI itself (depth 1) is not a background job.
            _watch.end(min_depth=2)
        aclose = getattr(gen, "aclose", None)
        if aclose:
            try:
                await aclose()
            except Exception:
                pass

    # A stream that ends with neither a ResultMessage NOR a deliberate
    # route_watcher break, and wasn't already explained by a buffer
    # overflow, is abnormal: the CLI session ended (or the SDK's async
    # generator was exhausted) without ever completing its turn — report
    # 7 (run 20260830T004106, Oscar): the last AssistantMessage carried
    # only a tool call (e.g. Bash/TaskOutput), no result ever arrived,
    # and the near-empty `text` this then returns reads as a malformed
    # report to _invoke_with_report_retry, which silently re-invokes
    # with the ORIGINAL task — restarting the delegation from scratch,
    # orphaning whatever the first attempt launched, with nothing
    # logged anywhere. This does not change that return behaviour (a
    # future decision, pending Elvis) — it only makes the fact visible.
    if last_result is None and not _deliberate_break and not _buffer_overflowed:
        _last_tool = None
        if last_assistant is not None:
            for _b in last_assistant.content:
                if isinstance(_b, ToolUseBlock):
                    _last_tool = _b.name
        record_stream_diagnostic(
            "STREAM_ENDED_WITHOUT_RESULT",
            "CLI stream ended without a ResultMessage or a deliberate "
            "route break — the turn may not have completed; its "
            "returned text can look like a malformed report and "
            "trigger a silent report-retry.",
            last_tool_in_flight=_last_tool,
            cli_session_id=getattr(last_assistant, "session_id", None),
        )

    text = ""
    if last_assistant is not None:
        for block in last_assistant.content:
            if isinstance(block, TextBlock):
                text += block.text
    if _buffer_overflowed:
        _note = (
            f"[STREAM NOTE: a tool returned more than {_max_buf_mb:.0f} MB "
            "in a single result and overflowed the message buffer; that "
            "result was dropped and this turn was cut short (the delegation "
            "did NOT crash). Re-run the tool with a smaller/narrower request "
            "— fewer items, or a summary/extract instead of full text.]"
        )
        text = (text + "\n\n" + _note) if text else _note
    return text

_settle_usage(last_result: Any, _msg_usage: dict, last_assistant: Any) -> None

Record this attempt's usage and session id. Runs from ainvoke's finally so a stream that RAISES (idle TimeoutError, API error) still reports the usage it had streamed; invoke sums the attempts.

Source code in src/adda/_src/backends/claude.py

def _settle_usage(self, last_result: Any, _msg_usage: dict,
                  last_assistant: Any) -> None:
    """Record this attempt's usage and session id. Runs from ainvoke's
    ``finally`` so a stream that RAISES (idle TimeoutError, API error)
    still reports the usage it had streamed; ``invoke`` sums the
    attempts."""
    # Capture token usage from ResultMessage for run-level accounting.
    if last_result is not None:
        self.last_usage = {
            **(last_result.usage or {}),
            **_cache_split(last_result.usage or {}),
            "total_cost_usd": last_result.total_cost_usd,
        }
    elif _msg_usage:
        # route_watcher broke the stream on the AssistantMessage that
        # closes the run (Done()) before the SDK's ResultMessage -- the
        # only carrier of the session's cumulative usage and cost -- ever
        # arrived. The strategizer's whole run is ONE long stream, so
        # taking just the last message's usage recorded ~1 output token
        # for the entire run (run 20260928T141126). Sum the per-API-call
        # usage the stream itself reported instead. total_cost_usd stays
        # None: cost is a session-level rollup no single message carries,
        # the same "unknown, not zero" convention openai_compatible uses.
        self.last_usage = {
            **_sum_message_usage(_msg_usage.values()),
            "total_cost_usd": None,
        }
    elif last_assistant is not None and getattr(last_assistant, "usage", None):
        # A backend/stream that never sent message_start/delta events or a
        # message id: the lone message's own (possibly stale) usage.
        self.last_usage = {
            **last_assistant.usage, "total_cost_usd": None,
        }
    else:
        self.last_usage = {}

    # Claude reports input_tokens WITHOUT cache tokens, so each count maps
    # one to one onto the telemetry schema.
    u = self.last_usage
    self.last_usage = {**u, **normalized_usage(
        fresh_input=u.get("input_tokens"),
        cache_read=u.get("cache_read_input_tokens"),
        cache_write=u.get("cache_creation_input_tokens"),
        output=u.get("output_tokens"))}
    if _msg_usage:
        # Claude's input_tokens excludes cache, so a call's whole prompt
        # is the three input counts together.
        self.last_usage.update(call_shape([
            (m.get("input_tokens") or 0)
            + (m.get("cache_read_input_tokens") or 0)
            + (m.get("cache_creation_input_tokens") or 0)
            for m in _msg_usage.values()]))

    self.last_session_id = (
        getattr(last_result, "session_id", None)
        or getattr(last_assistant, "session_id", None)
    )
    if self._attempt_usages is not None:
        self._attempt_usages.append(dict(self.last_usage))

invoke(messages: list[dict], *, idle_timeout: float | None = None, retry_max: int | None = None, resume: str | None = None, on_session_start: Any = None, on_session_end: Any = None, background_watch: Any = None) -> str

Synchronous wrapper around :meth:ainvoke.

Acquires _lock to serialize concurrent callers of this adapter object (each delegation has its own copy(), so they do not contend). Transient API/network failures are retried with exponential backoff (see retry_on_transient).

idle_timeout / retry_max override the run-wide stream-idle and retry budgets for THIS call only. A short advisory side-call (verdict validator) passes a tight idle + retry_max=1 so a hung CLI stream aborts in ~that window instead of inheriting a real turn's 5×600s budget (which once froze a whole run for ~89 min).

resume: see :meth:ainvoke.

on_session_start: called with no arguments the instant _lock is actually acquired -- i.e. when this call's real work begins, not when it was merely requested. This is the point at which the call's real work begins, distinct from when it was asked to start. Best-effort: swallows any exception so a broken callback never breaks the real turn.

on_session_end: called as on_session_end(session_id, usage) with THIS call's own session id and token usage while _lock is still held, once the turn is over (also when it raised, with whatever it produced: None / {} if nothing). Both are per-call output; a caller that needs ITS OWN values takes them from here rather than from last_session_id / last_usage. Best-effort like on_session_start.

background_watch: an infra.background_jobs.BackgroundJobWatch. It takes its baseline here and its end snapshot inside ainvoke just before the CLI is closed, while the CLI's children still live.

Source code in src/adda/_src/backends/claude.py

def invoke(
    self, messages: list[dict], *,
    idle_timeout: float | None = None, retry_max: int | None = None,
    resume: str | None = None,
    on_session_start: Any = None,
    on_session_end: Any = None,
    background_watch: Any = None,
) -> str:
    """Synchronous wrapper around :meth:`ainvoke`.

    Acquires _lock to serialize concurrent callers of this adapter object
    (each delegation has its own copy(), so they do not contend). Transient API/network
    failures are retried with exponential backoff (see retry_on_transient).

    ``idle_timeout`` / ``retry_max`` override the run-wide stream-idle and
    retry budgets for THIS call only. A short advisory side-call (verdict
    validator) passes a tight idle + ``retry_max=1`` so a hung CLI stream
    aborts in ~that window instead of inheriting a real turn's
    5×600s budget (which once froze a whole run for ~89 min).

    ``resume``: see :meth:`ainvoke`.

    ``on_session_start``: called with no arguments the instant ``_lock``
    is actually acquired -- i.e. when this call's real work begins, not
    when it was merely requested. This is the point at which the call's real work begins,
    distinct from when it was asked to start.
    Best-effort: swallows any exception so a broken callback never
    breaks the real turn.

    ``on_session_end``: called as ``on_session_end(session_id, usage)``
    with THIS call's own session id and token usage while ``_lock`` is
    still held, once the turn is over (also when it raised, with
    whatever it produced: ``None`` / ``{}`` if nothing). Both are
    per-call output; a caller that needs ITS OWN values takes them from
    here rather than from ``last_session_id`` / ``last_usage``. Best-effort
    like ``on_session_start``.

    ``background_watch``: an ``infra.background_jobs.BackgroundJobWatch``.
    It takes its baseline here and its end snapshot inside ``ainvoke``
    just before the CLI is closed, while the CLI's children still live.
    """
    from .base import retry_on_transient
    with self._lock:
        self._background_watch = background_watch
        if background_watch is not None:
            background_watch.start()
        if on_session_start is not None:
            try:
                on_session_start()
            except Exception:  # noqa: BLE001
                pass
        # Cleared so a turn that raises before ainvoke() sets them cannot
        # hand its caller the PREVIOUS call's values.
        self.last_session_id = None
        self.last_usage = {}
        self._attempt_usages = []
        try:
            return retry_on_transient(
                lambda: _run_async_safe(
                    self.ainvoke(
                        messages, idle_timeout=idle_timeout,
                        resume=resume)),
                max_attempts=retry_max,
                on_retry=getattr(self, "on_retry", None),
            )
        finally:
            self.last_usage = _combine_attempt_usage(self._attempt_usages)
            self._attempt_usages = None
            if on_session_end is not None:
                try:
                    on_session_end(
                        self.last_session_id, dict(self.last_usage or {}))
                except Exception:  # noqa: BLE001
                    pass

adda.OllamaAdapter

Adapter for Ollama-served open-weight models.

Uses ChatOpenAI pointed at Ollama's local OpenAI-compatible endpoint (default http://localhost:11434/v1, overridable via base_url in config.yaml). Ollama needs no real auth, so the API key is the conventional placeholder "local".

Source code in src/adda/_src/backends/ollama.py

class OllamaAdapter(OpenAICompatibleAdapter):
    """Adapter for Ollama-served open-weight models.

    Uses ChatOpenAI pointed at Ollama's local OpenAI-compatible endpoint
    (default ``http://localhost:11434/v1``, overridable via ``base_url`` in
    config.yaml). Ollama needs no real auth, so the API key is
    the conventional placeholder ``"local"``.
    """

    DEFAULT_BASE_URL = "http://localhost:11434/v1"
    API_KEY = "local"
    API_KEY_ENV = None

    def _probe_context_window(self) -> int | None:
        """Ollama's SERVED window, which is not the model's trained length.

        ``/v1/models`` reports neither, so this uses Ollama's native
        ``/api/show``. The distinction matters more here than anywhere: a
        Modelfile\'s ``num_ctx`` is what the server will actually accept, and
        Ollama\'s default is far below what any of these models were trained
        for. Taking ``context_length`` from ``model_info`` would report the
        trained window, sail past the real limit, and produce exactly the
        server-side truncation this trimming exists to prevent — so the
        parameter wins and the trained length is only a fallback.
        """
        import re

        import requests
        root = self._base_url.rstrip("/")
        if root.endswith("/v1"):
            root = root[: -len("/v1")]
        try:
            r = requests.post(f"{root}/api/show",
                              json={"model": self.model}, timeout=5)
            r.raise_for_status()
            body = r.json()
        except Exception:
            return None
        # `parameters` is the Modelfile block, e.g. "num_ctx   262144"
        params = body.get("parameters")
        if isinstance(params, str):
            m = re.search(r"^\s*num_ctx\s+(\d+)", params, re.M)
            if m:
                return int(m.group(1))
        info = body.get("model_info")
        if isinstance(info, dict):
            for key, val in info.items():
                if key.endswith(".context_length") and isinstance(val, int):
                    return val
        return None

model = model instance-attribute

system_prompt = system_prompt instance-attribute

study_dir = Path(study_dir) if study_dir else None instance-attribute

native_tools: list[str] = list(native_tools or []) instance-attribute

use_default_tools: bool = False instance-attribute

on_init_tools: Any = None instance-attribute

closure_tools: dict[str, Any] = dict(closure_tools or {}) instance-attribute

_base_url = base_url or self.DEFAULT_BASE_URL instance-attribute

_api_key = api_key if api_key is not None else self.API_KEY instance-attribute

extra_mcp_servers: dict = dict(extra_mcp_servers or {}) instance-attribute

extra_allowed_tools: list[str] = list(extra_allowed_tools or []) instance-attribute

persistent: bool = persistent instance-attribute

max_history_pairs: int = max_history_pairs instance-attribute

_lock: threading.Lock = threading.Lock() instance-attribute

_summary_cache: dict[str, str] = {} instance-attribute

_agent: Any = None instance-attribute

route_watcher: Any = None instance-attribute

_oracle_nudge = OracleNudgeBudget() instance-attribute

_notice_ctx: tuple = (None, None) instance-attribute

last_usage: dict = {} instance-attribute

last_session_id: str | None = None instance-attribute

DEFAULT_BASE_URL = 'http://localhost:11434/v1' class-attribute instance-attribute

API_KEY = 'local' class-attribute instance-attribute

API_KEY_ENV = None class-attribute instance-attribute

select_native_tools(agent_tools) -> list[str] classmethod

Pick which of an agent's declared tools are NATIVE (CLI) tools.

OpenAI-compatible backends run every tool through LangChain, so a tool is native unless it is one of the Python closure tools the node injects separately. (ClaudeAdapter overrides this with its own CLI-tool set.)

Source code in src/adda/_src/backends/openai_compatible.py

@classmethod
def select_native_tools(cls, agent_tools) -> list[str]:
    """Pick which of an agent's declared tools are NATIVE (CLI) tools.

    OpenAI-compatible backends run every tool through LangChain, so a tool
    is native unless it is one of the Python closure tools the node injects
    separately. (ClaudeAdapter overrides this with its own CLI-tool set.)
    """
    picked = [t for t in agent_tools
              if t not in _CLOSURE_TOOL_NAMES and t != DEFAULT_TOOLS]
    if DEFAULT_TOOLS in agent_tools:
        picked += [t for t in DEFAULT_NATIVE_TOOLS if t not in picked]
    return picked

copy() -> OpenAICompatibleAdapter

An independent adapter for ONE delegation (see ClaudeAdapter.copy): own closure_tools, _lock, built agent, oracle-nudge budget, summary cache and per-call last_* state.

Source code in src/adda/_src/backends/openai_compatible.py

def copy(self) -> OpenAICompatibleAdapter:
    """An independent adapter for ONE delegation (see ClaudeAdapter.copy):
    own ``closure_tools``, ``_lock``, built agent, oracle-nudge budget,
    summary cache and per-call ``last_*`` state."""
    import copy as _copy

    from .base import OracleNudgeBudget
    twin = _copy.copy(self)
    twin.closure_tools = dict(self.closure_tools)
    twin.native_tools = list(self.native_tools)
    twin.extra_allowed_tools = list(self.extra_allowed_tools)
    twin.extra_mcp_servers = dict(self.extra_mcp_servers)
    twin._lock = threading.Lock()
    twin._summary_cache = {}
    twin._agent = None
    twin._oracle_nudge = OracleNudgeBudget()
    twin._notice_ctx = (None, None)
    twin.last_usage = {}
    twin.last_session_id = None
    return twin

_post_tool_context(tool_name: str, tool_input: dict) -> str | None

Raw-oracle nudge plus queued campaign notices, same text the Claude backend's post-tool hook returns. Tool closures run on other threads, so the delegation context is the one bound at invoke time.

Source code in src/adda/_src/backends/openai_compatible.py

def _post_tool_context(self, tool_name: str, tool_input: dict) -> str | None:
    """Raw-oracle nudge plus queued campaign notices, same text the Claude
    backend's post-tool hook returns. Tool closures run on other threads,
    so the delegation context is the one bound at invoke time."""
    from ..infra.pending_notices import post_tool_context
    from ..runtime import features
    debug_dir, did = self._notice_ctx
    nudge = (self._oracle_nudge
             if features.enabled("science_monitor") else None)
    return post_tool_context(
        nudge, tool_name, tool_input, debug_dir, did)

_record_native_error(tool_name: str, message: str, args: dict) -> None

Write a native tool's error result as an ERROR_RETURN row.

Source code in src/adda/_src/backends/openai_compatible.py

def _record_native_error(self, tool_name: str, message: str,
                         args: dict) -> None:
    """Write a native tool's error result as an ERROR_RETURN row."""
    from .base import append_diagnostic
    debug_dir, did = self._notice_ctx
    if debug_dir is None:
        return
    append_diagnostic(debug_dir, did or "", "ERROR_RETURN", message,
                      tool=tool_name, args=args, fault="agent")

_build_tools() -> list[Any]

Source code in src/adda/_src/backends/openai_compatible.py

def _build_tools(self) -> list[Any]:
    import functools

    from langchain_core.tools import StructuredTool

    from ..prompts.tool_catalog import tool_summary
    native_map = _native_tool_map(
        self.study_dir, self._post_tool_context, self._record_native_error,
        # A callable, not thread-local state: LangGraph runs tools on its
        # own threads. _notice_ctx is set per invoke, on the invoking thread.
        delegation_id=lambda: self._notice_ctx[1])
    tools: list[Any] = [
        native_map[name]
        for name in self.native_tools
        if name in native_map
    ]
    # _agent (and this tool list) is built once and cached forever
    # (see _invoke_once below) -- copy() deliberately returns self, so
    # every delegation to the same worker shares this one instance.
    # nodes/tools/routing.py reassigns closure_tools["Write"] to a
    # freshly-sandboxed closure before EACH delegation, expecting that
    # to take effect live. StructuredTool.from_function(fn, ...) below
    # would otherwise bake in whichever callable `fn` was THE FIRST
    # time this method ran, silently freezing every later delegation
    # to the first one's sandbox (confirmed for real: MathExpert on
    # Ollama, 3 separate _sandboxed_write ERROR_RETURNs all attributing
    # later delegations' writes to the first delegation's own folder —
    # BACKLOG #29). Indirect through a live dict lookup instead, so a
    # later closure_tools[name] reassignment is honoured immediately.
    for name, fn in self.closure_tools.items():
        def _dispatch(*args, _name=name, **kwargs):
            return self.closure_tools[_name](*args, **kwargs)
        _dispatch = functools.wraps(fn)(_dispatch)
        # The one-line summary, not the docstring: the full text is
        # already in the system prompt's <tools> section, and
        # from_function's default would send it a second time.
        tools.append(StructuredTool.from_function(
            _dispatch, name=name, description=tool_summary(fn, name)))
    # Inject MCP-equivalent tools for declared extra_allowed_tools.
    if self.extra_allowed_tools:
        lit_tools = _make_literature_tools()
        allowed = set(self.extra_allowed_tools)
        tools += [_guard_native(t, self._record_native_error) for t in lit_tools
                  if t.name in allowed]
    return tools

_build_agent() -> Any

Source code in src/adda/_src/backends/openai_compatible.py

def _build_agent(self) -> Any:
    from langchain_core.messages import SystemMessage
    from langchain_openai import ChatOpenAI
    from langgraph.prebuilt import create_react_agent

    from ..prompts.tool_catalog import system_prompt_with_catalog
    max_output = self._resolve_max_output_tokens()
    llm = ChatOpenAI(
        model=self.model, base_url=self._base_url, api_key=self._api_key,
        max_tokens=max_output,
    )
    system = system_prompt_with_catalog(
        self.system_prompt, self.closure_tools)
    return create_react_agent(
        llm,
        self._build_tools(),
        prompt=SystemMessage(content=system),
        pre_model_hook=self._context_hook(system),
    )

_resolve_context_window() -> tuple[int, str]

(window, source) — explicit setting, then the server, then a declared default.

The source is returned, not just the number, because a run whose window came from a guess and a run whose window came from the server are not the same run. settings.py's precedence applies to the explicit channel, so a study can pin it and an ablation can sweep it.

Source code in src/adda/_src/backends/openai_compatible.py

def _resolve_context_window(self) -> tuple[int, str]:
    """``(window, source)`` — explicit setting, then the server, then a
    declared default.

    The source is returned, not just the number, because a run whose window
    came from a guess and a run whose window came from the server are not
    the same run. ``settings.py``'s precedence applies to the explicit
    channel, so a study can pin it and an ablation can sweep it.
    """
    cached = getattr(self, "_ctx_window", None)
    if cached is not None:
        return cached
    from ..runtime import settings
    explicit = settings.get_int("context_window", 0)
    if explicit > 0:
        out = (explicit, "setting")
    else:
        probed = self._probe_context_window()
        out = ((probed, "server") if probed
               else (context_budget.DEFAULT_CONTEXT_WINDOW, "default"))
    self._ctx_window = out
    return out

_resolve_max_output_tokens() -> int | None

Cap on ONE reply, or None for deliberately uncapped.

This is NOT part of the context_trim feature, and that is on purpose. Trimming decides what the model SEES and is scaffolding whose value is an open question; this bounds what the server will DO and is a safety limit. Gating it on the same knob would mean the arm that answers "is trimming worth it" also removes the only thing stopping a turn generating for an hour, which is not the question being asked. A constant applied to both arms is not a confound.

The resolved number is logged, so a run whose reply was cut short is distinguishable from a run whose model simply stopped.

Source code in src/adda/_src/backends/openai_compatible.py

def _resolve_max_output_tokens(self) -> int | None:
    """Cap on ONE reply, or None for deliberately uncapped.

    This is NOT part of the ``context_trim`` feature, and that is on
    purpose. Trimming decides what the model SEES and is scaffolding whose
    value is an open question; this bounds what the server will DO and is
    a safety limit. Gating it on the same knob would mean the arm that
    answers "is trimming worth it" also removes the only thing stopping a
    turn generating for an hour, which is not the question being asked. A
    constant applied to both arms is not a confound.

    The resolved number is logged, so a run whose reply was cut short is
    distinguishable from a run whose model simply stopped.
    """
    from ..runtime import settings
    window, source = self._resolve_context_window()
    explicit = settings.get_int("max_output_tokens", 0)
    cap = context_budget.resolve_max_output_tokens(window, explicit)
    log.info(
        "max output tokens: %s (window %d from %s, setting %d)",
        cap if cap is not None else "uncapped", window, source, explicit,
    )
    return cap

_summarize(prompt: str) -> str

One summary, from the model this run is already using.

Deliberately not a second, smaller model. A fixed summariser would remove one confound (summary quality no longer varies with the arm) and introduce two: a dependency the study does not otherwise have, and a capability the run itself never had. The interesting hypothesis on cheap models is that scaffolding compensates for capability, and a 27B run whose context is curated by a frontier model is not testing that.

Goes straight to the chat model, NOT through the agent: a summary produced by a tool-loop could call tools, and a pre_model_hook that re-enters the agent is a recursion, not a hook.

Source code in src/adda/_src/backends/openai_compatible.py

def _summarize(self, prompt: str) -> str:
    """One summary, from the model this run is already using.

    Deliberately not a second, smaller model. A fixed summariser would
    remove one confound (summary quality no longer varies with the arm)
    and introduce two: a dependency the study does not otherwise have, and
    a capability the run itself never had. The interesting hypothesis on
    cheap models is that scaffolding compensates for capability, and a
    27B run whose context is curated by a frontier model is not testing
    that.

    Goes straight to the chat model, NOT through the agent: a summary
    produced by a tool-loop could call tools, and a pre_model_hook that
    re-enters the agent is a recursion, not a hook.
    """
    from langchain_core.messages import HumanMessage
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(
        model=self.model, base_url=self._base_url, api_key=self._api_key,
        max_tokens=self._resolve_max_output_tokens(),
    )
    return str(llm.invoke([HumanMessage(content=prompt)]).content or "")

_context_hook(system_prompt: str)

A pre_model_hook that keeps one turn inside the served window.

context_policy picks HOW. Both values manage the context; neither is "off", because an unmanaged context is not an experimental arm — it is the crash this subsystem was written to stop (Ollama evicts the original user turn and its renderer then rejects the request with 500 no user query found in messages).

compact (the default) replaces the middle of the conversation with a summary, so a delegation's RESULT survives even when its prose does not. trim drops those messages instead: free, deterministic, and blind to what it is throwing away. The trade is fidelity against determinism — a summary can quietly restate a number, a dropped message obviously cannot — and it is recorded per run so an analysis can condition on it rather than assume it.

The hook returns llm_input_messages, which changes only what is SENT. Graph state keeps every message, so the transcript on disk stays complete and a compacted run is still fully auditable.

Source code in src/adda/_src/backends/openai_compatible.py

def _context_hook(self, system_prompt: str):
    """A ``pre_model_hook`` that keeps one turn inside the served window.

    ``context_policy`` picks HOW. Both values manage the context; neither
    is "off", because an unmanaged context is not an experimental arm —
    it is the crash this subsystem was written to stop (Ollama evicts the
    original user turn and its renderer then rejects the request with
    ``500 no user query found in messages``).

    ``compact`` (the default) replaces the middle of the conversation with
    a summary, so a delegation's RESULT survives even when its prose does
    not. ``trim`` drops those messages instead: free, deterministic, and
    blind to what it is throwing away. The trade is fidelity against
    determinism — a summary can quietly restate a number, a dropped
    message obviously cannot — and it is recorded per run so an analysis
    can condition on it rather than assume it.

    The hook returns ``llm_input_messages``, which changes only what is
    SENT. Graph state keeps every message, so the transcript on disk stays
    complete and a compacted run is still fully auditable.
    """
    window, source = self._resolve_context_window()
    reserve = context_budget.estimate_tokens(system_prompt)
    policy = self._context_policy()

    seen: list = [None]

    def _hook(state):
        msgs = (state or {}).get("messages") or []
        if policy == "compact":
            kept, report = context_compaction.compact_to_budget(
                msgs, context_window=window, reserve_tokens=reserve,
                summarize=self._summarize, cache=self._summary_cache)
        else:
            kept, report = context_budget.trim_to_budget(
                msgs, context_window=window, reserve_tokens=reserve)
        # The hook re-runs on every model call and graph state keeps the
        # full history, so an unchanged compaction re-fires each turn.
        # Record a compaction once, and again only when it moves.
        marker = (report.dropped, report.truncated)
        if report.fired and marker != seen[0]:
            seen[0] = marker
            from .base import (
                append_transcript,
                debug_enabled,
                record_stream_diagnostic,
            )
            log.warning(
                "context %s fired: %d message(s) removed, %d truncated "
                "(%d -> %d est. tokens, budget %d, window %d from %s)",
                policy, report.dropped, report.truncated, report.before,
                report.after, report.budget, window, source,
            )
            summary = ""
            if policy == "compact":
                header = context_compaction.summary_header(report.dropped)
                for m in kept:
                    text = context_budget._text_of(m)
                    if text.startswith(header):
                        summary = text[len(header):]
                        break
            record_stream_diagnostic(
                "CONTEXT_COMPACTED",
                f"adda's {policy} context policy fired: "
                f"{report.dropped} message(s) dropped, "
                f"{report.truncated} truncated "
                f"({report.before} -> {report.after} est. tokens).",
                compaction_data={
                    "policy": policy, "window": window,
                    "window_source": source, **report.as_dict()},
            )
            if debug_enabled():
                append_transcript({
                    "type": "ContextCompaction",
                    "text": f"policy={policy} window={window} source={source}",
                    "trim": report.as_dict(),
                    "policy": policy,
                    "summary": summary,
                })
        return {"llm_input_messages": kept}

    return _hook

_context_policy() -> str staticmethod

"compact" or "trim". An unknown value RAISES.

Silently falling back to a default would make a typo'd arm run as the baseline and report as a null result, which is the one failure an ablation cannot afford.

Source code in src/adda/_src/backends/openai_compatible.py

@staticmethod
def _context_policy() -> str:
    """``"compact"`` or ``"trim"``. An unknown value RAISES.

    Silently falling back to a default would make a typo'd arm run as the
    baseline and report as a null result, which is the one failure an
    ablation cannot afford.
    """
    from ..runtime import settings
    value = settings.get_str("context_policy", "compact").strip().lower()
    if value not in ("compact", "trim"):
        raise ValueError(
            f"context_policy must be 'compact' or 'trim', got {value!r}")
    return value

invoke(messages: list[dict], *, idle_timeout: float | None = None, retry_max: int | None = None, on_session_start: Any = None, on_session_end: Any = None, background_watch: Any = None) -> str

Run one full agent turn; return final assistant text.

Acquires _lock to serialize concurrent callers of this adapter object (each delegation has its own copy(), so they do not contend). Transient API/network failures are retried with exponential backoff (see retry_on_transient).

idle_timeout / retry_max give short advisory side-calls a tight budget. idle_timeout is accepted for signature parity with the Claude backend (HTTP requests carry their own socket timeout, so it is not separately applied here); retry_max caps retries for this call.

on_session_start: see ClaudeAdapter.invoke's docstring -- called the instant _lock is actually acquired, so a caller queued behind another same-role delegation learns when its wait is really over. Best-effort.

on_session_end: see ClaudeAdapter.invoke's docstring -- called as on_session_end(session_id, usage) under the lock, also when the turn raised (_capture_usage runs on the failure path too). This backend has no resumable sessions, so the id is always None.

Source code in src/adda/_src/backends/openai_compatible.py

def invoke(
    self, messages: list[dict], *,
    idle_timeout: float | None = None, retry_max: int | None = None,
    on_session_start: Any = None,
    on_session_end: Any = None,
    background_watch: Any = None,
) -> str:
    """Run one full agent turn; return final assistant text.

    Acquires _lock to serialize concurrent callers of this adapter object
    (each delegation has its own copy(), so they do not contend). Transient API/network
    failures are retried with exponential backoff (see retry_on_transient).

    ``idle_timeout`` / ``retry_max`` give short advisory side-calls a tight
    budget. ``idle_timeout`` is accepted for signature parity with the
    Claude backend (HTTP requests carry their own socket timeout, so it is
    not separately applied here); ``retry_max`` caps retries for this call.

    ``on_session_start``: see ClaudeAdapter.invoke's docstring -- called
    the instant ``_lock`` is actually acquired, so a caller queued
    behind another same-role delegation learns when its wait is really
    over. Best-effort.

    ``on_session_end``: see ClaudeAdapter.invoke's docstring -- called
    as ``on_session_end(session_id, usage)`` under the lock, also when
    the turn raised (``_capture_usage`` runs on the failure path too).
    This backend has no resumable sessions, so the id is always None.
    """
    from .base import retry_on_transient
    with self._lock:
        if background_watch is not None:
            background_watch.start()
        if on_session_start is not None:
            try:
                on_session_start()
            except Exception:  # noqa: BLE001
                pass
        # Cleared so a turn that raises before _capture_usage() runs
        # cannot hand its caller the PREVIOUS call's values.
        self.last_session_id = None
        self.last_usage = {}
        try:
            return retry_on_transient(
                lambda: self._invoke_once(messages), max_attempts=retry_max,
                on_retry=getattr(self, "on_retry", None))
        finally:
            if background_watch is not None:
                background_watch.end(min_depth=1)
            if on_session_end is not None:
                try:
                    on_session_end(
                        self.last_session_id, dict(self.last_usage or {}))
                except Exception:  # noqa: BLE001
                    pass

_invoke_once(messages: list[dict]) -> str

Core invoke logic — build agent if needed, run, return text.

Source code in src/adda/_src/backends/openai_compatible.py

def _invoke_once(self, messages: list[dict]) -> str:
    """Core invoke logic — build agent if needed, run, return text."""
    # One invoke == one delegation's worker run; reset the per-delegation
    # nudge cap. The cached agent's tool closures read this live.
    self._oracle_nudge.reset()
    from .base import get_delegation_id, get_run_config_path
    _rc = get_run_config_path()
    self._oracle_nudge.run_config_path = _rc
    self._notice_ctx = (Path(_rc).parent if _rc else None,
                        get_delegation_id())
    if self._agent is None:
        self._agent = self._build_agent()

    lc_msgs = _to_lc_messages(messages)
    # A request with no user turn is malformed, and some providers answer
    # it with an opaque 500 ("no user query found in messages") that names
    # nothing and looks intermittent. It is reachable here: BOTH
    # converters between graph state and this call — nodes.parsing's
    # _to_adapter_messages and _to_lc_messages above — keep only
    # Human/AI messages and silently discard every other role, so a
    # history carrying only system/tool messages filters to an empty
    # list. thread_id is fresh per invoke, so nothing server-side
    # backfills the missing turn either.
    #
    # This check tests for a NON-EMPTY user turn, not merely for a
    # HumanMessage. Testing the type alone is what let the reported 500
    # through: HumanMessage(content="") is a HumanMessage, so the guard
    # passed it, and the provider then rejected the request for having no
    # user query. An empty turn and a missing one are the same thing to
    # the server, so they are the same thing here.
    #
    # Fail here instead, naming what arrived and what survived. This is
    # an instrument as much as a guard: an attributable error is how the
    # question of which shape a real run reaches gets answered rather
    # than argued.
    from langchain_core.messages import HumanMessage as _HumanMessage
    if not any(isinstance(m, _HumanMessage)
               and str(m.content).strip() for m in lc_msgs):
        roles = [str(m.get("role", "user")) for m in messages]
        raise UserlessPayloadError(
            f"refusing to send a request with no user turn: "
            f"{len(messages)} message(s) in with role(s) {roles!r}, "
            f"{len(lc_msgs)} survived conversion. Two things are dropped "
            "by _to_adapter_messages and _to_lc_messages: roles other "
            "than user/human/ai/assistant, and turns whose content is "
            "empty after flattening. Empty content most often means a "
            "message carried only blocks with no readable payload."
        )
    cfg = {"configurable": {"thread_id": str(uuid.uuid4())}}

    # DEBUG: capture reasoning + tool-calls (parity with Claude).
    #
    # Streamed, and recorded AS IT GOES, for two reasons that only show
    # up on a real run. A batch write after invoke() returns means a
    # delegation in flight has no transcript at all — a literature
    # review that runs for half an hour is completely unobservable while
    # it is the thing you most want to watch. And if the call raises
    # (a provider error mid-turn), a write placed after it never
    # happens, so the delegation that failed leaves no trace of what it
    # did before dying — exactly the case where the transcript is worth
    # most. Observed on run 20260906T122744: D002 failed inside
    # langchain_openai and D003 ran 25+ minutes, and neither had a
    # single line on disk.
    from .base import append_transcript, debug_enabled
    _debug = debug_enabled()
    result = None
    seen = 0

    def _flush(state) -> None:
        """Append whatever messages are new since the last flush."""
        nonlocal seen
        msgs = (state or {}).get("messages") or []
        for _m in msgs[seen:]:
            append_transcript({
                "type": _m.__class__.__name__,
                "text": str(getattr(_m, "content", "")),
                "tools": getattr(_m, "tool_calls", None) or [],
            })
        seen = max(seen, len(msgs))

    # stream_mode="values" yields the whole state after each step, so the
    # last one seen is the final state invoke() would have returned — the
    # compiled graph's invoke() is itself that loop. Streaming is used on
    # BOTH paths (not just under --debug) so that a turn which dies
    # mid-loop still leaves its partial state behind: invoke() raises with
    # nothing in hand, and the tokens the server already generated for that
    # turn would be unrecoverable.
    try:
        raise_if_stopped("model call")
        for state in self._agent.stream(
            {"messages": lc_msgs}, config=cfg, stream_mode="values",
        ):
            result = state
            if _debug:
                _flush(state)
            raise_if_stopped("model call")
    except Exception:
        # A failed turn keeps everything captured up to the failure.
        if _debug and result is not None:
            _flush(result)
        self._capture_usage(lc_msgs, result)
        raise
    if result is None:
        result = {"messages": []}
    all_msgs = result["messages"]
    last = all_msgs[-1]

    self._capture_usage(lc_msgs, result)

    return str(last.content)

_capture_usage(lc_msgs: list, result: dict | None) -> None

Set last_usage from every model call this turn produced.

Called on the success path AND from the failure path, because a turn that raises has still spent whatever the server generated before it died — dropping it undercounts by the whole turn, which on a run whose delegations fail is most of the run.

result["messages"] is the WHOLE graph state, i.e. the input messages we passed in (lc_msgs) followed by everything the agent loop generated this call. The model is invoked once per tool-calling round trip, so a strategizer turn that loops several times before its final reply produces several AI messages, each carrying its own usage_metadata — reading only the last one silently dropped every intermediate call's tokens.

Summing must stop at the input boundary: thread_id is fresh per invoke, so the graph never carries earlier turns' messages into this call, and lc_msgs is exactly the prefix the graph started from — the reducer only ever appends. Slicing at len(lc_msgs) therefore counts exactly the messages this invocation produced, never the history that seeded it; summing over the whole state instead would double-count that history on every call and inflate every total.

Only AI messages carry usage_metadata (tool/human messages don't), so anything without it contributes zero rather than raising.

Source code in src/adda/_src/backends/openai_compatible.py

def _capture_usage(self, lc_msgs: list, result: dict | None) -> None:
    """Set ``last_usage`` from every model call this turn produced.

    Called on the success path AND from the failure path, because a turn
    that raises has still spent whatever the server generated before it
    died — dropping it undercounts by the whole turn, which on a run whose
    delegations fail is most of the run.

    ``result["messages"]`` is the WHOLE graph state, i.e. the input
    messages we passed in (``lc_msgs``) followed by everything the agent
    loop generated this call. The model is invoked once per tool-calling
    round trip, so a strategizer turn that loops several times before its
    final reply produces several AI messages, each carrying its own
    ``usage_metadata`` — reading only the last one silently dropped every
    intermediate call's tokens.

    Summing must stop at the input boundary: ``thread_id`` is fresh per
    invoke, so the graph never carries earlier turns' messages into this
    call, and ``lc_msgs`` is exactly the prefix the graph started from —
    the reducer only ever appends. Slicing at ``len(lc_msgs)`` therefore
    counts exactly the messages this invocation produced, never the
    history that seeded it; summing over the whole state instead would
    double-count that history on every call and inflate every total.

    Only AI messages carry usage_metadata (tool/human messages don't), so
    anything without it contributes zero rather than raising.
    """
    new_msgs = ((result or {}).get("messages") or [])[len(lc_msgs):]
    total_in = total_out = total_cache_read = total_cache_creation = 0
    total_fresh = 0
    call_inputs: list[int] = []
    for _m in new_msgs:
        meta = getattr(_m, "usage_metadata", None)
        if not meta:
            continue
        call_inputs.append(meta.get("input_tokens", 0) or 0)
        total_in += meta.get("input_tokens", 0) or 0
        total_out += meta.get("output_tokens", 0) or 0
        details = meta.get("input_token_details") or {}
        total_cache_read += details.get("cache_read", 0) or 0
        total_cache_creation += details.get("cache_creation", 0) or 0
        # prompt_tokens is the WHOLE prompt, cached part included; the
        # schema's fresh_input is the rest. Per call, so one call's cache
        # figure can never offset another's.
        total_fresh += max(
            (meta.get("input_tokens", 0) or 0)
            - (details.get("cache_read", 0) or 0)
            - (details.get("cache_creation", 0) or 0), 0)
    self.last_usage = {
        **normalized_usage(
            fresh_input=total_fresh, cache_read=total_cache_read,
            cache_write=total_cache_creation, output=total_out),
        "input_tokens": total_in,
        "output_tokens": total_out,
        "cache_read_input_tokens": total_cache_read,
        "cache_creation_input_tokens": total_cache_creation,
        "total_cost_usd": None,  # not available from open-weight/self-hosted
        # input_tokens above is a sum over the turn's model calls; these
        # show what one call read: how many calls, the first, the largest.
        **call_shape(call_inputs),
    }

_probe_context_window() -> int | None

Ollama's SERVED window, which is not the model's trained length.

/v1/models reports neither, so this uses Ollama's native /api/show. The distinction matters more here than anywhere: a Modelfile's num_ctx is what the server will actually accept, and Ollama's default is far below what any of these models were trained for. Taking context_length from model_info would report the trained window, sail past the real limit, and produce exactly the server-side truncation this trimming exists to prevent — so the parameter wins and the trained length is only a fallback.

Source code in src/adda/_src/backends/ollama.py

def _probe_context_window(self) -> int | None:
    """Ollama's SERVED window, which is not the model's trained length.

    ``/v1/models`` reports neither, so this uses Ollama's native
    ``/api/show``. The distinction matters more here than anywhere: a
    Modelfile\'s ``num_ctx`` is what the server will actually accept, and
    Ollama\'s default is far below what any of these models were trained
    for. Taking ``context_length`` from ``model_info`` would report the
    trained window, sail past the real limit, and produce exactly the
    server-side truncation this trimming exists to prevent — so the
    parameter wins and the trained length is only a fallback.
    """
    import re

    import requests
    root = self._base_url.rstrip("/")
    if root.endswith("/v1"):
        root = root[: -len("/v1")]
    try:
        r = requests.post(f"{root}/api/show",
                          json={"model": self.model}, timeout=5)
        r.raise_for_status()
        body = r.json()
    except Exception:
        return None
    # `parameters` is the Modelfile block, e.g. "num_ctx   262144"
    params = body.get("parameters")
    if isinstance(params, str):
        m = re.search(r"^\s*num_ctx\s+(\d+)", params, re.M)
        if m:
            return int(m.group(1))
    info = body.get("model_info")
    if isinstance(info, dict):
        for key, val in info.items():
            if key.endswith(".context_length") and isinstance(val, int):
                return val
    return None