feat(gateway): add resume verb for paused -> in_progress

i_am_idle auto-pauses owned in_progress tasks; lifecycle table allows
paused -> in_progress; no verb implemented it. Adds resume so an agent
respawned for a paused task can continue. Routes through
_validate_and_set_status mirroring the unclaim pattern (Task 9 fix).
This commit is contained in:
Renn F
2026-05-03 07:05:16 +02:00
parent bf3de1ab2e
commit cd9f999533
16 changed files with 355 additions and 4 deletions
+1
View File
@@ -28,6 +28,7 @@ You merge what your developers submit (leaf PRs into your cell branch via `compl
| `submit_up(task_id, notes)` | Open your cell-level PR up to Main PM's branch; transition YOUR task to `awaiting_pm_review`. | All your subtasks terminal; `notes` >= 20 chars; journal `decision` recorded. |
| `escalate_up(task_id, reason)` | Escalate to Main PM. | Task is yours or assigned to your cell. |
| `unclaim(task_id)` | Release this claim back to pending. Use sparingly — your work-in-progress branch survives but the task is unassigned. | Task assigned to you and in claimed/in_progress. |
| `resume(task_id)` | Resume a paused task. Transitions paused → in_progress. | Task assigned to you and in paused state. |
| `note(text, scope?, task_id?)` | Journal. Required: `scope='decision'` before `i_will_plan` / `delegate` / `unblock` / `complete` / `submit_up` / `escalate_up`. | None. |
| `say(channel, text)` / `dm(recipient, text)` | Channel post / DM. Channel slug without `#` (e.g. `"backend-cell"`). | None. |
| `evidence(task_id)` | Inspect a task's PR + commits + diff. | None. |
+1
View File
@@ -25,6 +25,7 @@ You write code; you do not coordinate. If you find yourself thinking "let me als
| `i_am_done(notes)` | Strict submit for QA. Requires PR already open — run `submit_for_qa` first. | Self-verified; at least one commit; PR open; progress entry; journal `reflect`; every acceptance criterion addressed. |
| `i_am_blocked(reason)` | Records the blocker, escalates to your PM, idles you. | Task is yours and active. |
| `unclaim(task_id)` | Release this claim back to pending. Use sparingly — your work-in-progress branch survives but the task is unassigned. | Task assigned to you and in claimed/in_progress. |
| `resume(task_id)` | Resume a paused task. Transitions paused → in_progress. | Task assigned to you and in paused state. |
| `note(text, scope?)` | Journal entry (`scope ∈ note|decision|reflect|learning|struggle`). | None. |
| `say(channel, text)` / `dm(recipient, text, skill?)` | Channel post / direct message. | Channel slug without `#`. |
| `evidence(task_id)` | Fetches PR diff, commits, files changed, dev summary. | None. |
+1
View File
@@ -22,6 +22,7 @@ You do NOT re-implement the developer's work. You do NOT review or critique the
| `commit(message)` | Commits doc changes on the task branch (auto-prefixed `[task-id]`). | Task in `in_progress`; on the task branch. |
| `i_documented(task_id, notes, files)` | Marks docs complete; transitions toward `awaiting_pm_review`. | At least one doc file in `files`; `notes` >= 20 chars. |
| `unclaim(task_id)` | Release this claim back to pending. Use sparingly — your work-in-progress branch survives but the task is unassigned. | Task assigned to you and in claimed/in_progress. |
| `resume(task_id)` | Resume a paused task. Transitions paused → in_progress. | Task assigned to you and in paused state. |
| `note(text, scope?)` | Journal entry. | None. |
| `say(channel, text)` / `dm(recipient, text, skill?)` | Channel post / direct message. | Channel slug without `#`. |
| `evidence(task_id)` | Re-fetches PR diff and commits if needed. | None. |
+1
View File
@@ -28,6 +28,7 @@ You merge what your Cell PMs submit (cell PRs into your root branch via `complet
| `escalate_up(task_id, reason)` | Escalate a stuck task up your chain to CEO. | Task is yours or assigned to a cell under your scope. |
| `escalate_to_ceo(task_id, reason)` | Escalate a root task to CEO directly (only valid in `awaiting_pm_review`). | Root task in `awaiting_pm_review`; `pr_number` set. |
| `unclaim(task_id)` | Release this claim back to pending. Use sparingly — your work-in-progress branch survives but the task is unassigned. | Task assigned to you and in claimed/in_progress. |
| `resume(task_id)` | Resume a paused task. Transitions paused → in_progress. | Task assigned to you and in paused state. |
| `note(text, scope?, task_id?)` | Journal. Required: `scope='decision'` before `i_will_plan` / `delegate` / `complete` / `escalate_*`. | None. |
| `say(channel, text)` / `dm(recipient, text)` | Channel post / DM. Channel slug without `#` (e.g. `"main-pm-board"`). | None. |
| `evidence(task_id)` | Inspect a task's PR + commits + diff. | None. |
+1
View File
@@ -21,6 +21,7 @@ A pass without evidence is a betrayal of your role: the entire downstream chain
| `pass(task_id, notes)` | Accepts the work; transitions to `awaiting_documentation`. | Task claimed by you; `notes` >= 80 chars; journal `learning` entry recorded. |
| `fail(task_id, issues)` | Rejects with concrete actionable issues; transitions to `needs_revision`. | Task claimed by you; each issue references criterion/file/line. |
| `unclaim(task_id)` | Release this claim back to pending. Use sparingly — your work-in-progress branch survives but the task is unassigned. | Task assigned to you and in claimed/in_progress. |
| `resume(task_id)` | Resume a paused task. Transitions paused → in_progress. | Task assigned to you and in paused state. |
| `note(text, scope?)` | Journal entry. Required: `scope='learning'` before `pass`/`fail`. | None. |
| `say(channel, text)` / `dm(recipient, text, skill?)` | Channel post / direct message. | Channel slug without `#`. |
| `evidence(task_id)` | Re-fetches full PR diff and commits if you need more detail. | None. |
+11
View File
@@ -14,6 +14,7 @@ from roboco.api.schemas.v2.flow import (
GiveMeWorkRequest,
IAmIdleRequest,
IWillPlanRequest,
ResumeRequest,
SubmitUpRequest,
TriageRequest,
UnblockRequest,
@@ -131,6 +132,16 @@ async def unclaim(
return env.as_dict()
@router.post("/resume")
async def resume(
body: ResumeRequest,
x_agent_id: _AgentIdHeader,
choreographer: _ChoreographerDep,
) -> dict:
env = await choreographer.resume(x_agent_id, body.task_id)
return env.as_dict()
@router.post("/i_am_idle")
async def i_am_idle(
_body: IAmIdleRequest,
+11
View File
@@ -14,6 +14,7 @@ from roboco.api.schemas.v2.flow import (
IAmIdleRequest,
IHaveCommittedRequest,
IWillWorkOnRequest,
ResumeRequest,
SubmitForQaRequest,
UnclaimRequest,
)
@@ -100,6 +101,16 @@ async def unclaim(
return env.as_dict()
@router.post("/resume")
async def resume(
body: ResumeRequest,
x_agent_id: _AgentIdHeader,
choreographer: _ChoreographerDep,
) -> dict:
env = await choreographer.resume(x_agent_id, body.task_id)
return env.as_dict()
@router.post("/i_am_idle")
async def i_am_idle(
_body: IAmIdleRequest,
+11
View File
@@ -12,6 +12,7 @@ from roboco.api.schemas.v2.flow import (
GiveMeWorkRequest,
IAmIdleRequest,
IDocumentedRequest,
ResumeRequest,
UnclaimRequest,
)
from roboco.services.gateway.choreographer import Choreographer
@@ -69,6 +70,16 @@ async def unclaim(
return env.as_dict()
@router.post("/resume")
async def resume(
body: ResumeRequest,
x_agent_id: _AgentIdHeader,
choreographer: _ChoreographerDep,
) -> dict:
env = await choreographer.resume(x_agent_id, body.task_id)
return env.as_dict()
@router.post("/i_am_idle")
async def i_am_idle(
_body: IAmIdleRequest,
+11
View File
@@ -15,6 +15,7 @@ from roboco.api.schemas.v2.flow import (
GiveMeWorkRequest,
IAmIdleRequest,
IWillPlanRequest,
ResumeRequest,
TriageRequest,
UnblockRequest,
UnclaimRequest,
@@ -131,6 +132,16 @@ async def unclaim(
return env.as_dict()
@router.post("/resume")
async def resume(
body: ResumeRequest,
x_agent_id: _AgentIdHeader,
choreographer: _ChoreographerDep,
) -> dict:
env = await choreographer.resume(x_agent_id, body.task_id)
return env.as_dict()
@router.post("/i_am_idle")
async def i_am_idle(
_body: IAmIdleRequest,
+11
View File
@@ -13,6 +13,7 @@ from roboco.api.schemas.v2.flow import (
GiveMeWorkRequest,
IAmIdleRequest,
PassReviewRequest,
ResumeRequest,
UnclaimRequest,
)
from roboco.services.gateway.choreographer import Choreographer
@@ -78,6 +79,16 @@ async def unclaim(
return env.as_dict()
@router.post("/resume")
async def resume(
body: ResumeRequest,
x_agent_id: _AgentIdHeader,
choreographer: _ChoreographerDep,
) -> dict:
env = await choreographer.resume(x_agent_id, body.task_id)
return env.as_dict()
@router.post("/i_am_idle")
async def i_am_idle(
_body: IAmIdleRequest,
+4
View File
@@ -36,6 +36,10 @@ class UnclaimRequest(BaseModel):
task_id: UUID
class ResumeRequest(BaseModel):
task_id: UUID
class IAmIdleRequest(BaseModel):
"""Empty request body."""
+6
View File
@@ -93,6 +93,11 @@ def unclaim(task_id: str) -> dict[str, Any]:
return _post(_role_path("unclaim"), {"task_id": task_id})
def resume(task_id: str) -> dict[str, Any]:
"""Resume a paused task. Transitions paused → in_progress for the assignee."""
return _post(_role_path("resume"), {"task_id": task_id})
def i_am_idle() -> dict[str, Any]:
"""Report no more work. Soft-blocks if you have unread A2A/mentions."""
return _post(_role_path("i_am_idle"), {})
@@ -219,6 +224,7 @@ _TOOLS: dict[str, Any] = {
"i_am_done": i_am_done,
"i_am_blocked": i_am_blocked,
"unclaim": unclaim,
"resume": resume,
"i_am_idle": i_am_idle,
# qa
"claim_review": claim_review,
+76 -4
View File
@@ -186,7 +186,7 @@ class Choreographer:
return Envelope.ok(
status=str(t.status),
task_id=str(t.id),
next=f"call i_will_work_on(task_id='{t.id}') to resume",
next=f"call resume(task_id='{t.id}') to continue paused work",
context_briefing=await self._briefing_for(agent_id, t.id),
)
return Envelope.ok(
@@ -820,6 +820,60 @@ class Choreographer:
context_briefing=briefing,
)
async def resume(self, agent_id: UUID, task_id: UUID) -> Envelope:
"""Resume a paused task this agent owns; transitions paused → in_progress.
Audit J33 ``i_am_idle`` auto-pauses owned in_progress tasks (so
the closure dispatcher can wake the agent when subtasks finish),
and the lifecycle table allows ``paused in_progress``, but no
verb exposed that transition to agents. ``i_will_work_on`` is
explicitly limited to needs_revision/pending/claimed; overloading
it would muddy state-machine intent. ``resume`` keeps it explicit.
State and authorization checks live here; the DB write itself is
in ``TaskService.resume_for_agent``.
"""
t = await self.task.get(task_id)
briefing = await self._briefing_for(agent_id, task_id)
if t is None:
return await self._emit_rejection(
Envelope.not_found(message=f"task {task_id} not found"),
agent_id=agent_id,
task_id=task_id,
verb="resume",
)
if t.assigned_to != agent_id:
return await self._emit_rejection(
Envelope.not_authorized(
message="not your claim",
remediate="only the current claimant can resume",
context_briefing=briefing,
),
agent_id=agent_id,
task_id=task_id,
verb="resume",
)
after = await self.task.resume_for_agent(task_id, agent_id)
if after is None:
return await self._emit_rejection(
Envelope.invalid_state(
message=f"cannot resume from status {t.status}",
remediate="only paused tasks can be resumed",
context_briefing=briefing,
),
agent_id=agent_id,
task_id=task_id,
verb="resume",
)
# Heartbeat — agent is back to active work after the resume.
await self._touch(task_id)
return Envelope.ok(
status=str(after.status),
task_id=str(task_id),
next="resumed; continue working — i_have_committed when ready",
context_briefing=briefing,
)
async def i_am_idle(self, agent_id: UUID) -> Envelope:
"""Report no more work. Soft-block if there are unread A2As or @mentions.
@@ -851,12 +905,23 @@ class Choreographer:
return await self._emit_rejection(
guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
)
await self._auto_pause_in_progress_tasks(agent_id)
paused_ids = await self._auto_pause_in_progress_tasks(agent_id)
await self.task.mark_agent_idle(agent_id)
if paused_ids:
# Tell the agent how to come back to these tasks. Without this,
# an agent respawned for a paused task has no signal that
# `resume(task_id)` is the way back in.
joined = ", ".join(f"resume(task_id='{tid}')" for tid in paused_ids)
next_msg = (
"container will shut down; on respawn, "
f"call {joined} to continue paused work"
)
else:
next_msg = "container will shut down"
return Envelope.ok(
status="idle",
task_id=None,
next="container will shut down",
next=next_msg,
context_briefing=briefing,
)
@@ -893,16 +958,23 @@ class Choreographer:
context_briefing=briefing,
)
async def _auto_pause_in_progress_tasks(self, agent_id: UUID) -> None:
async def _auto_pause_in_progress_tasks(self, agent_id: UUID) -> list[str]:
"""Pause every in_progress task assigned to this agent.
Restores the pre-Phase-4 auto-pause behavior: a PM that called
i_will_plan and is now idle leaves the parent at ``paused`` so the
closure dispatcher knows to respawn it when subtasks complete.
Returns the list of task IDs that were paused (as strings) so
``i_am_idle`` can tell the agent which ``resume(task_id)`` calls
await it on the next respawn. Empty list when nothing was active.
"""
in_progress = await self.task.list_in_progress_for_agent(agent_id)
paused_ids: list[str] = []
for t in in_progress:
await self.task.pause_for_agent(agent_id, t.id)
paused_ids.append(str(t.id))
return paused_ids
# --- Phase 2 (QA) verbs ---
+5
View File
@@ -30,6 +30,7 @@ _DEV_FLOW = (
"i_am_done",
"i_am_blocked",
"unclaim",
"resume",
"i_am_idle",
)
_DEV_DO = ("commit", "note", "say", "dm", "evidence")
@@ -40,6 +41,7 @@ _QA_FLOW = (
"pass",
"fail",
"unclaim",
"resume",
"i_am_idle",
)
_QA_DO = ("note", "say", "dm", "evidence")
@@ -49,6 +51,7 @@ _DOC_FLOW = (
"claim_doc_task",
"i_documented",
"unclaim",
"resume",
"i_am_idle",
)
_DOC_DO = ("commit", "note", "say", "dm", "evidence")
@@ -63,6 +66,7 @@ _CELL_PM_FLOW = (
"complete",
"escalate_up",
"unclaim",
"resume",
"i_am_idle",
)
_CELL_PM_DO = ("note", "say", "dm", "evidence")
@@ -77,6 +81,7 @@ _MAIN_PM_FLOW = (
"escalate_up",
"escalate_to_ceo",
"unclaim",
"resume",
"i_am_idle",
)
_MAIN_PM_DO = ("note", "say", "dm", "evidence")
+45
View File
@@ -1852,6 +1852,51 @@ class TaskService(BaseService):
await self.session.flush()
return task
async def resume_for_agent(self, task_id: UUID, agent_id: UUID) -> TaskTable | None:
"""Voluntary resume: transition paused task → in_progress for the assignee.
Distinct from ``resume`` (which takes only ``agent_role`` and is
called by closure-dispatcher / management code paths): this path
enforces that ``agent_id`` is the current claimant. Returns ``None``
and makes no write when:
- the task does not exist
- the requesting agent is not the current assignee
- the task status is not paused
- the lifecycle layer rejects the transition (defense-in-depth
against future ``VALID_TRANSITIONS``/``ROLE_RESTRICTED_TRANSITIONS``
changes; the choreographer pre-checks status today)
Routes through ``_validate_and_set_status`` (single point of truth)
to keep lifecycle enforcement consistent with claim/start/unclaim.
"""
task = await self.get(task_id)
if task is None or task.assigned_to != agent_id:
return None
if task.status != TaskStatus.PAUSED:
return None
# Look up the requesting agent's role so role-restricted-transition
# rules apply if/when paused→in_progress ever gains restrictions.
# Mirrors the pattern in `unclaim_for_agent` and `_finalize_claim`.
agent_result = await self.session.execute(
select(AgentTable).where(AgentTable.id == agent_id)
)
agent = agent_result.scalar_one_or_none()
agent_role = agent.role.value if agent and agent.role else None
# Route through the single point of truth for status transitions.
# _validate_and_set_status raises TaskLifecycleError on invalid
# transition or role; treat that as a clean rejection (return None)
# so the choreographer's "invalid_state" envelope still fires
# instead of a 500 leaking out.
try:
self._validate_and_set_status(task, TaskStatus.IN_PROGRESS, agent_role)
except TaskLifecycleError:
return None
await self.session.flush()
return task
async def block(
self,
task_id: UUID,
+159
View File
@@ -0,0 +1,159 @@
"""resume transitions a paused task back to in_progress for the assignee.
Audit J33 `paused -> in_progress` is a valid lifecycle transition but no
agent-callable verb implemented it. `i_will_work_on` only handles
needs_revision/pending/claimed; the closure dispatcher pauses owned
in_progress work on `i_am_idle` but nothing wakes it back up. `resume` fills
that gap. These tests pin the four behaviors:
- happy path: paused -> in_progress for the assignee
- not_found: unknown task id returns the not_found envelope
- not_authorized: only the current claimant can resume
- invalid_state: only paused tasks can be resumed
"""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4
import pytest
from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
def _make_deps(**overrides: Any) -> ChoreographerDeps:
"""Local dep-builder. Established pattern: per-test-file, not centralized."""
base: dict[str, Any] = {
"task": AsyncMock(),
"work_session": AsyncMock(),
"git": AsyncMock(),
"a2a": AsyncMock(),
"journal": AsyncMock(),
"audit": AsyncMock(),
"evidence_repo": AsyncMock(),
}
base.update(overrides)
repo = base["evidence_repo"]
for method in (
"list_unread_a2a",
"list_unread_mentions",
"list_pending_notifications",
"task_metadata_gaps",
"recent_team_activity",
"blockers_in_lane",
):
getattr(repo, method).return_value = []
return ChoreographerDeps(**base)
@pytest.mark.asyncio
async def test_resume_transitions_paused_to_in_progress() -> None:
aid = uuid4()
tid = uuid4()
t = MagicMock(id=tid, status="paused", assigned_to=aid)
task_svc = AsyncMock()
task_svc.get.return_value = t
task_svc.resume_for_agent.return_value = MagicMock(
id=tid, status="in_progress", assigned_to=aid
)
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.resume(aid, tid)
assert env.error is None
task_svc.resume_for_agent.assert_awaited_once_with(tid, aid)
assert env.status == "in_progress"
assert env.task_id == str(tid)
@pytest.mark.asyncio
async def test_resume_returns_not_found_for_unknown_task() -> None:
aid = uuid4()
tid = uuid4()
task_svc = AsyncMock()
task_svc.get.return_value = None
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.resume(aid, tid)
assert env.error == "not_found"
task_svc.resume_for_agent.assert_not_awaited()
@pytest.mark.asyncio
async def test_resume_rejects_when_not_claimant() -> None:
aid = uuid4()
other = uuid4()
tid = uuid4()
t = MagicMock(id=tid, status="paused", assigned_to=other)
task_svc = AsyncMock()
task_svc.get.return_value = t
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.resume(aid, tid)
assert env.error == "not_authorized"
task_svc.resume_for_agent.assert_not_awaited()
@pytest.mark.asyncio
async def test_resume_rejects_invalid_state() -> None:
aid = uuid4()
tid = uuid4()
# Task is owned but not paused (e.g. status drifted to in_progress between
# get and write). Service-level guard refuses by returning None.
t = MagicMock(id=tid, status="in_progress", assigned_to=aid)
task_svc = AsyncMock()
task_svc.get.return_value = t
task_svc.resume_for_agent.return_value = None
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.resume(aid, tid)
assert env.error == "invalid_state"
task_svc.resume_for_agent.assert_awaited_once_with(tid, aid)
@pytest.mark.asyncio
async def test_resume_rejection_writes_audit_row() -> None:
"""Every rejection envelope must call audit.log_event (Task 6 contract)."""
aid = uuid4()
tid = uuid4()
task_svc = AsyncMock()
task_svc.get.return_value = None
audit_svc = AsyncMock()
deps = _make_deps(task=task_svc, audit=audit_svc)
c = Choreographer(deps)
env = await c.resume(aid, tid)
assert env.error == "not_found"
audit_svc.log_event.assert_awaited_once()
kwargs = audit_svc.log_event.await_args.kwargs
assert kwargs["event_type"] == "gateway.rejected"
assert kwargs["details"]["verb"] == "resume"
@pytest.mark.asyncio
async def test_resume_success_writes_heartbeat() -> None:
"""Heartbeat fires on success — agent is back to active work."""
aid = uuid4()
tid = uuid4()
t = MagicMock(id=tid, status="paused", assigned_to=aid)
task_svc = AsyncMock()
task_svc.get.return_value = t
task_svc.resume_for_agent.return_value = MagicMock(
id=tid, status="in_progress", assigned_to=aid
)
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.resume(aid, tid)
assert env.error is None
task_svc.heartbeat.assert_awaited_once_with(tid)