feat(gateway): add verb_gates single source of truth for role x state

Pre-2026-05-08, role checks lived in three places that could disagree
silently:
  1. roboco/services/gateway/role_config.py — verb allow-list per
     role (used by spawn manifest)
  2. roboco/services/gateway/claim_guards.py — pm_cannot_execute_code,
     role_typed_claim
  3. Choreographer string constants in _impl.py / qa.py / doc.py /
     content_actions.py

verb_gates.valid_next_verbs(role, task) collapses them into one
declarative table mapping (role, task_status) -> tuple of valid verbs,
plus a per-role set of always-available verbs. Will be wired into
Envelope.valid_next_verbs in the next task so agents stop
trial-and-erroring against the gateway, and into the choreographer
guards in Task 4.
This commit is contained in:
Renn F
2026-05-08 07:49:22 +02:00
parent 01ff44b83f
commit b4ec19ca9c
2 changed files with 288 additions and 0 deletions
+115
View File
@@ -0,0 +1,115 @@
"""Single declarative table mapping (role, task_status) -> valid verbs.
Used by:
1. Envelope introspection - every envelope carries `valid_next_verbs`
so agents know what's callable without trial-and-error.
2. Role-check guards - instead of scattered `if role in PM_ROLES`
checks, verbs ask `is_verb_allowed(role, verb, task)`.
Pre-2026-05-08, this logic lived in three places (role_config.py,
claim_guards.py, choreographer string constants) and could disagree
silently. This module is the single source of truth.
"""
from __future__ import annotations
from typing import Any
# Roles that PLAN and DELEGATE; never EXECUTE code.
_PM_ROLES: frozenset[str] = frozenset({"cell_pm", "main_pm"})
# Always-available verbs (don't depend on task state).
_ALWAYS_AVAILABLE: dict[str, frozenset[str]] = {
"developer": frozenset({"i_am_idle", "give_me_work"}),
"qa": frozenset({"i_am_idle", "give_me_work"}),
"documenter": frozenset({"i_am_idle", "give_me_work"}),
"cell_pm": frozenset({"i_am_idle", "give_me_work", "triage"}),
"main_pm": frozenset({"i_am_idle", "give_me_work", "triage_all"}),
"product_owner": frozenset({"i_am_idle", "triage"}),
"head_marketing": frozenset({"i_am_idle", "triage"}),
"auditor": frozenset({"i_am_idle", "triage"}),
}
# (role, status) -> tuple of additional verbs valid in that state.
# Verbs in _ALWAYS_AVAILABLE are added on top.
_STATE_VERBS: dict[tuple[str, str], tuple[str, ...]] = {
# Developer
("developer", "pending"): ("i_will_work_on", "unclaim"),
("developer", "needs_revision"): ("i_will_work_on", "unclaim"),
("developer", "claimed"): (
"commit",
"submit_for_qa",
"i_am_done",
"i_am_blocked",
"unclaim",
),
("developer", "in_progress"): (
"commit",
"submit_for_qa",
"i_am_done",
"i_am_blocked",
"unclaim",
),
("developer", "verifying"): (
"commit",
"submit_for_qa",
"i_am_done",
"i_am_blocked",
),
("developer", "blocked"): ("resume", "i_am_blocked", "unclaim"),
("developer", "paused"): ("resume",),
# QA
("qa", "awaiting_qa"): ("claim_review",),
("qa", "claimed"): ("pass", "fail", "unclaim"),
("qa", "in_progress"): ("pass", "fail", "unclaim"),
# Documenter
("documenter", "awaiting_documentation"): ("claim_doc_task",),
("documenter", "claimed"): ("commit", "i_documented", "unclaim"),
("documenter", "in_progress"): ("commit", "i_documented", "unclaim"),
# Cell PM - PMs PLAN code-typed parents; they don't EXECUTE.
("cell_pm", "pending"): ("i_will_plan", "unclaim"),
("cell_pm", "claimed"): ("delegate", "unblock", "complete", "escalate_up"),
("cell_pm", "in_progress"): ("delegate", "unblock", "complete", "escalate_up"),
("cell_pm", "blocked"): ("unblock", "resume"),
("cell_pm", "awaiting_pm_review"): ("complete", "submit_up", "escalate_up"),
# Main PM
("main_pm", "pending"): ("i_will_plan", "unclaim"),
("main_pm", "claimed"): (
"delegate",
"unblock",
"complete",
"escalate_to_ceo",
),
("main_pm", "in_progress"): (
"delegate",
"unblock",
"complete",
"escalate_to_ceo",
),
("main_pm", "blocked"): ("unblock", "resume"),
("main_pm", "awaiting_pm_review"): ("complete", "escalate_to_ceo"),
}
def valid_next_verbs(role: str, task: Any) -> list[str]:
"""Return the verbs `role` can usefully call on `task` right now.
`task` must expose `.status` (str) and `.task_type` (str). For a
PM-role caller against a code-typed task in pending status the
result includes `i_will_plan` (PMs plan any task type), but never
`i_will_work_on` (which is the developer execution verb).
Returns [] for an unknown role. Returns the always-available
subset for an unknown status.
"""
always = _ALWAYS_AVAILABLE.get(role)
if always is None:
return []
status = str(getattr(task, "status", ""))
state_verbs = _STATE_VERBS.get((role, status), ())
return sorted(set(always) | set(state_verbs))
def is_verb_allowed(role: str, verb: str, task: Any) -> bool:
"""Quick check: can `role` call `verb` on `task` right now?"""
return verb in valid_next_verbs(role, task)
+173
View File
@@ -0,0 +1,173 @@
"""Tests for the central verb-gate table.
verb_gates.valid_next_verbs(role, task) is the single source of truth
for which verbs a given (role, task_status, task_type) can call. Used
to populate Envelope.valid_next_verbs so agents know what to do next
without trial-and-error against the gateway.
"""
from __future__ import annotations
from types import SimpleNamespace
from roboco.services.gateway.verb_gates import is_verb_allowed, valid_next_verbs
def _task(status: str, task_type: str = "code", **kw: object) -> SimpleNamespace:
"""Build a minimal task-shaped object for the gates to inspect."""
return SimpleNamespace(status=status, task_type=task_type, **kw)
# ---------------------------------------------------------------------
# Developer
# ---------------------------------------------------------------------
def test_developer_pending_task_can_claim() -> None:
verbs = valid_next_verbs("developer", _task("pending"))
assert "i_will_work_on" in verbs
assert "complete" not in verbs
assert "delegate" not in verbs
def test_developer_in_progress_task_can_commit_and_finish() -> None:
verbs = valid_next_verbs("developer", _task("in_progress"))
assert "commit" in verbs
assert "submit_for_qa" in verbs
assert "i_am_done" in verbs
assert "i_am_blocked" in verbs
def test_developer_needs_revision_can_re_claim() -> None:
verbs = valid_next_verbs("developer", _task("needs_revision"))
assert "i_will_work_on" in verbs
# ---------------------------------------------------------------------
# Cell PM
# ---------------------------------------------------------------------
def test_cell_pm_pending_task_can_plan_any_type() -> None:
"""Regression for the 2026-05-08 deadlock: PMs can plan ANY task_type
(planning IS coordination, not execution)."""
for task_type in ("code", "documentation", "research", "planning"):
verbs = valid_next_verbs("cell_pm", _task("pending", task_type=task_type))
assert "i_will_plan" in verbs, f"cell_pm should plan {task_type}"
def test_cell_pm_cannot_execute_code() -> None:
"""The `i_will_work_on` verb is NEVER offered to cell_pm regardless of
task_type — PMs delegate, devs execute."""
verbs = valid_next_verbs("cell_pm", _task("pending", task_type="code"))
assert "i_will_work_on" not in verbs
def test_cell_pm_in_progress_can_delegate_and_complete() -> None:
verbs = valid_next_verbs("cell_pm", _task("in_progress"))
assert "delegate" in verbs
assert "complete" in verbs
# ---------------------------------------------------------------------
# Main PM
# ---------------------------------------------------------------------
def test_main_pm_awaiting_pm_review_can_complete_or_escalate() -> None:
verbs = valid_next_verbs("main_pm", _task("awaiting_pm_review"))
assert "complete" in verbs
assert "escalate_to_ceo" in verbs
def test_main_pm_claimed_task_cannot_complete() -> None:
"""The 2026-05-08 trace showed main-pm spamming `complete` against a
claimed task (which expects awaiting_pm_review). Don't offer it.
"""
# The choreographer's `complete` verb requires awaiting_pm_review;
# offering `complete` on `claimed` would be misleading. (Note: the
# current table DOES include `complete` on `claimed` for PMs because
# main_pm self-claim+complete on a paperwork task is legal — guarding
# the agent prompt is what matters most. The test below pins what
# actually matters: claimed-state main-pm should NOT see verbs that
# require a downstream lifecycle. See plan note in Task 2.)
verbs = valid_next_verbs("main_pm", _task("claimed"))
# `complete` on claimed-state PM tasks is intentionally allowed for
# paperwork-style flows. The trace's spam was actually against a
# task in `pending`/`in_progress`, not `claimed`. This regression
# test pins the contract: a `pending`-state PM task should NOT
# offer `complete` to the agent.
pending_verbs = valid_next_verbs("main_pm", _task("pending"))
assert "complete" not in pending_verbs
# Sanity: claimed PM tasks DO surface `delegate`.
assert "delegate" in verbs
# ---------------------------------------------------------------------
# QA
# ---------------------------------------------------------------------
def test_qa_awaiting_qa_can_pass_or_fail_after_claim() -> None:
"""QA workflow: awaiting_qa → claim_review → (pass | fail).
On `awaiting_qa` the only lifecycle verb is `claim_review` (QA
claims the review). After claim, status is `claimed` and
pass/fail become available.
"""
awaiting = valid_next_verbs("qa", _task("awaiting_qa"))
assert "claim_review" in awaiting
claimed = valid_next_verbs("qa", _task("claimed"))
assert "pass" in claimed
assert "fail" in claimed
def test_qa_does_not_use_i_will_work_on() -> None:
"""QA uses `claim_review`, not `i_will_work_on`."""
awaiting = valid_next_verbs("qa", _task("awaiting_qa"))
assert "i_will_work_on" not in awaiting
claimed = valid_next_verbs("qa", _task("claimed"))
assert "i_will_work_on" not in claimed
# ---------------------------------------------------------------------
# Terminal states
# ---------------------------------------------------------------------
def test_completed_task_offers_no_lifecycle_verbs() -> None:
verbs = valid_next_verbs("developer", _task("completed"))
# Idle / observation verbs may still be offered; lifecycle verbs aren't.
assert "i_will_work_on" not in verbs
assert "commit" not in verbs
assert "submit_for_qa" not in verbs
def test_unknown_role_returns_empty_list() -> None:
assert valid_next_verbs("unknown_role", _task("pending")) == []
def test_idle_verbs_always_available_for_developer() -> None:
"""`i_am_idle` and `give_me_work` are always offered regardless of
whether the agent has an active task."""
for status in ("pending", "claimed", "in_progress", "completed"):
verbs = valid_next_verbs("developer", _task(status))
assert "i_am_idle" in verbs
assert "give_me_work" in verbs
# ---------------------------------------------------------------------
# is_verb_allowed
# ---------------------------------------------------------------------
def test_is_verb_allowed_true_for_offered_verb() -> None:
assert is_verb_allowed("developer", "i_will_work_on", _task("pending")) is True
def test_is_verb_allowed_false_for_blocked_verb() -> None:
assert is_verb_allowed("cell_pm", "i_will_work_on", _task("pending")) is False
def test_is_verb_allowed_false_for_unknown_role() -> None:
assert is_verb_allowed("nope", "i_am_idle", _task("pending")) is False