feat(orchestrator): embed verb->MCP-server map + key preconditions in agent briefing

Agents fumble their first move — raw bash/http/shell-git, calling
evidence on roboco-flow when it lives on roboco-do, omitting the nature
argument on delegate, or skipping the required journal note before
claiming. The role docs cover this but agents cannot read them at spawn.

Generate a concise, role-accurate block from the role's actual manifest
(get_role_config): a verb->server map (roboco-flow / roboco-do /
roboco-git-readonly / roboco-optimal / roboco-docs) plus key
preconditions (note(scope='decision') before i_will_work_on / i_will_plan;
delegate requires nature; evidence is on roboco-do; never use raw
bash/http/shell-git). Embed it into the written session briefing.
This commit is contained in:
Renn F
2026-06-03 19:03:22 +02:00
parent 78faf6f4da
commit 3d04943bf3
2 changed files with 204 additions and 0 deletions
+85
View File
@@ -2190,6 +2190,7 @@ class AgentOrchestrator:
)
_TOOL_LOAD_CACHE: ClassVar[dict[str, str]] = {}
_VERB_SERVER_CACHE: ClassVar[dict[str, str]] = {}
# Per-role built-in tools, enumerated in the briefing so the agent
# knows exactly what it has. These are pre-loaded at spawn via the
@@ -2255,6 +2256,88 @@ class AgentOrchestrator:
self._TOOL_LOAD_CACHE[role] = block
return block
# Roles whose containers also mount the docs MCP server. Mirrors the
# gating in the MCP-server registration so the verb-server map stays
# accurate without re-deriving it.
_DOCS_SERVER_ROLES: ClassVar[tuple[str, ...]] = (
"documenter",
"cell_pm",
"main_pm",
"product_owner",
"head_marketing",
)
def _build_verb_server_block(self, role: str) -> str:
"""Briefing block: which MCP server hosts each verb + key preconditions.
Agents fumble their first move — raw bash/http/shell-git, calling
``evidence`` on roboco-flow when it lives on roboco-do, omitting the
``nature`` argument on ``delegate``, or skipping the required journal
note before claiming. The role docs cover this but agents cannot read
them at spawn, so the map is generated here from the role's actual
manifest (``get_role_config``) and stays accurate as the spec changes.
Cached per role.
"""
from roboco.services.gateway.role_config import ROLE_CONFIGS, get_role_config
if role in self._VERB_SERVER_CACHE:
return self._VERB_SERVER_CACHE[role]
if role not in ROLE_CONFIGS:
self._VERB_SERVER_CACHE[role] = ""
return ""
cfg = get_role_config(role)
lines = [
"## Which MCP server hosts each verb",
"",
"Call the verb on the right server — the server name is the MCP",
"tool prefix (`mcp__<server>__<verb>`). Never reach for raw bash,",
"raw http, or shell git (`git commit`/`push`/`checkout`); the",
"bash-guard blocks them. Use these verbs instead:",
"",
f"- **roboco-flow** (intent verbs): {', '.join(cfg.flow_tools)}",
f"- **roboco-do** (content tools): {', '.join(cfg.do_tools)}",
"- **roboco-git-readonly** (read-only git): roboco_git_status,"
" roboco_git_log, roboco_git_diff, roboco_git_branch_list",
"- **roboco-optimal** (knowledge base): roboco_ask_mentor,"
" roboco_kb_search",
]
if role in self._DOCS_SERVER_ROLES:
lines.append(
"- **roboco-docs** (project docs files): roboco_docs_read,"
" roboco_docs_write, roboco_docs_list"
)
preconditions = [
"`evidence` lives on roboco-do, NOT roboco-flow — inspect a task"
" there before acting on it.",
]
if "i_will_work_on" in cfg.flow_tools:
preconditions.append(
"note(scope='decision') is REQUIRED before i_will_work_on —"
" log your approach first or the claim is rejected."
)
if "i_will_plan" in cfg.flow_tools:
preconditions.append(
"note(scope='decision') is REQUIRED before i_will_plan /"
" complete / escalate — log the decision first."
)
if "delegate" in cfg.flow_tools:
preconditions.append(
"delegate requires `nature` (one of: technical |"
" non_technical) — omitting it is rejected."
)
lines.append("")
lines.append("### Key preconditions")
lines.extend(f"- {p}" for p in preconditions)
lines.append("")
lines.append("")
block = "\n".join(lines)
self._VERB_SERVER_CACHE[role] = block
return block
@staticmethod
def _format_task_briefing_block(task_id: str, task: dict[str, Any]) -> str:
"""Build the ``## Current task`` markdown block from a fetched task."""
@@ -2320,6 +2403,7 @@ class AgentOrchestrator:
escalate_to = get_escalation_target(agent_id) or "main-pm"
tool_load_block = self._build_tool_load_block(role)
verb_server_block = self._build_verb_server_block(role)
task_block = ""
if task_id:
task = await self._fetch_task_for_briefing(agent_id, task_id)
@@ -2330,6 +2414,7 @@ class AgentOrchestrator:
f"# Session briefing — {agent_id}\n"
"\n"
f"{tool_load_block}"
f"{verb_server_block}"
"## You are\n"
f"- **Agent:** `{agent_id}`\n"
f"- **Role:** {role}\n"
@@ -0,0 +1,119 @@
"""The session briefing must carry a verb->MCP-server map + key preconditions.
Agents repeatedly fumble their first move: calling raw bash/http/shell-git,
invoking ``evidence`` on roboco-flow (it lives on roboco-do), omitting the
``nature`` argument on ``delegate``, or skipping the required journal note
before claiming. The role docs cover this but agents cannot read them at
spawn, so the briefing embeds a concise, role-accurate block generated from
the role's actual manifest (``get_role_config``).
"""
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from unittest.mock import patch
from roboco.runtime.orchestrator import AgentOrchestrator
from roboco.services.gateway.role_config import get_role_config
def _orch() -> AgentOrchestrator:
with patch.object(AgentOrchestrator, "__init__", return_value=None):
orch = AgentOrchestrator.__new__(AgentOrchestrator)
orch._VERB_SERVER_CACHE = {}
return orch
def test_developer_block_maps_flow_verbs_to_flow_server() -> None:
block = _orch()._build_verb_server_block("developer")
cfg = get_role_config("developer")
# Every flow verb the role can call is attributed to roboco-flow.
flow_section = block.split("roboco-flow", 1)[1].split("roboco-do", 1)[0]
for verb in cfg.flow_tools:
assert verb in flow_section, f"{verb} missing from roboco-flow line"
def test_developer_block_puts_evidence_on_do_not_flow() -> None:
block = _orch()._build_verb_server_block("developer")
do_section = block.split("roboco-do", 1)[1].split("roboco-git-readonly", 1)[0]
flow_section = block.split("roboco-flow", 1)[1].split("roboco-do", 1)[0]
assert "evidence" in do_section
assert "evidence" not in flow_section
def test_developer_block_lists_git_readonly_and_optimal_servers() -> None:
block = _orch()._build_verb_server_block("developer")
assert "roboco-git-readonly" in block
assert "roboco-optimal" in block
assert "roboco_ask_mentor" in block
def test_developer_block_states_note_before_claim_precondition() -> None:
block = _orch()._build_verb_server_block("developer")
assert "note(scope='decision')" in block
assert "i_will_work_on" in block.split("note(scope='decision')", 1)[1]
def test_developer_block_forbids_raw_bash_http_shell_git() -> None:
block = _orch()._build_verb_server_block("developer")
lowered = block.lower()
assert "shell git" in lowered or "shell-git" in lowered
assert "raw" in lowered
def test_pm_block_states_delegate_requires_nature() -> None:
block = _orch()._build_verb_server_block("cell_pm")
assert "delegate" in block
assert "nature" in block
def test_qa_block_has_no_delegate_or_note_before_claim_noise() -> None:
# QA has no delegate verb, so the nature precondition must not appear;
# QA has no claim-with-plan verb, so note-before-claim must not appear.
block = _orch()._build_verb_server_block("qa")
assert "delegate" not in block
assert "note(scope='decision')" not in block
# But it must still carry the no-raw-bash rule and its own flow verbs.
assert "pass_review" in block
assert "shell" in block.lower()
def test_unknown_role_returns_empty() -> None:
assert _orch()._build_verb_server_block("nonexistent") == ""
def test_block_is_cached_per_role() -> None:
orch = _orch()
first = orch._build_verb_server_block("developer")
second = orch._build_verb_server_block("developer")
assert first is second
def test_block_is_embedded_in_written_briefing() -> None:
orch = _orch()
orch._VERB_SERVER_CACHE = {}
orch._TOOL_LOAD_CACHE = {}
with (
patch("roboco.runtime.orchestrator.get_agent_role", return_value="developer"),
patch("roboco.runtime.orchestrator.get_agent_team", return_value="backend"),
patch(
"roboco.runtime.orchestrator.get_escalation_target", return_value="be-pm"
),
patch("roboco.runtime.orchestrator.PROJECT_HOST_PATH", None),
patch(
"roboco.runtime.orchestrator.tempfile.gettempdir",
return_value=tempfile.gettempdir(),
),
):
path = asyncio.run(
orch._write_agent_briefing("be-dev-1", None, "/data/workspaces/x")
)
assert path is not None
content = Path(path).read_text()
assert "roboco-flow" in content
assert "roboco-do" in content
assert "note(scope='decision')" in content