diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 62008a3e..12bb7c9e 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -2190,7 +2190,6 @@ 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 @@ -2256,88 +2255,6 @@ 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____`). 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='note') 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.""" @@ -2403,7 +2320,6 @@ 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) @@ -2414,7 +2330,6 @@ 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" diff --git a/tests/unit/runtime/test_briefing_verb_server_map.py b/tests/unit/runtime/test_briefing_verb_server_map.py deleted file mode 100644 index 92366ec1..00000000 --- a/tests/unit/runtime/test_briefing_verb_server_map.py +++ /dev/null @@ -1,181 +0,0 @@ -"""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.foundation.policy.journaling import SCOPE_TO_TYPE, Scope -from roboco.foundation.policy.tracing import Requirement -from roboco.models.base import JournalEntryType -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: - # The i_will_work_on claim gate is journal:note_at_claim, enforced via - # JournalService.has_note_for_task -> JournalEntryType.GENERAL. The only - # note scope that maps to GENERAL is scope='note' (scope='decision' maps - # to DECISION_LOG and is the PM's i_will_plan gate). The briefing MUST - # name the scope that actually satisfies the gate, not 'decision'. - block = _orch()._build_verb_server_block("developer") - assert "note(scope='note')" in block - assert "i_will_work_on" in block.split("note(scope='note')", 1)[1] - # The decision scope must NOT be the dev-claim precondition. - assert "note(scope='decision')" not in block - - -def test_developer_claim_precondition_scope_matches_real_gate_type() -> None: - # Validate against the REAL boundary: the JOURNAL_NOTE_AT_CLAIM requirement - # on i_will_work_on is satisfied by has_note_for_task, which queries - # JournalEntryType.GENERAL. Whatever scope the briefing emits MUST be the - # scope whose SCOPE_TO_TYPE mapping equals the type the gate checks. - assert Requirement.JOURNAL_NOTE_AT_CLAIM.value == "journal:note_at_claim" - gate_type = JournalEntryType.GENERAL # has_note_for_task checks this type - satisfying_scopes = { - scope.value for scope, jtype in SCOPE_TO_TYPE.items() if jtype is gate_type - } - assert satisfying_scopes == {Scope.NOTE.value} - - block = _orch()._build_verb_server_block("developer") - for scope_value in satisfying_scopes: - assert f"note(scope='{scope_value}')" in block - # A scope that does NOT map to the gate type must not be presented as the - # i_will_work_on precondition. - non_satisfying = {s.value for s in Scope} - satisfying_scopes - for scope_value in non_satisfying: - claim_clause = f"note(scope='{scope_value}') is REQUIRED before i_will_work_on" - assert claim_clause not in block - - -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_pm_plan_precondition_stays_decision_scope() -> None: - # The PM's i_will_plan gate is journal:decision_at_claim, enforced via - # has_decision_for_task -> JournalEntryType.DECISION_LOG. The only scope - # that maps to DECISION_LOG is scope='decision', so the i_will_plan - # precondition must keep scope='decision' (this branch is correct). - decision_scopes = { - scope.value - for scope, jtype in SCOPE_TO_TYPE.items() - if jtype is JournalEntryType.DECISION_LOG - } - assert decision_scopes == {Scope.DECISION.value} - - block = _orch()._build_verb_server_block("cell_pm") - assert "note(scope='decision')" in block - assert "i_will_plan" in block.split("note(scope='decision')", 1)[1] - - -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_head_marketing_block_lists_docs_server() -> None: - # Head of Marketing is handed the roboco-docs MCP at spawn for read-only - # oversight; the briefing must surface it (and the service READ_ROLES must - # agree, or list/read 403 against a tool the agent was given). - block = _orch()._build_verb_server_block("head_marketing") - assert "roboco-docs" in block - assert "roboco_docs_read" in block - - -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 - # Developer briefing: the claim precondition is scope='note' (the GENERAL - # entry has_note_for_task checks), never scope='decision'. - assert "note(scope='note')" in content - assert "note(scope='decision')" not in content