fix(store): enforce runnable-phase claim release across all write paths

Work items re-entering a runnable phase (ready/ready_for_rework) now have
ownership released unconditionally at the store layer: claim CAS no longer
consults metadata mirror keys (columns are the only ownership truth),
update_delegation_work_item blanks columns+mirror on any runnable-phase
write, review REJECT resolution releases ownership when the target phase is
runnable, and the startup sweep also covers runnable-phase residue. Closes
the 0011 rework livelock (stale four-field claim CAS vs. un-released claim).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

opc/database/store.py

@@ -74,8 +74,10 @@ from opc.layer2_organization.phase import (
     IN_PROGRESS_PHASES,
     IN_REVIEW_PHASES,
     InvalidPhaseTransition,
+    RUNNABLE_PHASES,
     TODO_PHASES,
     coerce_phase,
+    is_runnable,
     is_stale_claim_releasable,
     is_terminal,
     kanban_column,
@@ -4579,7 +4581,11 @@ class OPCStore:
                 phase = coerce_phase(phase_str)
             except (TypeError, ValueError):
                 continue
-            if not is_stale_claim_releasable(phase):
+            # Runnable phases are covered too: the phase-write invariant
+            # keeps READY / READY_FOR_REWORK rows unowned, so any claim
+            # found on one is corrupt legacy state that would starve the
+            # claim CAS forever.
+            if not (is_stale_claim_releasable(phase) or is_runnable(phase)):
                 continue
             metadata = _json_loads(metadata_json, {})
             metadata["claimed_by_role_session_id"] = ""
@@ -5465,6 +5471,18 @@ class OPCStore:
             if metadata_updates:
                 metadata.update(dict(metadata_updates))
             item.metadata = metadata
+        if phase is not None and item.phase in RUNNABLE_PHASES:
+            # Invariant: a fresh-runnable card is unowned. The claim CAS
+            # refuses cards with any leftover claim, so entering READY /
+            # READY_FOR_REWORK must release ownership in the same write —
+            # otherwise the card is permanently unclaimable and the
+            # dispatcher livelocks (project-0011 wedge).
+            item.claimed_by_role_runtime_session_id = ""
+            item.claimed_by_seat_id = ""
+            metadata = dict(item.metadata or {})
+            metadata["claimed_by_role_session_id"] = ""
+            metadata["claimed_task_id"] = ""
+            item.metadata = metadata
         item.updated_at = datetime.now()
         await self.save_delegation_work_item(item)
         return item
@@ -5485,6 +5503,12 @@ class OPCStore:
         shutdown transition.  Keeping the phase, claim, queue, and durable
         hold predicates in the same UPDATE prevents that stale snapshot from
         resurrecting a suspended WorkItem.
+
+        Ownership truth is the two claim columns. The metadata mirror keys
+        (``claimed_by_role_session_id`` / ``claimed_task_id``) are written for
+        observability but must never gate the claim: a mirror key that a
+        release path forgot to blank would make a runnable card permanently
+        unclaimable (the project-0011 dispatcher livelock).
         """
 
         phase = coerce_phase(expected_phase)
@@ -5528,8 +5552,6 @@ class OPCStore:
                  AND phase = ?
                  AND COALESCE(claimed_by_role_runtime_session_id, '') = ''
                  AND COALESCE(claimed_by_seat_id, '') = ''
-                 AND COALESCE(json_extract(metadata, '$.claimed_by_role_session_id'), '') = ''
-                 AND COALESCE(json_extract(metadata, '$.claimed_task_id'), '') = ''
                  AND COALESCE(json_extract(metadata, '$.dispatch_hold'), '') = ''
                  AND COALESCE(json_extract(metadata, '$.queued_behind_session'), '') = ''
                  AND COALESCE(
@@ -5602,12 +5624,22 @@ class OPCStore:
         expected_source = str(source_report_work_item_id or "").strip()
         db = self._require_db()
 
+        # A rework verdict sends the card back to the dispatch queue; the
+        # claim CAS refuses owned cards, so ownership must be released in
+        # the same UPDATE that writes the runnable phase (project-0011
+        # wedge: REJECT kept the worker's claim and the card became
+        # permanently unclaimable).
+        release_ownership = target in RUNNABLE_PHASES
+
         for _attempt in range(3):
             item = await self.get_delegation_work_item(work_item_id)
             if item is None or item.phase != Phase.AWAITING_MANAGER_REVIEW:
                 return None
             metadata = dict(item.metadata or {})
             metadata.update(dict(metadata_updates or {}))
+            if release_ownership:
+                metadata["claimed_by_role_session_id"] = ""
+                metadata["claimed_task_id"] = ""
             if (
                 self._metadata_has_work_item_projection_identity(metadata)
                 or str(item.projection_id or "").strip()
@@ -5622,9 +5654,18 @@ class OPCStore:
                 )
             previous_updated_at = item.updated_at.isoformat()
             updated_at = datetime.now()
+            claimed_session = (
+                "" if release_ownership
+                else str(item.claimed_by_role_runtime_session_id or "")
+            )
+            claimed_seat = (
+                "" if release_ownership else str(item.claimed_by_seat_id or "")
+            )
             cursor = await db.execute(
                 """UPDATE delegation_work_items
-                   SET phase = ?, blocked_reason = ?, metadata = ?, updated_at = ?
+                   SET phase = ?, blocked_reason = ?, metadata = ?, updated_at = ?,
+                       claimed_by_role_runtime_session_id = ?,
+                       claimed_by_seat_id = ?
                    WHERE work_item_id = ?
                      AND phase = ?
                      AND updated_at = ?
@@ -5651,6 +5692,8 @@ class OPCStore:
                     str(blocked_reason or ""),
                     _json_dumps(metadata),
                     updated_at.isoformat(),
+                    claimed_session,
+                    claimed_seat,
                     work_item_id,
                     Phase.AWAITING_MANAGER_REVIEW.value,
                     previous_updated_at,

opc/layer2_organization/work_item_transition.py

@@ -179,12 +179,11 @@ async def transition_work_item(
     if release_claim:
         # Fold claim release into the same write as the phase change: the
         # legacy two-call sequence could commit the phase and then fail the
-        # release, stranding a dead claim. Terminal phases additionally drop
+        # release, stranding a dead claim. The metadata mirror keys are
-        # the metadata claim mirror keys so the row can never satisfy a
+        # blanked alongside the columns so the mirror never outlives the
-        # future claim-CAS predicate by accident.
+        # ownership it mirrors.
         kwargs["claimed_by_role_runtime_session_id"] = ""
         kwargs["claimed_by_seat_id"] = ""
-        if phase in DONE_PHASES:
         merged.setdefault("claimed_by_role_session_id", "")
         merged.setdefault("claimed_task_id", "")
     try:
@@ -1147,22 +1146,20 @@ async def refresh_dependents_for_run(
                 elif work_item.phase == Phase.RUNNING:
                     target_phase = Phase.WAITING_FOR_CHILDREN
                 metadata_updates["waiting_on_work_item_ids"] = dependency_ids
-            # Clear the parent claim whenever the parent truly leaves
+            # Waking a parent out of WAITING_FOR_CHILDREN must orphan it so
-            # WAITING_FOR_CHILDREN toward a non-terminal phase. The old
+            # the dispatcher can re-pick it. For READY targets the store's
-            # condition ("only when all children approved AND target is
+            # phase-write invariant releases ownership; only the direct
-            # RUNNING") left a gap: when a child went READY_FOR_REWORK,
+            # →RUNNING wake still needs the explicit clear (columns and
-            # the refresh now fires (per _DEPENDENT_REFRESH_TARGETS) but
+            # mirror in the same write). Terminal targets keep the claim as
-            # the parent stayed in WAITING_FOR_CHILDREN with a stale claim,
+            # a historical audit record of "last executor".
-            # so the dispatcher couldn't re-pick it even though the child
-            # was back on the worker's queue.
-            # We exclude DONE_PHASES because for terminal parents the
-            # claim is a historical audit record of "last executor".
             clear_claim_on_wake = (
                 work_item.phase == Phase.WAITING_FOR_CHILDREN
-                and target_phase != work_item.phase
+                and target_phase == Phase.RUNNING
-                and target_phase not in DONE_PHASES
             )
-            if target_phase != work_item.phase or metadata_updates or clear_claim_on_wake:
+            if clear_claim_on_wake:
+                metadata_updates["claimed_by_role_session_id"] = ""
+                metadata_updates["claimed_task_id"] = ""
+            if target_phase != work_item.phase or metadata_updates:
                 try:
                     await store.update_delegation_work_item(
                         work_item.work_item_id,

tests/test_claim_release_invariant.py

@@ -0,0 +1,192 @@
+"""Regression tests for the claim-release invariant (project-0011 livelock).
+
+The claim CAS refuses any card whose claim columns are non-empty, so every
+write that puts a card back into a fresh-runnable phase (READY /
+READY_FOR_REWORK) must release ownership in the same write. Before this
+invariant existed, two paths leaked claims and wedged whole runs:
+
+- review REJECT → READY_FOR_REWORK kept the worker's claim columns and
+  metadata mirror (``apply_delegation_review_resolution``);
+- the synthesis wake WAITING_FOR_CHILDREN → READY cleared the columns but
+  left the metadata mirror, which the CAS also used to gate claims.
+
+The dispatcher retried the claim every tick and lost every time — a silent
+livelock that survived restarts because the startup sweep skipped runnable
+phases.
+"""
+from __future__ import annotations
+
+import asyncio
+from functools import wraps
+from pathlib import Path
+
+from opc.core.models import DelegationWorkItem, Phase
+from opc.database.store import OPCStore
+
+
+def _async_test(func):
+    @wraps(func)
+    def runner(*args, **kwargs):
+        return asyncio.run(func(*args, **kwargs))
+
+    return runner
+
+
+def _work_item(
+    work_item_id: str,
+    *,
+    phase: Phase,
+    metadata: dict | None = None,
+    claimed_session: str = "",
+    claimed_seat: str = "",
+) -> DelegationWorkItem:
+    return DelegationWorkItem(
+        work_item_id=work_item_id,
+        run_id="claim-invariant-run",
+        cell_id="team::executor",
+        role_id="executor",
+        seat_id="seat::executor",
+        title=work_item_id,
+        kind="execute",
+        projection_id=work_item_id,
+        phase=phase,
+        claimed_by_role_runtime_session_id=claimed_session,
+        claimed_by_seat_id=claimed_seat,
+        metadata=dict(metadata or {}),
+    )
+
+
+async def _assert_claimable(store: OPCStore, work_item_id: str, phase: Phase) -> None:
+    claimed = await store.claim_delegation_work_item_if_dispatchable(
+        work_item_id,
+        expected_phase=phase,
+        role_runtime_session_id="fresh-session",
+        seat_id="seat::executor",
+        task_id="fresh-task",
+    )
+    assert claimed is not None, f"{work_item_id} must be claimable after release"
+    assert claimed.phase == Phase.RUNNING
+    assert claimed.claimed_by_role_runtime_session_id == "fresh-session"
+
+
+@_async_test
+async def test_rework_verdict_releases_ownership(tmp_path: Path) -> None:
+    """REJECT → READY_FOR_REWORK blanks claim columns and mirror in one write."""
+    store = OPCStore(tmp_path / "tasks.db")
+    await store.initialize()
+    try:
+        item = _work_item(
+            "rework-target",
+            phase=Phase.AWAITING_MANAGER_REVIEW,
+            claimed_session="role-runtime::dead-worker",
+            claimed_seat="seat::executor",
+            metadata={
+                "claimed_by_role_session_id": "role-runtime::dead-worker",
+                "claimed_task_id": "dead-task",
+            },
+        )
+        await store.save_delegation_work_item(item)
+        applied = await store.apply_delegation_review_resolution(
+            item.work_item_id,
+            source_report_work_item_id="",
+            target_phase=Phase.READY_FOR_REWORK,
+            blocked_reason="",
+            metadata_updates={"rework_feedback": "fix the numbers"},
+        )
+        assert applied is not None
+        assert applied.phase == Phase.READY_FOR_REWORK
+        assert applied.claimed_by_role_runtime_session_id == ""
+        assert applied.claimed_by_seat_id == ""
+        assert applied.metadata["claimed_by_role_session_id"] == ""
+        assert applied.metadata["claimed_task_id"] == ""
+        await _assert_claimable(store, item.work_item_id, Phase.READY_FOR_REWORK)
+    finally:
+        await store.close()
+
+
+@_async_test
+async def test_phase_write_to_runnable_releases_ownership(tmp_path: Path) -> None:
+    """Any update that lands in READY/READY_FOR_REWORK drops the claim."""
+    store = OPCStore(tmp_path / "tasks.db")
+    await store.initialize()
+    try:
+        item = _work_item(
+            "synthesis-parent",
+            phase=Phase.WAITING_FOR_CHILDREN,
+            claimed_session="role-runtime::dead-parent",
+            claimed_seat="seat::executor",
+            metadata={
+                "claimed_by_role_session_id": "role-runtime::dead-parent",
+                "claimed_task_id": "parent-task",
+            },
+        )
+        await store.save_delegation_work_item(item)
+        updated = await store.update_delegation_work_item(
+            item.work_item_id,
+            phase=Phase.READY,
+            metadata_updates={"work_kind": "synthesize"},
+        )
+        assert updated is not None
+        assert updated.claimed_by_role_runtime_session_id == ""
+        assert updated.claimed_by_seat_id == ""
+        assert updated.metadata["claimed_by_role_session_id"] == ""
+        assert updated.metadata["claimed_task_id"] == ""
+        await _assert_claimable(store, item.work_item_id, Phase.READY)
+    finally:
+        await store.close()
+
+
+@_async_test
+async def test_claim_cas_ignores_stale_mirror(tmp_path: Path) -> None:
+    """Ownership truth is the claim columns; a stale mirror must not gate."""
+    store = OPCStore(tmp_path / "tasks.db")
+    await store.initialize()
+    try:
+        item = _work_item(
+            "stale-mirror",
+            phase=Phase.READY,
+            metadata={
+                "claimed_by_role_session_id": "role-runtime::forgotten",
+                "claimed_task_id": "forgotten-task",
+            },
+        )
+        await store.save_delegation_work_item(item)
+        await _assert_claimable(store, item.work_item_id, Phase.READY)
+    finally:
+        await store.close()
+
+
+@_async_test
+async def test_startup_sweep_heals_claimed_runnable_rows(tmp_path: Path) -> None:
+    """Legacy rows wedged in a runnable phase with a claim heal on restart."""
+    db_path = tmp_path / "tasks.db"
+    store = OPCStore(db_path)
+    await store.initialize()
+    try:
+        item = _work_item(
+            "legacy-wedged",
+            phase=Phase.READY_FOR_REWORK,
+            claimed_session="role-runtime::dead-worker",
+            claimed_seat="seat::executor",
+            metadata={
+                "claimed_by_role_session_id": "role-runtime::dead-worker",
+                "claimed_task_id": "dead-task",
+            },
+        )
+        await store.save_delegation_work_item(item)
+    finally:
+        await store.close()
+
+    reopened = OPCStore(db_path)
+    await reopened.initialize()
+    try:
+        healed = await reopened.get_delegation_work_item("legacy-wedged")
+        assert healed is not None
+        assert healed.phase == Phase.READY_FOR_REWORK
+        assert healed.claimed_by_role_runtime_session_id == ""
+        assert healed.claimed_by_seat_id == ""
+        assert healed.metadata["claimed_by_role_session_id"] == ""
+        assert healed.metadata["claimed_task_id"] == ""
+        await _assert_claimable(reopened, "legacy-wedged", Phase.READY_FOR_REWORK)
+    finally:
+        await reopened.close()