From 1f7c9adaba17ecddc3b605b969ea627ac1e3b1e1 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 3 May 2026 06:38:27 +0200 Subject: [PATCH] feat(gateway): add unclaim verb MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Choreographer remediate strings already pointed agents at unclaim; the verb didn't exist. Now it does — claimed/in_progress -> pending, clears assigned_to, available to dev/qa/doc/cell_pm/main_pm. --- 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 | 49 ++++++++ roboco/services/gateway/role_config.py | 5 + roboco/services/task.py | 29 +++++ tests/unit/gateway/test_unclaim.py | 137 +++++++++++++++++++++++ 16 files changed, 290 insertions(+) create mode 100644 tests/unit/gateway/test_unclaim.py diff --git a/agents/prompts/roles/cell_pm.md b/agents/prompts/roles/cell_pm.md index 4b653932..cf9f4f37 100644 --- a/agents/prompts/roles/cell_pm.md +++ b/agents/prompts/roles/cell_pm.md @@ -27,6 +27,7 @@ You merge what your developers submit (leaf PRs into your cell branch via `compl | `complete(task_id, notes)` | Review a SUBTASK in `awaiting_pm_review`; auto-merges the leaf PR into your cell branch. | All descendants of the subtask terminal; PR open and mergeable. | | `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. | | `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 1a2b641a..10e52da3 100644 --- a/agents/prompts/roles/developer.md +++ b/agents/prompts/roles/developer.md @@ -24,6 +24,7 @@ You write code; you do not coordinate. If you find yourself thinking "let me als | `submit_for_qa(task_id)` | Push your branch and open a PR. Run after your last commit, before `i_am_done`. | Task assigned to you; at least one commit; no PR yet. | | `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. | | `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 f34795be..f9be431d 100644 --- a/agents/prompts/roles/documenter.md +++ b/agents/prompts/roles/documenter.md @@ -21,6 +21,7 @@ You do NOT re-implement the developer's work. You do NOT review or critique the | `claim_doc_task(task_id)` | Claims the doc task; returns PR data inline. | Task in `awaiting_documentation`; you are not the original developer. | | `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. | | `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 e4ebbb86..dab696cf 100644 --- a/agents/prompts/roles/main_pm.md +++ b/agents/prompts/roles/main_pm.md @@ -27,6 +27,7 @@ You merge what your Cell PMs submit (cell PRs into your root branch via `complet | `complete(task_id, notes)` | For a cell-PM task in `awaiting_pm_review`: merges the cell PR into your root branch. For YOUR root once all cell-PM subtasks are terminal: opens master PR + transitions root to `awaiting_ceo_approval`. | All descendants terminal; journal `decision` recorded. | | `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. | | `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 0c8ce4d4..763ade2c 100644 --- a/agents/prompts/roles/qa.md +++ b/agents/prompts/roles/qa.md @@ -20,6 +20,7 @@ A pass without evidence is a betrayal of your role: the entire downstream chain | `claim_review(task_id)` | Claims the QA task; returns PR data inline. | Task in `awaiting_qa`; you are not the original developer. | | `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. | | `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 8757afd5..815f36c3 100644 --- a/roboco/api/routes/v2/flow_cell_pm.py +++ b/roboco/api/routes/v2/flow_cell_pm.py @@ -17,6 +17,7 @@ from roboco.api.schemas.v2.flow import ( SubmitUpRequest, TriageRequest, UnblockRequest, + UnclaimRequest, ) from roboco.services.gateway.choreographer import Choreographer, DelegateInputs @@ -120,6 +121,16 @@ async def escalate_up( return env.as_dict() +@router.post("/unclaim") +async def unclaim( + body: UnclaimRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.unclaim(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 7ab49964..6da69b06 100644 --- a/roboco/api/routes/v2/flow_dev.py +++ b/roboco/api/routes/v2/flow_dev.py @@ -15,6 +15,7 @@ from roboco.api.schemas.v2.flow import ( IHaveCommittedRequest, IWillWorkOnRequest, SubmitForQaRequest, + UnclaimRequest, ) from roboco.services.gateway.choreographer import Choreographer @@ -89,6 +90,16 @@ async def i_am_blocked( return env.as_dict() +@router.post("/unclaim") +async def unclaim( + body: UnclaimRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.unclaim(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 cf45c642..5b67fc8c 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, + UnclaimRequest, ) from roboco.services.gateway.choreographer import Choreographer @@ -58,6 +59,16 @@ async def i_documented( return env.as_dict() +@router.post("/unclaim") +async def unclaim( + body: UnclaimRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.unclaim(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 eaf15003..b3f95e71 100644 --- a/roboco/api/routes/v2/flow_main_pm.py +++ b/roboco/api/routes/v2/flow_main_pm.py @@ -17,6 +17,7 @@ from roboco.api.schemas.v2.flow import ( IWillPlanRequest, TriageRequest, UnblockRequest, + UnclaimRequest, ) from roboco.services.gateway.choreographer import Choreographer, DelegateInputs @@ -120,6 +121,16 @@ async def unblock( return env.as_dict() +@router.post("/unclaim") +async def unclaim( + body: UnclaimRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.unclaim(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 01b56562..8e903a93 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, + UnclaimRequest, ) from roboco.services.gateway.choreographer import Choreographer @@ -67,6 +68,16 @@ async def qa_fail( return env.as_dict() +@router.post("/unclaim") +async def unclaim( + body: UnclaimRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.unclaim(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 df819c70..958859a3 100644 --- a/roboco/api/schemas/v2/flow.py +++ b/roboco/api/schemas/v2/flow.py @@ -32,6 +32,10 @@ class IAmBlockedRequest(BaseModel): reason: str = Field(..., min_length=1) +class UnclaimRequest(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 973dc6cc..5aef9130 100644 --- a/roboco/mcp/flow_server.py +++ b/roboco/mcp/flow_server.py @@ -88,6 +88,11 @@ def i_am_blocked(task_id: str, reason: str) -> dict[str, Any]: return _post(_role_path("i_am_blocked"), {"task_id": task_id, "reason": reason}) +def unclaim(task_id: str) -> dict[str, Any]: + """Release this claim back to pending. Branch survives; task is unassigned.""" + return _post(_role_path("unclaim"), {"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"), {}) @@ -213,6 +218,7 @@ _TOOLS: dict[str, Any] = { "submit_for_qa": submit_for_qa, "i_am_done": i_am_done, "i_am_blocked": i_am_blocked, + "unclaim": unclaim, "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 d1df3e97..eb07bee9 100644 --- a/roboco/services/gateway/choreographer.py +++ b/roboco/services/gateway/choreographer.py @@ -771,6 +771,55 @@ class Choreographer: context_briefing=await self._briefing_for(agent_id, task_id), ) + async def unclaim(self, agent_id: UUID, task_id: UUID) -> Envelope: + """Voluntarily release a claimed/in_progress task back to pending. + + Audit J33 — ``_pending_assignment_guard`` already remediates with + "or unclaim it first," but the verb didn't exist. This makes that + promise true. The work-in-progress branch survives; only the claim + is released so another agent (or the same one, fresh) can pick it + up. State and authorization checks live here; the DB write itself + is in ``TaskService.unclaim_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="unclaim", + ) + if t.assigned_to != agent_id: + return await self._emit_rejection( + Envelope.not_authorized( + message="not your claim", + remediate="only the current claimant can unclaim", + context_briefing=briefing, + ), + agent_id=agent_id, + task_id=task_id, + verb="unclaim", + ) + after = await self.task.unclaim_for_agent(task_id, agent_id) + if after is None: + return await self._emit_rejection( + Envelope.invalid_state( + message=f"cannot unclaim from status {t.status}", + remediate="only claimed/in_progress tasks can be unclaimed", + context_briefing=briefing, + ), + agent_id=agent_id, + task_id=task_id, + verb="unclaim", + ) + return Envelope.ok( + status=str(after.status), + task_id=str(task_id), + next="task returned to pending; another agent (or you, fresh) can claim", + 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. diff --git a/roboco/services/gateway/role_config.py b/roboco/services/gateway/role_config.py index de4f8366..f51fe9c5 100644 --- a/roboco/services/gateway/role_config.py +++ b/roboco/services/gateway/role_config.py @@ -29,6 +29,7 @@ _DEV_FLOW = ( "submit_for_qa", "i_am_done", "i_am_blocked", + "unclaim", "i_am_idle", ) _DEV_DO = ("commit", "note", "say", "dm", "evidence") @@ -38,6 +39,7 @@ _QA_FLOW = ( "claim_review", "pass", "fail", + "unclaim", "i_am_idle", ) _QA_DO = ("note", "say", "dm", "evidence") @@ -46,6 +48,7 @@ _DOC_FLOW = ( "give_me_work", "claim_doc_task", "i_documented", + "unclaim", "i_am_idle", ) _DOC_DO = ("commit", "note", "say", "dm", "evidence") @@ -59,6 +62,7 @@ _CELL_PM_FLOW = ( "unblock", "complete", "escalate_up", + "unclaim", "i_am_idle", ) _CELL_PM_DO = ("note", "say", "dm", "evidence") @@ -72,6 +76,7 @@ _MAIN_PM_FLOW = ( "complete", "escalate_up", "escalate_to_ceo", + "unclaim", "i_am_idle", ) _MAIN_PM_DO = ("note", "say", "dm", "evidence") diff --git a/roboco/services/task.py b/roboco/services/task.py index 910d0348..f9f078b6 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -1798,6 +1798,35 @@ class TaskService(BaseService): ) await self.session.flush() + async def unclaim_for_agent( + self, task_id: UUID, agent_id: UUID + ) -> TaskTable | None: + """Voluntary unclaim by the current claimant. + + Distinct from ``unclaim_for_reaper`` (which the orchestrator's + stale-claim sweeper calls when the holder is provably dead): this + path is the agent itself releasing the lock. Returns ``None`` and + makes no write when: + + - the task does not exist + - the requesting agent is not the current claimant + - the task status is not claimed/in_progress + + On success, clears ``assigned_to`` and transitions the row back to + ``pending`` so another agent (or the same one, fresh) can pick it + up. The work-in-progress branch is preserved — only the claim is + released. + """ + task = await self.get(task_id) + if task is None or task.assigned_to != agent_id: + return None + if task.status not in (TaskStatus.CLAIMED, TaskStatus.IN_PROGRESS): + return None + task.assigned_to = cast("Any", None) + task.status = TaskStatus.PENDING + await self.session.flush() + return task + async def block( self, task_id: UUID, diff --git a/tests/unit/gateway/test_unclaim.py b/tests/unit/gateway/test_unclaim.py new file mode 100644 index 00000000..6f41a31d --- /dev/null +++ b/tests/unit/gateway/test_unclaim.py @@ -0,0 +1,137 @@ +"""unclaim returns claimed/in_progress task to pending and clears assigned_to. + +Audit J33 — `Choreographer._pending_assignment_guard` remediate string says +"or unclaim it first" pointing to a verb that did not exist. The verb now +exists. These tests pin the four behaviors: + +- happy path: claimed -> pending, assigned_to cleared +- not_found: unknown task id returns the not_found envelope +- not_authorized: only the current claimant can unclaim +- invalid_state: only claimed/in_progress tasks can be unclaimed +""" + +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_unclaim_returns_task_to_pending() -> None: + aid = uuid4() + tid = uuid4() + t = MagicMock(id=tid, status="claimed", assigned_to=aid) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.unclaim_for_agent.return_value = MagicMock( + id=tid, status="pending", assigned_to=None + ) + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.unclaim(aid, tid) + + assert env.error is None + task_svc.unclaim_for_agent.assert_awaited_once_with(tid, aid) + assert env.status == "pending" + assert env.task_id == str(tid) + + +@pytest.mark.asyncio +async def test_unclaim_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.unclaim(aid, tid) + + assert env.error == "not_found" + task_svc.unclaim_for_agent.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_unclaim_rejects_when_not_claimant() -> None: + aid = uuid4() + other = uuid4() + tid = uuid4() + t = MagicMock(id=tid, status="claimed", assigned_to=other) + task_svc = AsyncMock() + task_svc.get.return_value = t + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.unclaim(aid, tid) + + assert env.error == "not_authorized" + task_svc.unclaim_for_agent.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_unclaim_rejects_invalid_state() -> None: + aid = uuid4() + tid = uuid4() + # Status is claimed (assigned_to matches), but the service-level guard + # refuses (e.g. status drifted to verifying between get and write). + t = MagicMock(id=tid, status="verifying", assigned_to=aid) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.unclaim_for_agent.return_value = None + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.unclaim(aid, tid) + + assert env.error == "invalid_state" + task_svc.unclaim_for_agent.assert_awaited_once_with(tid, aid) + + +@pytest.mark.asyncio +async def test_unclaim_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.unclaim(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"] == "unclaim"