fix(gateway): improve unclaim/resume rejection when task was reassigned

Investigation finding for Task 6 of the gateway introspection plan:
the 2026-05-08 trace's "not your claim" rejection at 02:51:22 / 02:52:48
was NOT caused by a UUID-comparator bug. AGENT_UUIDS in
roboco/seeds/initial_data.py are static so identity is stable across
restarts, and SQLAlchemy + Pydantic both round-trip UUIDs cleanly
(pinned by two new regression tests in test_task_service_misc.py).

The actual cause: the task was REASSIGNED out from under main-pm by
an upstream verb between when main-pm last touched it and when it
tried to unclaim/resume. Common triggers:
  - cell_pm_complete propagates up via _maybe_advance_parent_to_pm_review,
    which reassigns the parent to the cell PM for the team
  - main_pm_complete clears assigned_to to None (CEO acts via UI)
  - unblock with restore=True flips assigned_to back to pre_block_state

Pre-fix the rejection said only "not your claim" — agents can't tell
whether they hit a transient race or whether the task was legitimately
moved on. Fix: surface the current_owner UUID and hint that an
upstream verb did this, telling the agent to call give_me_work() to
find its current work.

roboco/services/gateway/choreographer/_impl.py

@@ -881,10 +881,26 @@ class Choreographer:
         agent = await self.task.agent_for(agent_id)
         role = str(agent.role) if agent is not None else "developer"
         if t.assigned_to != agent_id:
+            # The task was reassigned out from under this agent — most
+            # commonly by an upstream verb that legitimately changed
+            # ownership (cell_pm_complete propagating to the parent,
+            # main_pm_complete clearing assigned_to to None when
+            # escalating to CEO, or a PM unblocking with restore=True).
+            # The agent's local state is stale; tell it concretely.
+            current_owner = (
+                str(t.assigned_to) if t.assigned_to is not None else "<unassigned>"
+            )
             return await self._emit_rejection(
                 Envelope.not_authorized(
-                    message="not your claim",
-                    remediate="only the current claimant can unclaim",
+                    message=(
+                        f"task {task_id} is no longer assigned to you "
+                        f"(current owner: {current_owner})"
+                    ),
+                    remediate=(
+                        "the task was reassigned by an upstream verb "
+                        "(cell_pm_complete / main_pm_complete / unblock). "
+                        "call give_me_work() to find your current work."
+                    ),
                     context_briefing=briefing,
                 ).with_introspection(task=t, role=role),
                 agent_id=agent_id,
@@ -938,10 +954,23 @@ class Choreographer:
         agent = await self.task.agent_for(agent_id)
         role = str(agent.role) if agent is not None else "developer"
         if t.assigned_to != agent_id:
+            # See unclaim's matching branch for the rationale: the task
+            # was reassigned by an upstream verb. Surface the actual
+            # current owner so the agent can stop looping on a stale
+            # task_id and call give_me_work() instead.
+            current_owner = (
+                str(t.assigned_to) if t.assigned_to is not None else "<unassigned>"
+            )
             return await self._emit_rejection(
                 Envelope.not_authorized(
-                    message="not your claim",
-                    remediate="only the current claimant can resume",
+                    message=(
+                        f"task {task_id} is no longer assigned to you "
+                        f"(current owner: {current_owner})"
+                    ),
+                    remediate=(
+                        "the task was reassigned by an upstream verb. "
+                        "call give_me_work() to find your current work."
+                    ),
                     context_briefing=briefing,
                 ).with_introspection(task=t, role=role),
                 agent_id=agent_id,

tests/integration/test_task_service_misc.py

@@ -27,7 +27,7 @@ from __future__ import annotations
 import asyncio
 from typing import TYPE_CHECKING, Any
 from unittest.mock import AsyncMock, MagicMock
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 import pytest_asyncio
@@ -1211,3 +1211,48 @@ async def test_resolve_pm_for_review_returns_none_when_chain_broken(
     monkeypatch.setattr(svc, "get", _stub_get)
     out = await svc._resolve_pm_for_review(child)
     assert out is None
+
+
+@pytest.mark.asyncio
+async def test_unclaim_for_agent_works_with_uuid_round_trip(
+    task_setup: dict,
+    db_session: AsyncSession,
+) -> None:
+    """Regression for 2026-05-08: main-pm got 'not your claim' on its OWN
+    claim. Pin the UUID-comparator behavior: same agent_id in (whether
+    fresh UUID or string-coerced UUID round-trip), unclaim succeeds.
+    """
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.CLAIMED
+    task.assigned_to = task_setup["agent_id"]
+    await db_session.flush()
+
+    # Round-trip the UUID through string → UUID to mirror what an HTTP
+    # request body / header would do via Pydantic. The comparator must
+    # treat both the freshly-constructed and the round-tripped UUID as
+    # equal (which Python's UUID class does — pinning so a future move
+    # to a custom comparator can't silently break it).
+    same_uuid_via_str = UUID(str(task_setup["agent_id"]))
+    out = await svc.unclaim_for_agent(task.id, agent_id=same_uuid_via_str)
+    assert out is not None, "round-tripped UUID rejected — comparator broken"
+    assert out.assigned_to is None
+    assert out.status == TaskStatus.PENDING
+
+
+@pytest.mark.asyncio
+async def test_resume_for_agent_works_with_uuid_round_trip(
+    task_setup: dict,
+    db_session: AsyncSession,
+) -> None:
+    """Mirror of the unclaim regression for resume_for_agent."""
+    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()
+
+    same_uuid_via_str = UUID(str(task_setup["agent_id"]))
+    out = await svc.resume_for_agent(task.id, agent_id=same_uuid_via_str)
+    assert out is not None, "round-tripped UUID rejected — comparator broken"
+    assert out.status == TaskStatus.IN_PROGRESS