orchestrata_ultimate.agents.volks.senior_dev.agent

Reference › orchestrata_ultimate › agents › volks › senior_dev › agent

SeniorDevAgent (alias U:SD, role Senior Developer) is the out-of-band troubleshooter, not a stage of the main line. It extends BaseAgent (base) and is registered by the kernel pipeline; it is activated when the goal runner sees oscillation, meaning fixes keep breaking each other and iterations do not converge. The curated role description is in volks.

What it receives: a packet whose payload holds iteration_history (conversation excerpts), original_intent and fail_summary (the last three iterations). What it produces: a prompt_refinement packet addressed to the Concierge (U:Cx) carrying root_cause, a list of constraints, a revised_addendum for the original intent and the full analysis text.

process resets its three result fields, builds a prompt from the payload and calls the LLM with the tool loop and SENIOR_DEV_PROMPT. _parse_response reads the labelled parts: ROOT_CAUSE (one line), CONSTRAINTS (one per line, stripped of dashes) and REVISED_INTENT_ADDENDUM (the remaining text). The agent is a single class; the package __init__ (init) re-exports it.

At a glance

Class diagram

Dependency diagram

API

Classes

SeniorDevAgent (pydantic)

Bases: BaseAgent · Line 49

Source

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

# mypy: ignore-errors
"""U:SD — Senior Dev agent for oscillation recovery.

Activates when the goal runner detects oscillation (fix A breaks B).
Reads the last N iteration conversations, identifies the root cause
pattern, and rewrites the Concierge prompt with additional constraints
to break the loop.

Uses Opus (reasoning tier) for deep analysis.
"""

from pydantic import Field

from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

SENIOR_DEV_PROMPT = """\
You are U:SD, the Senior Developer — a principal engineer who debugs \
systemic failures in the AI code generation pipeline.

You are called when the pipeline is OSCILLATING: fixing one issue \
introduces another, and multiple iterations fail to converge.

YOUR MISSION: Analyze the failure history, identify the root cause \
pattern, and produce CONCRETE constraints that prevent the pattern \
from recurring.

PROCESS:
1. Read the last 3 iteration conversations carefully.
2. Identify what was fixed and what broke each time.
3. Find the root cause (wrong architecture, missing dependency, \
   incorrect API assumption, circular import, etc.).
4. Write 3-5 CONSTRAINTS that must be added to the original intent \
   to prevent the oscillation.

OUTPUT FORMAT:
ROOT_CAUSE: <one sentence describing the systemic issue>
CONSTRAINTS:
- <constraint 1>
- <constraint 2>
- <constraint 3>
REVISED_INTENT_ADDENDUM: <paragraph to append to the original intent>

Do NOT rewrite the entire intent. Output ONLY the addendum that \
prevents the oscillation pattern.
"""


class SeniorDevAgent(BaseAgent):
    # W-tool-wave: U:SD needs its own alias (previously unset -- inherited
    # BaseAgent's default alias == "") so that BaseAgent._call_llm_with_tools
    # can resolve its granted tool roster via
    # tool_registry.get_agent_tools(self.alias) -> AGENT_TOOLS["U:SD"].
    # Without this, get_agent_tools("") returns [] and the tool loop silently
    # falls back to the no-tools _call_llm path -- matches the
    # ArchitectVolk/DocumentorVolk convention of declaring alias as a Field.
    alias: str = Field(default="U:SD")
    role: str = "Senior Developer"
    root_cause: str = ""
    constraints: list[str] = []
    revised_addendum: str = ""

    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)

    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        self.summary = "Analyzing oscillation pattern"

        # Reset parse state at the top of every process() call.
        # _parse_response only APPENDS to self.constraints and never clears
        # root_cause/constraints/revised_addendum, so re-invoking process()
        # on one long-lived SeniorDevAgent instance (the goal runner may call
        # it again on a subsequent oscillation) would otherwise leak the
        # prior run's constraints into the new response packet.
        self.root_cause = ""
        self.constraints = []
        self.revised_addendum = ""

        iteration_history = packet.payload.get("iteration_history", "")
        original_intent = packet.payload.get("original_intent", "")
        fail_summary = packet.payload.get("fail_summary", "")

        prompt = (
            f"Original intent:\n{original_intent}\n\n"
            f"Failure summary (last 3 iterations):\n{fail_summary}\n\n"
            f"Full conversation excerpts:\n{iteration_history}\n\n"
            "Analyze the oscillation and output ROOT_CAUSE, "
            "CONSTRAINTS, and REVISED_INTENT_ADDENDUM."
        )

        print("  [U:SD] Analyzing oscillation with Opus...", flush=True)

        # W-tool-wave: U:SD was granted tools in AGENT_TOOLS (tool_registry.py
        # -- repl/enter_worktree/exit_worktree/lsp/web_search/web_fetch), but
        # this call went through _call_reasoner, which has no tool support --
        # the granted tools were inert. Route through the tool-calling loop
        # instead, matching ArchitectVolk._handle_design / DocumentorVolk.
        # _handle_documentation: system_prompt/user_message passed explicitly
        # (SENIOR_DEV_PROMPT was previously never forwarded to the LLM at all
        # -- call_reasoner takes only a single user_message parameter, no
        # system prompt -- so SENIOR_DEV_PROMPT was dead code before this
        # change). No graph_context kwarg on this path (unlike call_reasoner/
        # call_llm) -- process() never extracts one from the packet, so there
        # is nothing to fold into the user turn here. Default max_turns=10:
        # unlike the Architect's grounded codebase design work (dogfood F13,
        # 6-9 exploration turns against read_file/search_code/list_dir), U:SD
        # carries no codebase-read tools at all -- its root-cause analysis
        # reasons over the iteration-history text already in the prompt, using
        # repl/lsp/web_search only for incidental verification -- so there is
        # no equivalent precedent here for a larger budget.
        response = await self._call_llm_with_tools(SENIOR_DEV_PROMPT, prompt)

        self._parse_response(response)

        print(
            f"  [U:SD] Root cause: {self.root_cause}",
            flush=True,
        )

        for c in self.constraints:
            print(f"  [U:SD] Constraint: {c}", flush=True)

        self.status = WorkStatus.RESOLVED
        self.summary = f"Root cause identified: {self.root_cause[:60]}"
        self.output_artifacts.append(response)

        return CXPPacket(
            from_agent="U:SD",
            to_agent="U:Cx",
            thread_id=packet.thread_id,
            subject="prompt_refinement",
            payload={
                "root_cause": self.root_cause,
                "constraints": self.constraints,
                "revised_addendum": self.revised_addendum,
                "full_analysis": response,
            },
            work_status=WorkStatus.PENDING,
        )

    def _parse_response(self, response: str) -> None:
        if "ROOT_CAUSE:" in response:
            self.root_cause = response.split("ROOT_CAUSE:")[1].split("\n")[0].strip()

        if "CONSTRAINTS:" in response:
            constraints_block = response.split("CONSTRAINTS:")[1]
            for marker in ("REVISED_INTENT_ADDENDUM:", "ROOT_CAUSE:"):
                if marker in constraints_block:
                    constraints_block = constraints_block.split(marker)[0]
            for line in constraints_block.strip().splitlines():
                line = line.strip().lstrip("- ").strip()
                if line:
                    self.constraints.append(line)

        if "REVISED_INTENT_ADDENDUM:" in response:
            self.revised_addendum = response.split("REVISED_INTENT_ADDENDUM:")[
                1
            ].strip()