Files
roboco/roboco/agents/factories/_base.py
T
Renn F 4829f93a68 fix(gateway): unblock task claim; full Phase 0/1/2 remediation
Resolves the 100% claim-failure rate introduced by the gateway rewrite
  (commit 62bda0c plus 78 follow-ups). Live smoke runs hit
  `404 /api/v2/flow/developer/...` on every dev verb plus a manifest
  fallback that silently exposed off-role verbs to PMs — confirmed
  firing simultaneously in NAS agent logs (be-dev-1, be-pm, main-pm).

  Audit reports under docs/internal/audit_2026_05_04/ catalogue 49
  defects across gateway, services, prompts, MCP transport, substrate,
  and tests (8 detail reports + master synthesis). Six smoking guns;
  three proven in production logs.

  Phase 0 — unblock claim:
  - URL prefix /api/v2/flow/dev → /developer; slug-map board roles
    (product_owner, head_marketing) → /board (D-01)
  - _i_will_work_on AttributeError on None across pending /
    needs_revision / claimed re-entry branches (D-02)
  - Seed last_heartbeat_at in _qa_or_doc_claim (D-03)
  - Drop misleading i_have_committed verb; dev flow uses commit() (D-04)
  - Manifest mount via compose; flow_server + do_server fail loud
    instead of exposing all-verbs fallback (D-12)
  - MCP _post() surfaces envelope body on 4xx so agents see remediate
    hints (D-13)
    on git failure so retries aren't blocked by half-state (S-01)

  Phase 1 — lifecycle stability:
  - _resolve_skill falls back to AgentTable.capabilities (D-06)
  - main_pm_complete uses kwargs for escalate_to_ceo (D-07)
  - i_am_done auto-runs submit_verification when in_progress (D-08)
  - active_claimant_id wired in claim/unclaim paths — single-claimant
    invariant now functional (D-05)
  - qa_pass/qa_fail assert claimed_by parity with qa_agent_id (D-18)
  - Prompt-drift sweep: fail() shape, i_am_done(task_id, notes),
    subtask cap (12 hard / 8 soft), error-code symbology rewritten in
    base.md + per-role anti-patterns (D-10/11/29/30/31, D-37)

  Phase 2 — invariants + architecture:
  - Real-DB integration test exercising claim → in_progress → commit
    → submit_for_qa → i_am_done → awaiting_qa (P2-1)
  - choreographer.py → package; 3 of 6 role mixins extracted
    (board, doc, qa). _impl.py 2,526 → 2,080 lines (-18%). Continuation
    plan in docs/internal/audit_2026_05_04/p2_2_decompose_plan.md (P2-2)
  - Closure guards consolidated via _subtasks_not_terminal_envelope (P2-3)
  - TaskService.unclaim_for_reaper routed through canonical
    _validate_and_set_status; in_progress → pending added to
    VALID_TRANSITIONS (P2-4)
  - Dead code removed: i_am_done_with_catchup verb, _run_catch_up helper
    (P2-5)
  - 6 state-machine invariants asserted via property test (P2-6)
  - attempt_id (uuid4) stamped on every gateway.rejected audit row (P2-7)
  - _reconcile_orphan_claims_on_startup rolls back tasks left CLAIMED
    with branch_name=NULL from prior crashes (P2-8)
  - scripts/regenerate_verb_tables.py introspects Pydantic schemas +
    role_config; compose_prompt injects per-role tables as a layer.
    Eliminates the prompt-drift class structurally (P2-9)

  Other:
  - D-48: orchestrator mounts host's ~/.claude.json when present so
    agents don't boot from backup recovery on every spawn
  - D-49: dev dispatcher rejects role-mismatched spawns (e.g. doc task
    assigned to dev agent)

  Tests: 553 pass · ruff + mypy clean. Live NAS smoke verification
  pending — needs the stack brought back up.
2026-05-04 23:43:55 +02:00

174 lines
5.2 KiB
Python

"""
Factory Base Utilities
Shared utilities for agent factory functions.
"""
import re
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from roboco.models import AgentRole, Team
def _get_prompts_base_path() -> Path:
"""Get the base path for prompt layers.
Resolves to project_root/agents/prompts/ by finding the roboco package
and going up one level.
"""
# Start from this file's location: roboco/agents/factories/_base.py
# Go up: factories -> agents -> roboco -> project_root
this_file = Path(__file__).resolve()
project_root = this_file.parent.parent.parent.parent
prompts_path = project_root / "agents" / "prompts"
# Fallback: try relative path if above doesn't exist
if not prompts_path.exists():
prompts_path = Path("agents/prompts")
return prompts_path
# Base path for prompt layers
PROMPTS_BASE_PATH = _get_prompts_base_path()
def _load_layer(layer_path: Path) -> str:
"""
Load a prompt layer file.
Args:
layer_path: Path to the layer markdown file
Returns:
File content or empty string if not found
"""
if not layer_path.exists():
return ""
return layer_path.read_text().strip()
_ROLE_LAYER_MAP: dict[str, str] = {
# Cell members
"developer": "developer.md",
"qa": "qa.md",
"documenter": "documenter.md",
# PMs have separate files
"main_pm": "main_pm.md",
"cell_pm": "cell_pm.md",
# Board members use board layer
"product_owner": "board.md",
"head_marketing": "board.md",
"auditor": "board.md",
}
_TEAM_LAYER_MAP: dict[str, str] = {
"backend": "backend.md",
"frontend": "frontend.md",
"ux_ui": "ux_ui.md",
}
def _role_layer(prompts_path: Path, role: "AgentRole") -> str | None:
"""Load the role-specific prompt layer, or None if unknown."""
role_value = role.value if hasattr(role, "value") else str(role)
role_file = _ROLE_LAYER_MAP.get(role_value)
if not role_file:
return None
return _load_layer(prompts_path / "roles" / role_file)
def _team_layer(prompts_path: Path, team: "Team | None") -> str | None:
"""Load the team-specific prompt layer, or None if unset/unknown."""
if not team:
return None
team_value = team.value if hasattr(team, "value") else str(team)
team_file = _TEAM_LAYER_MAP.get(team_value)
if not team_file:
return None
return _load_layer(prompts_path / "teams" / team_file)
def _autogen_verbs_layer(prompts_path: Path, role: "AgentRole") -> str | None:
"""Load the autogenerated verb signatures for this role.
The file at ``_generated/<role>.md`` is rebuilt by
``scripts/regenerate_verb_tables.py`` from the Pydantic schemas in
``roboco/api/schemas/v2/`` plus the role-config. Including it as a
composition layer means the prompt always shows the literal accepted
body shape for every verb the agent has — eliminating the prompt
drift class identified in audit P2-9 (D-04, D-10, D-11, D-29-D-31).
"""
role_value = role.value if hasattr(role, "value") else str(role)
return _load_layer(prompts_path / "_generated" / f"{role_value}.md")
def compose_prompt(
role: "AgentRole",
team: "Team | None",
agent_slug: str,
base_path: Path | None = None,
) -> str:
"""
Compose a system prompt from layered components.
Combines:
1. base.md - Universal rules (all agents)
2. roles/{role}.md - Role-specific behavior
3. _generated/{role}.md - Autogenerated verb-signature table from schemas
4. teams/{team}.md - Team context (if team is set)
5. identities/{agent_slug}.md - Agent identity
Args:
role: Agent's role (developer, qa, pm, documenter, board)
team: Agent's team (backend, frontend, ux_ui, or None for board)
agent_slug: Agent's slug identifier (e.g., "be-dev-1")
base_path: Optional override for prompts base path
Returns:
Composed system prompt string
"""
prompts_path = base_path or PROMPTS_BASE_PATH
parts: list[str] = []
for layer in (
_load_layer(prompts_path / "base.md"),
_role_layer(prompts_path, role),
_autogen_verbs_layer(prompts_path, role),
_team_layer(prompts_path, team),
_load_layer(prompts_path / "identities" / f"{agent_slug}.md"),
):
if layer:
parts.append(layer)
# Join with separator
return "\n\n---\n\n".join(parts)
def load_blueprint_prompt(blueprint_path: str, default_prompt: str) -> str:
"""
Load system prompt from a blueprint file.
Args:
blueprint_path: Relative path to the blueprint markdown file
default_prompt: Default prompt if file doesn't exist or parsing fails
Returns:
The extracted system prompt or the default
"""
path = Path(blueprint_path)
if not path.exists():
return default_prompt
content = path.read_text()
# Extract system prompt section (between ```blocks after ## System Prompt)
match = re.search(r"## System Prompt\s*```\s*(.*?)```", content, re.DOTALL)
return match.group(1).strip() if match else default_prompt
def make_slug(name: str) -> str:
"""Convert a name to a URL-safe slug."""
return name.lower().replace(" ", "-")