feat(gateway): add unclaim verb

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

@@ -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. |

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. |

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. |

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. |

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. |

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,

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,

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,

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,

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,

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."""
 

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,

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.
 

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")

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,

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"