feat(gateway): add Envelope introspection (current_state + valid_next_verbs)

Pre-fix, agents had no way to introspect what verbs were valid from a
task's current state — the 2026-05-08 trace showed them spamming
escalate_to_ceo/complete/unblock/resume against a `claimed` task and
racking up rejections. Pre-gateway agents could ground reasoning in
VALID_TRANSITIONS[status] from a doc; the gateway hid that.

Now every Envelope carries:
  - current_state: the task's status string (or None for tool-discovery
    envelopes that aren't task-bound)
  - valid_next_verbs: the verbs the caller can usefully call right now,
    sourced from verb_gates.valid_next_verbs(role, task)

Wired into i_will_work_on (highest-traffic verb) for both the OK path
and the wrong-state rejection path. Remaining lifecycle verbs to be
wired in subsequent commits (Task 3.9).
This commit is contained in:
Renn F
2026-05-08 08:03:32 +02:00
parent b4ec19ca9c
commit fd344511a5
4 changed files with 144 additions and 1 deletions
@@ -995,3 +995,57 @@ async def test_submit_up_not_assigned_rejected() -> None:
body = env.as_dict()
assert body["error"] == "not_authorized"
assert "not assigned" in body["message"]
# ---------------------------------------------------------------------------
# Task 3: Envelope introspection — verb returns carry current_state +
# valid_next_verbs so agents stop trial-and-erroring.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_i_will_work_on_envelope_carries_introspection_on_success() -> None:
"""Successful claim+start path stamps current_state + valid_next_verbs."""
agent_id = uuid4()
task_id = uuid4()
task_svc = _wire_dev_task_svc(task_id, status="pending", assigned_to=agent_id)
claimed_task = MagicMock(
status="in_progress",
assigned_to=agent_id,
plan="ok plan",
id=task_id,
title="t",
task_type="code",
)
task_svc.claim.return_value = claimed_task
task_svc.set_plan.return_value = claimed_task
task_svc.start.return_value = claimed_task
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_work_on(agent_id, task_id, plan="ok plan")
body = env.as_dict()
assert body["error"] is None
assert body["current_state"] == "in_progress"
assert isinstance(body["valid_next_verbs"], list)
assert "commit" in body["valid_next_verbs"]
assert "i_am_done" in body["valid_next_verbs"]
@pytest.mark.asyncio
async def test_i_will_work_on_envelope_carries_introspection_on_rejection() -> None:
"""A wrong-state rejection still stamps current_state + valid_next_verbs
so the agent learns what verbs are actually valid right now."""
agent_id = uuid4()
task_id = uuid4()
task_svc = _wire_dev_task_svc(
task_id, status="completed", assigned_to=agent_id
)
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_work_on(agent_id, task_id, plan="x")
body = env.as_dict()
assert body["error"] == "invalid_state"
assert body["current_state"] == "completed"
assert isinstance(body["valid_next_verbs"], list)
# Lifecycle verbs are NOT in the list for a completed task.
assert "i_will_work_on" not in body["valid_next_verbs"]
@@ -0,0 +1,60 @@
"""Envelope must carry current_state + valid_next_verbs after Task 3."""
from __future__ import annotations
from types import SimpleNamespace
from roboco.services.gateway.envelope import Envelope
def test_envelope_ok_carries_introspection_when_task_supplied() -> None:
task = SimpleNamespace(id="abc", status="in_progress", task_type="code")
env = Envelope.ok(
status="in_progress",
task_id="abc",
next="commit(message='...')",
).with_introspection(task=task, role="developer")
body = env.as_dict()
assert body["current_state"] == "in_progress"
assert "commit" in body["valid_next_verbs"]
assert "i_am_done" in body["valid_next_verbs"]
def test_envelope_error_carries_introspection_too() -> None:
task = SimpleNamespace(id="abc", status="claimed", task_type="code")
env = Envelope.invalid_state(
message="task is in claimed, expected awaiting_pm_review",
remediate="wait for QA + docs to complete",
).with_introspection(task=task, role="main_pm")
body = env.as_dict()
assert body["error"] == "invalid_state"
assert body["current_state"] == "claimed"
# main_pm on a claimed task DOES see `delegate` (legal in claimed),
# but the lifecycle-spam pattern verbs we want to suppress on
# `pending` are absent there. Pin the shape rather than the contents.
assert isinstance(body["valid_next_verbs"], list)
def test_envelope_without_introspection_omits_fields() -> None:
"""Backwards-compat: callers that don't supply a task get None."""
env = Envelope.ok(status="ok", task_id=None, next="continue")
body = env.as_dict()
assert body["current_state"] is None
assert body["valid_next_verbs"] is None
def test_with_introspection_returns_self_for_chaining() -> None:
task = SimpleNamespace(id="x", status="pending", task_type="code")
env = Envelope.ok(status="pending", task_id="x", next="i_will_work_on")
result = env.with_introspection(task=task, role="developer")
assert result is env
def test_with_introspection_handles_unknown_role() -> None:
"""Unknown role -> empty list, not None."""
task = SimpleNamespace(id="x", status="pending", task_type="code")
env = Envelope.ok(status="pending", task_id="x", next="???")
env.with_introspection(task=task, role="space_marine")
body = env.as_dict()
assert body["valid_next_verbs"] == []
assert body["current_state"] == "pending"