From cd9f999533c264b34a262d428cac2f3bf7a30a75 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 3 May 2026 07:05:16 +0200 Subject: [PATCH] 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). --- agents/prompts/roles/cell_pm.md | 1 + agents/prompts/roles/developer.md | 1 + agents/prompts/roles/documenter.md | 1 + agents/prompts/roles/main_pm.md | 1 + agents/prompts/roles/qa.md | 1 + roboco/api/routes/v2/flow_cell_pm.py | 11 ++ roboco/api/routes/v2/flow_dev.py | 11 ++ roboco/api/routes/v2/flow_doc.py | 11 ++ roboco/api/routes/v2/flow_main_pm.py | 11 ++ roboco/api/routes/v2/flow_qa.py | 11 ++ roboco/api/schemas/v2/flow.py | 4 + roboco/mcp/flow_server.py | 6 + roboco/services/gateway/choreographer.py | 80 +++++++++++- roboco/services/gateway/role_config.py | 5 + roboco/services/task.py | 45 +++++++ tests/unit/gateway/test_resume.py | 159 +++++++++++++++++++++++ 16 files changed, 355 insertions(+), 4 deletions(-) create mode 100644 tests/unit/gateway/test_resume.py diff --git a/agents/prompts/roles/cell_pm.md b/agents/prompts/roles/cell_pm.md index cf9f4f37..0426d7d6 100644 --- a/agents/prompts/roles/cell_pm.md +++ b/agents/prompts/roles/cell_pm.md @@ -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. | diff --git a/agents/prompts/roles/developer.md b/agents/prompts/roles/developer.md index 10e52da3..52392b69 100644 --- a/agents/prompts/roles/developer.md +++ b/agents/prompts/roles/developer.md @@ -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. | diff --git a/agents/prompts/roles/documenter.md b/agents/prompts/roles/documenter.md index f9be431d..579b1227 100644 --- a/agents/prompts/roles/documenter.md +++ b/agents/prompts/roles/documenter.md @@ -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. | diff --git a/agents/prompts/roles/main_pm.md b/agents/prompts/roles/main_pm.md index dab696cf..0ff8350e 100644 --- a/agents/prompts/roles/main_pm.md +++ b/agents/prompts/roles/main_pm.md @@ -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. | diff --git a/agents/prompts/roles/qa.md b/agents/prompts/roles/qa.md index 763ade2c..4b926124 100644 --- a/agents/prompts/roles/qa.md +++ b/agents/prompts/roles/qa.md @@ -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. | diff --git a/roboco/api/routes/v2/flow_cell_pm.py b/roboco/api/routes/v2/flow_cell_pm.py index 815f36c3..059e2887 100644 --- a/roboco/api/routes/v2/flow_cell_pm.py +++ b/roboco/api/routes/v2/flow_cell_pm.py @@ -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, diff --git a/roboco/api/routes/v2/flow_dev.py b/roboco/api/routes/v2/flow_dev.py index 6da69b06..4c501c3f 100644 --- a/roboco/api/routes/v2/flow_dev.py +++ b/roboco/api/routes/v2/flow_dev.py @@ -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, diff --git a/roboco/api/routes/v2/flow_doc.py b/roboco/api/routes/v2/flow_doc.py index 5b67fc8c..8bec721f 100644 --- a/roboco/api/routes/v2/flow_doc.py +++ b/roboco/api/routes/v2/flow_doc.py @@ -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, diff --git a/roboco/api/routes/v2/flow_main_pm.py b/roboco/api/routes/v2/flow_main_pm.py index b3f95e71..4e2f61c2 100644 --- a/roboco/api/routes/v2/flow_main_pm.py +++ b/roboco/api/routes/v2/flow_main_pm.py @@ -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, diff --git a/roboco/api/routes/v2/flow_qa.py b/roboco/api/routes/v2/flow_qa.py index 8e903a93..d2e3124c 100644 --- a/roboco/api/routes/v2/flow_qa.py +++ b/roboco/api/routes/v2/flow_qa.py @@ -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, diff --git a/roboco/api/schemas/v2/flow.py b/roboco/api/schemas/v2/flow.py index 958859a3..12645743 100644 --- a/roboco/api/schemas/v2/flow.py +++ b/roboco/api/schemas/v2/flow.py @@ -36,6 +36,10 @@ class UnclaimRequest(BaseModel): task_id: UUID +class ResumeRequest(BaseModel): + task_id: UUID + + class IAmIdleRequest(BaseModel): """Empty request body.""" diff --git a/roboco/mcp/flow_server.py b/roboco/mcp/flow_server.py index 5aef9130..bf087837 100644 --- a/roboco/mcp/flow_server.py +++ b/roboco/mcp/flow_server.py @@ -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, diff --git a/roboco/services/gateway/choreographer.py b/roboco/services/gateway/choreographer.py index eb07bee9..4de93b1c 100644 --- a/roboco/services/gateway/choreographer.py +++ b/roboco/services/gateway/choreographer.py @@ -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 --- diff --git a/roboco/services/gateway/role_config.py b/roboco/services/gateway/role_config.py index f51fe9c5..597ff4e1 100644 --- a/roboco/services/gateway/role_config.py +++ b/roboco/services/gateway/role_config.py @@ -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") diff --git a/roboco/services/task.py b/roboco/services/task.py index 9f5f4a38..e5b276ac 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -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, diff --git a/tests/unit/gateway/test_resume.py b/tests/unit/gateway/test_resume.py new file mode 100644 index 00000000..da6ffc5e --- /dev/null +++ b/tests/unit/gateway/test_resume.py @@ -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)