fix(gateway): unclaim releases a pending-assigned task — escape trap (#176)

An agent assigned a `pending` task it never claimed was structurally
trapped: from pending-assigned, unclaim returned None ("cannot unclaim
from status pending"), i_am_idle rejected ("assigned but never claimed"),
i_am_blocked rejected ("block requires in_progress"). Any persistent
claim-time rejection (a gate the agent cannot satisfy, a transient
validation error) therefore looped the agent until budget-reap AND left
the task orphaned (pending, assigned, no progress). Observed in smoke-16
and smoke-17.

unclaim_for_agent now releases a pending task assigned to the caller:
no status change (already pending → no lifecycle transition, no
WorkSession to abandon since it was never claimed), just clear
assigned_to/active_claimant_id so the dispatcher can reassign. The
choreographer spec gate already permits unclaim from pending (composes=()
— role-only), so the service branch is the whole fix. Updated the now-
stale unclaim remediate string; rewrote the test that encoded the buggy
trap and added a paused-status negative case.

roboco/services/gateway/choreographer/_impl.py

@@ -2222,7 +2222,10 @@ class Choreographer:
             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",
+                    remediate=(
+                        "only a task assigned to you in pending / claimed / "
+                        "in_progress can be unclaimed"
+                    ),
                     context_briefing=briefing,
                 ).with_introspection(task=t, role=role_str),
                 agent_id=agent_id,

roboco/services/task.py

@@ -1997,6 +1997,20 @@ class TaskService(BaseService):
         task = await self.get(task_id)
         if task is None or task.assigned_to != agent_id:
             return None
+        # #176: an agent assigned a `pending` task it never claimed (any
+        # persistent claim-time rejection — e.g. a gate the agent cannot
+        # satisfy) is otherwise trapped: unclaim/i_am_idle/i_am_blocked all
+        # reject from pending-assigned, so it loops until budget-reap and
+        # the task is left orphaned (pending, assigned, no progress).
+        # Releasing the assignment is a no-status-change escape (the row is
+        # already pending; no transition, so no lifecycle validation and no
+        # WorkSession to abandon — it was never claimed). The task returns
+        # to the pool for the dispatcher to reassign.
+        if task.status == TaskStatus.PENDING:
+            task.assigned_to = cast("Any", None)
+            task.active_claimant_id = cast("Any", None)
+            await self.session.flush()
+            return task
         if task.status not in (TaskStatus.CLAIMED, TaskStatus.IN_PROGRESS):
             return None
 

tests/integration/test_task_service_transitions.py

@@ -247,14 +247,36 @@ async def test_unclaim_for_agent_returns_none_when_wrong_assignee(
 
 
 @pytest.mark.asyncio
-async def test_unclaim_for_agent_returns_none_when_not_claimed(
+async def test_unclaim_for_agent_releases_pending_assignment(
     task_setup: dict, db_session: AsyncSession
 ) -> None:
+    """#176: an agent assigned a pending task it never claimed must be
+    able to unclaim it (escape the pending-assigned trap). The row stays
+    pending; only the assignment is released so the dispatcher can
+    reassign it instead of orphaning the task + looping the agent."""
     svc = task_setup["svc"]
     task = await svc.create(_req(task_setup))
     task.assigned_to = task_setup["agent_id"]
     await db_session.flush()
-    # status is PENDING, not CLAIMED/IN_PROGRESS
+    # status is PENDING, never claimed — the smoke-17 trap scenario.
+    out = await svc.unclaim_for_agent(task.id, agent_id=task_setup["agent_id"])
+    assert out is not None
+    assert out.status == TaskStatus.PENDING
+    assert out.assigned_to is None
+    assert out.active_claimant_id is None
+
+
+@pytest.mark.asyncio
+async def test_unclaim_for_agent_returns_none_when_paused(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """A non-pending/claimed/in_progress status (e.g. paused — which has
+    `resume`, not `unclaim`) is still not unclaimable."""
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.PAUSED
+    task.assigned_to = task_setup["agent_id"]
+    await db_session.flush()
     assert await svc.unclaim_for_agent(task.id, agent_id=task_setup["agent_id"]) is None