orchestrata_ultimate.agents.volks.base

Reference › orchestrata_ultimate › agents › volks › base

base defines BaseAgent, the Pydantic model that every volk in the delivery team inherits from. Each concrete volk (analyst, architect, codedirector and the others) imports it. It uses tool_registry for tool schemas, persona for the shared preamble and memory_context for recall. The curated narrative is volks.

Identity and state are fields: generated id, timestamps, alias, role, a WorkStatus status, a one-line summary, the current_thread and a list of output_artifacts. Collaborators are private attributes set after construction: set_postman, set_kg, set_volk_registry, set_runtime_writer and set_context_injector. Messaging goes through send (a CXP packet through Postman) and emit_event, which first notifies local observers on the volk's events node and then sends a packet addressed to SYSTEM; if Postman is unset the send raises RuntimeError. Subclasses override process, which takes an incoming CXPPacket and returns an optional response, and get_output_display.

The LLM paths are _call_llm, _call_reasoner and _call_llm_with_tools. _call_llm accepts an injected override callable for tests; otherwise it stamps the volk alias for per-volk provider routing, folds recalled memory into the graph context (_recall_block, empty when no memory engine is registered for the thread), wraps the system prompt with compose_persona and calls call_llm from the llm client. With a payload and a context injector it runs the validator feedback loop, retrying when citations are missing, and it records the call with the run tracer. _call_llm_with_tools takes the tool list from get_agent_tools for the alias, falls back to _call_llm when there are none, and loops for up to max_turns tool-executing turns (default 10) before forcing a final call without tools. _strip_dsml_tool_markup removes DeepSeek tool-call markup from that final response.

At a glance

Class diagram

Dependency diagram

Truncated at 25 nodes: showing 12 of 15 imports and 12 of 13 importers.

API

Classes

BaseAgent (pydantic)

Bases: BaseModel · Line 60

Base class for senior agents.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/base.py

# mypy: ignore-errors
"""Base agent class.

Inherits from Pydantic BaseModel for validation, serialization, and identity.
"""

from __future__ import annotations

import logging
import re
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any
from uuid import UUID, uuid4

from pydantic import BaseModel, ConfigDict, Field, PrivateAttr

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
from orchestrata_ultimate.core.patterns.observer import AsyncObservable

if TYPE_CHECKING:
    from orchestrata_ultimate.core.protocols import VolkHealthProbe
    from orchestrata_ultimate.knowledge_graph.kg_context import KGCodeContext
    from orchestrata_ultimate.knowledge_graph.runtime_writer import RuntimeSignalWriter
    from orchestrata_ultimate.postman.postman import Postman

logger = logging.getLogger(__name__)

# Dogfood run-3e: DeepSeek v4 ignores tool_choice="none" while the `tools`
# schema is still present in the request and instead emits its own internal
# tool-call markup literally as response text -- runs delimited by U+FF5C
# (FULLWIDTH VERTICAL BAR), e.g. `<｜｜DSML｜｜tool_calls>\n<｜｜DSML｜｜invoke
# name="read_file">...`, which then passes the volk-level "non-empty
# response" guards and ships garbage downstream. Fix 1 (the OpenAI-compatible
# provider adapters now omitting `tools`/`tool_choice` from the request when
# `tool_choice=="none"`, mirroring `anthropic_adapter.py`) prevents this at
# the source; ``_strip_dsml_tool_markup`` below is a backstop for whatever
# slips through anyway. It strips any span starting with the U+FF5C marker
# "<｜" through its closing "｜>", or to end-of-string if the markup is never
# closed (the common case -- these blobs trail off mid-invocation instead of
# completing).
_DSML_MARKUP_RE = re.compile(r"<｜.*?(?:｜>|$)", re.DOTALL)


def _strip_dsml_tool_markup(content: str) -> str:
    """Strip DeepSeek DSML tool-call markup from a finalization response.

    Returns the surrounding prose with any DSML markup spans removed and
    leading/trailing whitespace stripped. Content with no markup passes
    through with only whitespace-stripping applied (module-level pure
    function -- no side effects, no dependency on ``self`` -- so it is
    directly unit-testable in isolation of the tool loop).
    """
    if not content:
        return content
    return _DSML_MARKUP_RE.sub("", content).strip()


class BaseAgent(BaseModel):
    """Base class for senior agents.

    Inherits from BaseModel to gain:
    - Validation and serialization
    - Custom fields with defaults
    """

    model_config = ConfigDict(extra="allow")

    id: UUID = Field(default_factory=uuid4)
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    alias: str = ""
    role: str = ""
    status: WorkStatus = WorkStatus.PENDING
    summary: str = "Idle"
    current_thread: str | None = None
    output_artifacts: list[str] = Field(default_factory=list)

    _postman: Any | None = PrivateAttr(default=None)
    _kg: Any | None = PrivateAttr(default=None)
    _volk_registry: Any | None = PrivateAttr(default=None)
    _runtime_writer: RuntimeSignalWriter | None = PrivateAttr(default=None)
    _context_injector: Any | None = PrivateAttr(default=None)
    _events: AsyncObservable[dict[str, Any]] = PrivateAttr(
        default_factory=AsyncObservable
    )

    @property
    def events(self) -> AsyncObservable[dict[str, Any]]:
        """Read-only handle to this volk's event mesh node.

        Other volks (or any subscriber) attach via
        ``volk.events.subscribe(observer)`` to receive every observation
        emitted through ``_emit_observation``/``emit_event``.
        """
        return self._events

    @traces(
        spec="002",
        story="FR-011",
        reqs=["REQ-002-FR11-021"],  # Agent registration with Postman
    )
    def set_postman(self, postman: Postman) -> None:
        """Set postman reference for agents that need direct audit logging.

        Args:
            postman: Postman instance.
        """
        self._postman = postman

    def set_kg(self, kg_context: KGCodeContext) -> None:
        """Set knowledge graph context for code-aware agents.

        Args:
            kg_context: KGCodeContext instance for Neo4j queries.
        """
        self._kg = kg_context

    def set_volk_registry(self, registry: VolkHealthProbe) -> None:
        """Set the volk registry so tool calls can query peer health/status.

        Args:
            registry: VolkRegistry-compatible instance (structurally
                satisfies ``VolkHealthProbe``).
        """
        self._volk_registry = registry

    def set_runtime_writer(self, writer: RuntimeSignalWriter | None) -> None:
        """Set the runtime signal writer so tool calls are recorded in the KG.

        Args:
            writer: RuntimeSignalWriter instance, or None to disable.
        """
        self._runtime_writer = writer

    def set_context_injector(self, injector: Any | None) -> None:
        """Set the context injector for the validator feedback loop.

        Args:
            injector: ContextInjector instance, or None to disable expansion.
        """
        self._context_injector = injector

    @traces(
        spec="005",
        story="US-008",
        reqs=[
            "REQ-005-FR10-001",  # Senior-to-junior delegation uses CXP
            "REQ-005-AC11-001",  # All agent communication uses CXP
        ],
    )
    async def send(self, packet: CXPPacket) -> CXPPacket | None:
        """Send a CXP packet through the postman.

        This is how agents communicate with other agents,
        emit events, and log audit entries.
        """
        if self._postman is None:
            raise RuntimeError(f"{self.alias}: postman not set — cannot send messages")
        return await self._postman.aroute(packet)

    async def _emit_observation(self, event_type: str, payload: dict[str, Any]) -> None:
        """Emit an observation to this volk's local event mesh node.

        Delivered to every subscriber of ``self.events`` (see
        ``AsyncObservable``), independent of whether a postman is set —
        unlike ``emit_event``'s CXP leg, this never raises: the observable's
        ``emit`` is already log-and-continue.
        """
        await self._events.emit(
            {
                "volk": self.alias,
                "event_type": event_type,
                "thread_id": self.current_thread,
                "timestamp": datetime.now(UTC).isoformat(),
                "payload": payload,
            }
        )

    async def emit_event(self, event_type: str, payload: dict) -> None:
        """Emit a system event: to local observers, then via CXP.

        Dual-emit: local subscribers (``self.events``) always receive the
        observation first, even if no postman is set for this volk; the CXP
        packet send follows unchanged, including its ``RuntimeError`` when
        postman is unset.
        """
        await self._emit_observation(event_type, payload)
        packet = CXPPacket(
            from_agent=self.alias,
            to_agent="SYSTEM",
            subject=event_type,
            payload=payload,
            work_status=WorkStatus.RESOLVED,
        )
        await self.send(packet)

    async def on_event(self, event: dict[str, Any]) -> None:
        """Default ``AsyncObserver`` implementation: log at DEBUG.

        Structurally satisfies the runtime-checkable
        ``core.patterns.observer.AsyncObserver`` protocol, so any
        ``BaseAgent`` can itself subscribe to another volk's ``events``
        (``a.events.subscribe(b)``). Subclasses may override to react to
        observed events.
        """
        logger.debug("%s observed event: %r", self.alias, event)

    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Process an incoming CXP packet.

        Async: every concrete volk overrides this with ``async def`` and the
        executor awaits it (``await volk.process(packet)``); the base contract
        must match so the override is type-compatible.

        Args:
            packet: Incoming CXP packet to process.

        Returns:
            Response packet, or None.

        Raises:
            NotImplementedError: Subclasses must implement.
        """
        raise NotImplementedError

    def get_output_display(self) -> str:
        """Return formatted output for human review."""
        if not self.output_artifacts:
            return "No artifacts yet."
        return "\n".join(self.output_artifacts)

    async def _recall_block(self, query: str) -> str:
        """Return prior-memory context for this run's thread (SPEC-010), or "".

        Best-effort: empty when no memory engine is registered.
        """
        from orchestrata_ultimate.agents.volks.memory_context import recall

        return await recall(getattr(self, "current_thread", None), query)

    async def _call_llm(
        self, system_prompt: str, user_message: str, **kwargs: Any
    ) -> str:
        """Call LLM with override support for testing.

        Subclasses and tests can inject a custom callable via
        ``_call_llm`` constructor kwarg (BaseAgent allows extra fields)
        or by setting ``agent._call_llm = fn`` after construction.

        When a ``payload`` kwarg is supplied and ``set_context_injector`` has
        been called, the validator feedback loop is used: the response is
        checked against injected graph context, and on missing citations the
        context is expanded and the call retries up to a bounded limit.
        """
        import time as _time

        from orchestrata_ultimate.core.tracing.context import get_tracer

        override = self.__dict__.get("_call_llm") or getattr(
            self, "__pydantic_extra__", {}
        ).get("_call_llm")
        tracer = get_tracer(getattr(self, "current_thread", None))
        start = _time.monotonic()

        async def _do_call(sys: str, user: str, **kw: Any) -> str:
            if override is not None:
                return await override(sys, user, **kw)
            from orchestrata_ultimate.agents.volks.persona import compose_persona
            from orchestrata_ultimate.llm.client import call_llm

            # Stamp the volk alias so per-volk provider/model routing applies and
            # the conversation capture attributes the exchange to this volk.
            kw.setdefault("volk", getattr(self, "alias", None))
            # Fold relevant prior memories into the injected context (best-effort;
            # empty when no memory engine is registered for this run's thread).
            memory_block = await self._recall_block(user)
            if memory_block:
                existing = kw.get("graph_context") or ""
                kw["graph_context"] = (
                    f"{memory_block}\n\n---\n\n{existing}" if existing else memory_block
                )
            return await call_llm(compose_persona(sys), user, **kw)

        payload = kwargs.pop("payload", None)
        injector = getattr(self, "_context_injector", None)
        if payload is not None and injector is not None:
            from orchestrata_ultimate.knowledge_graph.context_validator_loop import (
                ValidatorLoop,
            )

            loop = ValidatorLoop(injector)
            call_kwargs = {k: v for k, v in kwargs.items() if k != "graph_context"}
            loop_result = await loop.run(
                payload,
                _do_call,
                system_prompt=system_prompt,
                user_message=user_message,
                call_kwargs=call_kwargs,
            )
            result = loop_result.response
        else:
            result = await _do_call(system_prompt, user_message, **kwargs)
        if tracer is not None:
            tracer.trace_llm_call(
                volk=getattr(self, "alias", "") or "",
                prompt_length=len(system_prompt) + len(user_message),
                history_length=0,
                temperature=float(kwargs.get("temperature", 0.0) or 0.0),
                response_length=len(result or ""),
                duration_ms=round((_time.monotonic() - start) * 1000, 2),
            )
        return result

    async def _call_reasoner(self, prompt: str, **kwargs: Any) -> str:
        """Call reasoner with override support for testing."""
        import time as _time

        from orchestrata_ultimate.core.tracing.context import get_tracer

        override = self.__dict__.get("_call_reasoner") or getattr(
            self, "__pydantic_extra__", {}
        ).get("_call_reasoner")
        tracer = get_tracer(getattr(self, "current_thread", None))
        start = _time.monotonic()
        if override is not None:
            result = await override(prompt, **kwargs)
        else:
            from orchestrata_ultimate.agents.volks.persona import compose_persona
            from orchestrata_ultimate.llm.client import call_reasoner

            # Fold relevant prior memories into the injected context (best-effort;
            # empty when no memory engine is registered for this run's thread).
            memory_block = await self._recall_block(prompt)
            if memory_block:
                existing = kwargs.get("graph_context") or ""
                kwargs["graph_context"] = (
                    f"{memory_block}\n\n---\n\n{existing}" if existing else memory_block
                )
            result = await call_reasoner(compose_persona(prompt), **kwargs)
        if tracer is not None:
            tracer.trace_llm_call(
                volk=getattr(self, "alias", "") or "",
                prompt_length=len(prompt),
                history_length=0,
                temperature=float(kwargs.get("temperature", 0.0) or 0.0),
                response_length=len(result or ""),
                duration_ms=round((_time.monotonic() - start) * 1000, 2),
            )
        return result

    async def _call_llm_with_tools(
        self,
        system_prompt: str,
        user_message: str,
        tools: list[dict[str, Any]] | None = None,
        max_turns: int = 10,
        **kwargs: Any,
    ) -> str:
        """Call LLM with tool-calling support (multi-turn loop).

        The agent's available tools are determined by its alias via
        ``get_agent_tools()``. The LLM can call tools and receive results
        in a loop until it produces a final text response.

        Args:
            system_prompt: System prompt for the LLM.
            user_message: User message / task description.
            tools: Optional tool list override (defaults to agent's tools).
            max_turns: Number of tool-executing turns to allow before forcing
                a no-tools finalization call (dogfood run-4/F13: grounded
                design work against a real codebase can burn 6-9 turns just
                exploring, so callers doing that kind of work may need a
                larger budget than the default).
            **kwargs: Passed to call_with_tools (temperature, volk, etc.).

        Returns:
            Final text response after all tool calls complete.
        """

        import json

        from orchestrata_ultimate.agents.tool_registry import (
            ToolExecutor,
            get_agent_tools,
        )
        from orchestrata_ultimate.core.protocols import Message
        from orchestrata_ultimate.llm.client import call_with_tools

        def _tool_call_name_and_args(tc: dict[str, Any]) -> tuple[str, dict[str, Any]]:
            """Parse an OpenAI/DeepSeek-compatible tool_calls entry.

            ``response.tool_calls`` entries are raw provider dicts, e.g.
            ``{"id": ..., "function": {"name": ..., "arguments": "<json>"}}``
            (see ``providers/adapters/deepseek_adapter.py`` and
            ``refactor/agent.py``'s identical tool loop) — not objects with
            ``.name``/``.arguments`` attributes.
            """
            fn = tc.get("function", {})
            name = fn.get("name", "")
            try:
                arguments = json.loads(fn.get("arguments") or "{}")
            except json.JSONDecodeError:
                arguments = {}
            return name, arguments

        if tools is None:
            tools = get_agent_tools(getattr(self, "alias", None) or "")

        if not tools:
            return await self._call_llm(system_prompt, user_message, **kwargs)

        # Fold relevant prior memories into the user turn (best-effort; empty
        # when no memory engine is registered for this run's thread).
        # call_with_tools takes a message list (no graph_context kwarg), so the
        # fold happens here rather than via kwargs. Only in this tools branch —
        # the no-tools fallback above already recalls via ``_call_llm``, so this
        # avoids folding memory twice.
        memory_block = await self._recall_block(user_message)
        user_content = (
            f"{memory_block}\n\n---\n\n{user_message}" if memory_block else user_message
        )

        executor = ToolExecutor(
            self._get_workspace(),
            postman=self._postman,
            from_alias=getattr(self, "alias", None),
            thread_id=getattr(self, "current_thread", None),
            volk_registry=self._volk_registry,
            codebase_kb_environ=self._codebase_kb_environ(),
        )
        messages: list[Message] = [
            Message(role="system", content=system_prompt),
            Message(role="user", content=user_content),
        ]

        response = None
        for _ in range(max_turns):
            response = await call_with_tools(
                messages=messages,
                tools=tools,
                volk=getattr(self, "alias", None),
                **{k: v for k, v in kwargs.items() if k in ("temperature", "model")},
            )
            if not response.tool_calls:
                return response.content or ""
            for tc in response.tool_calls:
                name, arguments = _tool_call_name_and_args(tc)
                result = await executor.execute(name, arguments)
                content = result.output
                if not result.success and result.error:
                    content = f"Error: {result.error}\n\n{result.output}"
                writer = getattr(self, "_runtime_writer", None)
                if writer is not None:
                    try:
                        from orchestrata_ultimate.knowledge_graph.signal_extraction import (
                            extract_entity_ids,
                            extract_file_paths,
                        )

                        writer.write_tool_call(
                            tool_call_id=tc.get("id", ""),
                            thread_id=getattr(self, "current_thread", None) or "",
                            tool_name=name,
                            arguments=arguments,
                            result_summary=content[:1000],
                            outcome="success" if result.success else "error",
                            mentioned_files=extract_file_paths(arguments),
                            mentioned_entities=extract_entity_ids(arguments),
                        )
                    except Exception as exc:
                        logger.debug("Runtime tool-call signal write failed: %s", exc)
                messages.append(
                    Message(
                        role="assistant",
                        content=response.content or "",
                        tool_calls=[tc],
                        reasoning_content=response.reasoning_content,
                    )
                )
                messages.append(
                    Message(
                        role="tool",
                        content=content[:4000],
                        tool_call_id=tc.get("id", ""),
                    )
                )
        # Dogfood run-2 F3 defect 1: the loop above spent its entire
        # ``max_turns`` budget on tool calls, so ``response.content`` here is
        # whatever text accompanied the LAST tool_calls turn — a truncated
        # mid-thought fragment, not a finished answer. Make ONE additional
        # call with ``tool_choice="none"`` so the model cannot request yet
        # another tool and must produce a complete text answer from
        # everything gathered so far. ``tools`` is still passed (not emptied):
        # dogfood run-3e showed DeepSeek v4 ignoring ``tool_choice="none"``
        # while the real ``tools`` schema was still attached and instead
        # emitting its internal tool-call markup as text (see
        # ``_strip_dsml_tool_markup`` above). Fix 1 -- the OpenAI-compatible
        # provider adapters (openai/deepseek/openrouter) now mirror the
        # Anthropic adapter's behavior and omit both `tools` and
        # `tool_choice` from the outgoing request whenever `tool_choice`
        # resolves to "none" -- addresses this at the source, so passing the
        # real schema here no longer risks an empty ``tools: []`` array
        # reaching an API that rejects it when tool_choice isn't "none".
        alias = getattr(self, "alias", None) or "<unknown volk>"
        logger.warning(
            "%s: tool budget exhausted after %d turns without a final answer "
            "-- forcing a no-tools finalization call",
            alias,
            max_turns,
        )
        messages.append(
            Message(
                role="user",
                content=(
                    "Tool budget exhausted — produce your complete final "
                    "answer now from what you have gathered. Respond with "
                    "plain text only; do not attempt to call tools."
                ),
            )
        )
        response = await call_with_tools(
            messages=messages,
            tools=tools,
            tool_choice="none",
            volk=getattr(self, "alias", None),
            **{k: v for k, v in kwargs.items() if k in ("temperature", "model")},
        )
        # Backstop for whatever DSML markup slips through despite Fix 1 (see
        # ``_strip_dsml_tool_markup``'s module-level docstring above).
        raw_final = response.content or ""
        finalized = _strip_dsml_tool_markup(raw_final)
        if finalized != raw_final.strip():
            logger.warning(
                "%s: finalization response contained DSML tool-call markup "
                "-- stripped before returning",
                alias,
            )
        return finalized

    def _get_workspace(self) -> Path:
        from pathlib import Path

        from orchestrata_ultimate.core.infra_config import WORKSPACE_DIR

        return Path(WORKSPACE_DIR) if WORKSPACE_DIR else Path.cwd()

    def _codebase_kb_environ(self) -> dict[str, str] | None:
        """Test-injection seam for the codebase-intelligence query tools'
        production KB lifecycle (``codebase_query_tools._build_production_kb``).

        ``None`` (the default) means "use the real process environment and
        the real project root" -- the production behaviour. A test overrides
        this method (the same instance-attribute-swap convention as
        ``_get_workspace``) to point ``codebase_impact``/``codebase_module_deps``/
        etc. at an isolated fixture KB instead of the shared production one.
        """
        return None