agent.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/agent.py
 
"""
``Agent`` — one execution unit owning its dependencies, reading its session through ``ctx``.
 
Purpose
-------
Promote what was a god-tuple of arguments (``session``, ``client``,
``presenter``, ``model_manager``, ``tool_registry``,
``tool_schemas``, ``ctx``, ``turn_stats``) into a class. The REPL,
tests, and any future sub-agent worker construct an :class:`Agent`
once and call :meth:`run_turn` per user input. The same shape works
for an interactive shell, a :class:`LoggingPresenter`-driven test
harness, and (when Phase O wants it) a :class:`multiprocessing.Process`
worker reading :class:`AgentConfig` off a pipe.
 
Today's job is *not* to ship subprocess plumbing — only to make sure
the engine objects are constructible from declarative config so that
adding a serializer later is a small follow-up.
 
Responsibilities
----------------
- :class:`AgentConfig` — the declarative knobs an Agent needs to
  build its dependencies: which provider, which model, what
  reasoning effort, optional tool whitelist, optional system prompt
  override, token profile, ReAct cap.
- :class:`Agent` — combiner that either accepts pre-built engine
  objects (``client``, ``model_manager``, ``tool_registry``, ...) or
  constructs them from :class:`AgentConfig`.
- :meth:`Agent.run_turn` — the one-call-per-user-input entry point
  the REPL replaces ``run_turn(...)`` with.
- :meth:`Agent.compress_if_needed` — wraps
  :func:`oc_chat.core.context.maybe_compress` so callers don't need
  to thread the client through.
- :meth:`Agent.save` — wraps
  :func:`oc_chat.core.persistence.save_all` for symmetric
  ergonomics; also closes the presenter if it has a ``close()``
  method (LoggingPresenter file-descriptor lifecycle).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (Agent built, turn dispatched, saved).
    L3 — semantic details (tool filter applied, default model used).
    L4 — deep tracing.
 
Contracts
---------
- The :class:`AgentConfig` is a frozen-ish dataclass — mutating it
  after :class:`Agent` construction is not supported (the Agent has
  already cached its derived state). Use a new ``Agent`` to switch
  providers.
- :meth:`run_turn` accepts ``user_input: str | None``. When ``str``,
  the agent appends it as a user message before driving the loop.
  When ``None``, the loop runs against the session as-is — the path
  used after a ``/goal`` slash command that already added the user
  message itself.
- The agent does *not* own the slash-command :class:`Context`; the
  REPL sets ``agent.ctx`` and ``agent.turn_stats`` after construction
  when those features are in play. Headless callers leave them as
  ``None``.
- ``tool_filter`` (when set) restricts both ``tool_registry`` and
  ``tool_schemas`` so the model only sees the whitelisted tools.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.client_interface (provider-agnostic chat-completions facade)
- oc_chat.core.context (maybe_compress — context-window compression)
- oc_chat.core.models (ModelManager and capability lookups)
- oc_chat.core.persistence (save_all — REPL-exit save wrapper)
- oc_chat.core.provider_plugin (registry, ProviderPlugin)
- oc_chat.core.react_loop (run_turn — the per-turn ReAct loop)
- oc_chat.core.session (ChatSession, CANONICAL_POLICY, helpers)
- oc_chat.core.tool_registry (build_tool_registry — tool auto-discovery)
- oc_chat.core.cancellation (lazy; CancellationToken - the per-turn token the SIGINT handler flips)
- oc_chat.core.commands (lazy; Context - type hint for the attached slash-command context)
- oc_chat.core.session_transition (lazy; replace_session - the transition seam for a post-attach session assignment)
- oc_ui.presenter_base (lazy; Presenter - type hints for the presenter)
- oc_ui.terminal.input (lazy; TurnStats - type hints for the REPL turn statistics)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
from oc_chat.core.client_interface import ClientInterface
from oc_chat.core.context import maybe_compress
from oc_chat.core.models import ModelManager
from oc_chat.core.persistence import save_all
from oc_chat.core import provider_plugin
from oc_chat.core.react_loop import run_turn as _run_turn
from oc_chat.core.session import (
    ChatSession,
    CANONICAL_POLICY,
    build_system_prompt,
)
from oc_chat.core.tool_registry import build_tool_registry
 
if TYPE_CHECKING:
    from oc_chat.core.cancellation import CancellationToken
    from oc_chat.core.commands import Context
    from oc_ui.presenter_base import Presenter
    from oc_ui.terminal.input import TurnStats
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
@dataclass
class AgentConfig:
    """Declarative knobs an :class:`Agent` needs to build its dependencies.
 
    Carries the minimum information to (re)construct an Agent without
    handing it pre-built engine objects. The REPL and tests typically
    pass pre-built objects; future subprocess workers will reconstruct
    from this config alone.
    """
 
    provider_name: str = "deepseek"
    model: str | None = None  # None → use provider.default_model at construction
    reasoning_effort: str = "high"
    tool_filter: list[str] | None = None
    system_prompt: str | None = None
    # OCT-LINT: disable=no_hardcoded_secrets reason="AgentConfig defaults trip the name heuristic: token_profile ('balanced') is a token-BUDGET profile name, not a credential"
    token_profile: str = "balanced"
    max_react_iterations: int | None = None
 
 
class _FactoryDefault:
    """Sentinel meaning "build a fresh container" for the manual initialiser.
 
    Renders as ``<factory>`` so ``inspect.signature(Agent)`` is identical to the
    dataclass-generated one it replaces — the public ABI is preserved, not just
    approximated.
    """
 
    __slots__ = ()
 
    def __repr__(self) -> str:
        return "<factory>"
 
 
_FACTORY = _FactoryDefault()
 
 
@dataclass(init=False)
class Agent:
    """One self-contained execution unit.
 
    ``session`` is NOT owned here. Once :attr:`ctx` is attached, :attr:`session`
    reads THROUGH to ``ctx.session``, which is the single authority (plan 152
    D1). Before attachment — batch and benchmark callers that never build a
    ``Context`` — it reads a private backing slot.
 
    Construct with an :class:`AgentConfig` plus a *presenter*; pass pre-built
    ``client``/``model_manager``/``tool_registry``/``tool_schemas``/``session``
    to skip the construction path (the REPL does this; tests usually let the
    agent build everything).
 
    The initialiser is written by hand (``@dataclass(init=False)``) so
    ``session`` can be a property while the public ABI stays exactly what it was:
    ten positional-or-keyword parameters, two fresh-per-instance containers, and
    unchanged repr/eq. Narrowing to keyword-only would have been a breaking
    change to a documented contract with live call sites.
    """
 
    config: AgentConfig
    presenter: "Presenter"
    session: ChatSession = field(default=None)  # type: ignore[assignment]
    client: ClientInterface = field(default=None)  # type: ignore[assignment]
    model_manager: ModelManager = field(default=None)  # type: ignore[assignment]
    tool_registry: dict = field(default_factory=dict)
    tool_schemas: list = field(default_factory=list)
 
    # Slash-command and prompt integration — set by the REPL after
    # construction; ``None`` for headless callers.
    ctx: "Context | None" = None
    turn_stats: "TurnStats | None" = None
 
    # Phase O3d — the currently-active CancellationToken for the in-
    # flight turn. ``None`` between turns. The terminal signal handler
    # in main.py reads this attribute to flip the flag when SIGINT is
    # received; any presenter implementing a stop button does the same.
    cancellation: "CancellationToken | None" = None
 
    def __init__(
        self,
        config: AgentConfig,
        presenter: "Presenter",
        session: ChatSession = None,  # type: ignore[assignment]
        client: ClientInterface = None,  # type: ignore[assignment]
        model_manager: ModelManager = None,  # type: ignore[assignment]
        tool_registry: dict = _FACTORY,  # type: ignore[assignment]
        tool_schemas: list = _FACTORY,  # type: ignore[assignment]
        ctx: "Context | None" = None,
        turn_stats: "TurnStats | None" = None,
        cancellation: "CancellationToken | None" = None,
    ) -> None:
        func = "__init__"
        self.config = config
        self.presenter = presenter
        self.client = client
        self.model_manager = model_manager
        # Fresh containers per instance — two Agents must never share these.
        self.tool_registry = {} if tool_registry is _FACTORY else tool_registry
        self.tool_schemas = [] if tool_schemas is _FACTORY else tool_schemas
        self.turn_stats = turn_stats
        self.cancellation = cancellation
        # The backing slot is written DIRECTLY here, never through the property:
        # construction is not a session transition, and routing it through the
        # seam would take the entry lease (or be rejected) mid-``__init__``.
        self._session_slot: ChatSession = session
        self.ctx = ctx
 
        cfg = self.config
 
        # Engine objects: build from config when not supplied.
        if self.client is None or self.model_manager is None:
            provider = provider_plugin.registry.get(cfg.provider_name)
            if self.client is None:
                self.client = ClientInterface(provider)
            if self.model_manager is None:
                self.model_manager = ModelManager(provider)
 
        # Tool registry: auto-discover when not supplied.
        if not self.tool_registry and not self.tool_schemas:
            self.tool_registry, self.tool_schemas = build_tool_registry()
 
        # Tool filter: restrict both registry and schemas to the whitelist.
        if cfg.tool_filter is not None:
            allow = set(cfg.tool_filter)
            self.tool_registry = {
                k: v for k, v in self.tool_registry.items() if k in allow
            }
            self.tool_schemas = [
                s for s in self.tool_schemas
                if s.get("function", {}).get("name") in allow
            ]
            _dbg("OC_CHAT.MAIN", func,
                 f"tool_filter applied: {len(self.tool_registry)} tool(s) remaining", 3)
 
        # Session: build from config when not supplied. Apply the system
        # prompt override (or default + embedded tool schemas), and the
        # session-level knobs from config. Writes go to the backing SLOT, not
        # through the property — see the note above ``self._session_slot``.
        if self._session_slot is None:
            base_system = cfg.system_prompt or CANONICAL_POLICY
            system_text = build_system_prompt(base_system, self.tool_schemas)
            model = cfg.model or self.model_manager.provider.default_model
            self._session_slot = ChatSession(
                model=model,
                reasoning_effort=cfg.reasoning_effort,
                token_profile=cfg.token_profile,
                messages=[{"role": "system", "content": system_text}],
            )
            self._session_slot.max_react_iterations = cfg.max_react_iterations
            _dbg("OC_CHAT.MAIN", func,
                 f"new session={self._session_slot.name} model={model}", 2)
        else:
            _dbg("OC_CHAT.MAIN", func,
                 f"using supplied session={self._session_slot.name}", 2)
 
    # ── Session identity: read-through to the single authority (D1) ────────
 
    @property
    def session(self) -> ChatSession:
        """The active session — ``ctx.session`` once a context is attached.
 
        Reading through is what makes ``agent.session is ctx.session`` true for
        EVERY path, so a turn can no longer run against a session the UI has
        already replaced. Before attachment (batch / benchmark callers that
        never build a ``Context``) this reads the private backing slot.
        """
        ctx = getattr(self, "ctx", None)
        if ctx is not None:
            return ctx.session  # type: ignore[return-value]
        return self._session_slot
 
    @session.setter
    def session(self, value: ChatSession) -> None:
        """Set the active session.
 
        PRE-attach this writes the backing slot — the batch and benchmark
        entrypoints construct an Agent, assign a session, and only later (or
        never) attach a context, and those writes are slot writes by design.
 
        POST-attach it routes through the transition seam, because a bare
        ``ctx.session =`` re-derives none of the dependent state. The seam takes
        the entry lease, so an off-main-thread, mid-turn or reentrant assignment
        is REJECTED with ``TransitionRejectedError`` and has zero effects —
        enforced here rather than trusted to the caller, since this setter is
        reachable from arbitrary code.
        """
        ctx = getattr(self, "ctx", None)
        if ctx is None:
            self._session_slot = value
            return
        from oc_chat.core.session_transition import replace_session
        replace_session(ctx, value)
 
    # ── Per-turn driver ────────────────────────────────────────────────────
 
    def run_turn(self, user_input: str | None = None) -> None:
        """Drive one ReAct turn against the session.
 
        When *user_input* is a non-empty string, append it as a user
        message before running the loop. When ``None``, run the loop
        against the session as-is — the path used after a ``/goal``
        slash command that already added the user message itself.
 
        Owns a fresh :class:`CancellationToken` for the duration of the
        turn (Phase O3d). Stored on ``self.cancellation`` so the SIGINT
        handler in main.py can flip it. Cleared in the ``finally``
        block so subsequent turns start with a clean slate.
        """
        from oc_chat.core.cancellation import CancellationToken
        self.cancellation = CancellationToken()
        try:
            if user_input is not None:
                self.session.add_user(user_input)
            _run_turn(
                self.session,
                self.client,
                self.presenter,
                self.model_manager,
                self.tool_registry,
                self.tool_schemas,
                ctx=self.ctx,
                turn_stats=self.turn_stats,
                cancellation=self.cancellation,
            )
        finally:
            self.cancellation = None
 
    # ── Lifecycle helpers ──────────────────────────────────────────────────
 
    def compress_if_needed(self) -> bool:
        """Trigger :func:`maybe_compress` against the agent's session.
 
        Returns ``True`` when compression actually ran. The REPL calls
        this between user input and :meth:`run_turn` so the prompt
        stays under the model's context window.
        """
        return maybe_compress(self.session, self.client, self.presenter)
 
    def save(self) -> None:
        """Persist session metadata and chat log; close the presenter.
 
        Wraps :func:`save_all` with the presenter-close step so the
        REPL's ``finally`` block becomes a single call. Headless
        callers can skip ``save`` entirely by calling
        :func:`save_all` directly when they want only persistence.
        """
        if hasattr(self.presenter, "close"):
            self.presenter.close()
        save_all(self.session)