diff --git a/opc/database/store.py b/opc/database/store.py index 9eeaed5..a63e5e2 100644 --- a/opc/database/store.py +++ b/opc/database/store.py @@ -5458,12 +5458,30 @@ async def update_delegation_work_item( item.claimed_by_role_runtime_session_id = str(claimed_by_role_runtime_session_id or "").strip() if claimed_by_seat_id is not None: item.claimed_by_seat_id = str(claimed_by_seat_id or "").strip() - if metadata_unset or metadata_updates: + # Centralized claim invariant: clearing the runtime-claim column MUST + # also clear the metadata mirror (claimed_by_role_session_id / + # claimed_task_id) that the dispatch CAS + # (claim_delegation_work_item_if_dispatchable) predicates on. + # Any release path that zeroed only the column — the turn-boundary + # release into a non-terminal phase (transition_work_item), the + # dependency-frontier wake (refresh_dependents_for_run), a review + # recovery exit — otherwise left the card in READY / READY_FOR_REWORK + # but permanently unclaimable: the 3-agent parent-resume deadlock and + # the 4-agent QA/delivery deadlock. Enforcing it here closes the + # whole bug class at the single write chokepoint. + clear_claim_mirror = ( + claimed_by_role_runtime_session_id is not None + and not str(claimed_by_role_runtime_session_id or "").strip() + ) + if metadata_unset or metadata_updates or clear_claim_mirror: metadata = dict(item.metadata or {}) for key in list(metadata_unset or []): metadata.pop(str(key), None) if metadata_updates: metadata.update(dict(metadata_updates)) + if clear_claim_mirror: + 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) @@ -5722,6 +5740,16 @@ async def reopen_approved_delegation_work_item_for_rework( metadata.pop(str(key), None) if metadata_updates: metadata.update(dict(metadata_updates)) + if release_claim: + # Centralized claim invariant (mirrors the update_delegation_work_item + # chokepoint): this path clears the runtime-claim column with its own + # SQL and bypasses that helper, so it must clear the metadata mirror + # the dispatch CAS predicates on. Otherwise a reopened APPROVED card + # lands in READY / READY_FOR_REWORK with a stale + # claimed_by_role_session_id / claimed_task_id and stays permanently + # unclaimable — the same deadlock class the store fix closes. + 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() diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 6a91c99..14f71ee 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -6454,7 +6454,22 @@ def _rehydrate_parked_member_sessions(self, work_items: list[Any]) -> None: session.focused_work_item_id = "" session.updated_at = now continue - if is_dispatchable(work_item): + # If focused work item is in a terminal phase (approved, done, etc.), + # clear the focus so the session can pick up new work. + if work_item.phase in DONE_PHASES: + session.status = "idle" + session.resident_status = "idle" + session.focused_work_item_id = "" + session.updated_at = now + continue + # Unpark only when the focused card is genuinely dispatchable. + # is_dispatchable already covers READY / READY_FOR_REWORK and + # orphaned in-flight cards, while still respecting dispatch_hold / + # queued_behind_session / attempt-ledger brakes. A bare + # is_runnable(phase) check would override those holds and unpark a + # session whose card must stay parked. + dispatchable = is_dispatchable(work_item) + if dispatchable: session.status = "idle" session.resident_status = "idle" session.focused_work_item_id = "" diff --git a/tests/test_claim_mirror_release.py b/tests/test_claim_mirror_release.py new file mode 100644 index 0000000..e46bbfb --- /dev/null +++ b/tests/test_claim_mirror_release.py @@ -0,0 +1,241 @@ +"""Regression tests for the claim-mirror release invariant. + +The dispatch CAS (``claim_delegation_work_item_if_dispatchable``) predicates +on BOTH the runtime-claim column (``claimed_by_role_runtime_session_id`` / +``claimed_by_seat_id``) AND the metadata mirror +(``claimed_by_role_session_id`` / ``claimed_task_id``) being empty. Any +release path that clears only the column leaves the card in a runnable +phase but permanently unclaimable — a deadlock. + +These tests pin the invariant "clearing the column also clears the mirror" +for the three release paths that matter: + +1. ``transition_work_item(release_claim=True)`` into a non-terminal phase + (the review-reject → READY_FOR_REWORK release), routed through the + centralized ``update_delegation_work_item`` chokepoint. +2. ``refresh_dependents_for_run`` waking a parent out of + WAITING_FOR_CHILDREN once all children are APPROVED (clear_claim_on_wake). +3. ``reopen_approved_delegation_work_item_for_rework(release_claim=True)``, + which clears the column with its own SQL and must clear the mirror too. + +Each asserts that after the release both the column and the mirror are +empty AND the CAS can re-claim the card. +""" +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from opc.core.models import DelegationWorkItem, Phase +from opc.database.store import OPCStore +from opc.layer2_organization import phase_hooks # noqa: F401 (register hooks) +from opc.layer2_organization.work_item_transition import ( + refresh_dependents_for_run, + transition_work_item, +) + + +class ClaimMirrorReleaseTests(unittest.IsolatedAsyncioTestCase): + async def _store(self) -> OPCStore: + tmpdir = tempfile.TemporaryDirectory() + self.addCleanup(tmpdir.cleanup) + store = OPCStore(Path(tmpdir.name) / "tasks.db") + await store.initialize() + self.addAsyncCleanup(store.close) + return store + + async def _seed_item( + self, + store: OPCStore, + *, + work_item_id: str = "wi-1", + run_id: str = "run-1", + phase: Phase = Phase.READY, + metadata: dict | None = None, + ) -> DelegationWorkItem: + item = DelegationWorkItem( + work_item_id=work_item_id, + run_id=run_id, + role_id="executor", + seat_id="seat-1", + title="Execution", + summary="Do the work.", + kind="execute", + projection_id="execution", + phase=phase, + metadata=dict(metadata or {}), + ) + await store.save_delegation_work_item(item) + return item + + async def _claim( + self, + store: OPCStore, + work_item_id: str, + phase: Phase, + *, + role_session_id: str = "role-sess-1", + task_id: str = "task-1", + ) -> DelegationWorkItem: + claimed = await store.claim_delegation_work_item_if_dispatchable( + work_item_id, + expected_phase=phase, + role_runtime_session_id=role_session_id, + seat_id="seat-1", + task_id=task_id, + ) + assert claimed is not None, "claim CAS unexpectedly failed" + return claimed + + def _assert_claim_held(self, item: DelegationWorkItem) -> None: + metadata = dict(item.metadata or {}) + self.assertTrue(str(item.claimed_by_role_runtime_session_id or "").strip()) + self.assertTrue(str(metadata.get("claimed_by_role_session_id", "") or "").strip()) + self.assertTrue(str(metadata.get("claimed_task_id", "") or "").strip()) + + def _assert_claim_fully_released(self, item: DelegationWorkItem) -> None: + metadata = dict(item.metadata or {}) + self.assertEqual(str(item.claimed_by_role_runtime_session_id or "").strip(), "") + self.assertEqual(str(item.claimed_by_seat_id or "").strip(), "") + self.assertEqual(str(metadata.get("claimed_by_role_session_id", "") or "").strip(), "") + self.assertEqual(str(metadata.get("claimed_task_id", "") or "").strip(), "") + + async def test_reject_release_clears_column_and_mirror_and_reclaims(self) -> None: + """Scenario 1: review-reject release into READY_FOR_REWORK. + + A claimed card moves RUNNING → AWAITING_MANAGER_REVIEW, then the + reviewer rejects it: transition_work_item(release_claim=True) into + READY_FOR_REWORK. Both the column and the mirror must be empty and + the CAS must re-claim the card. + """ + store = await self._store() + await self._seed_item(store) + claimed = await self._claim(store, "wi-1", Phase.READY) + self._assert_claim_held(claimed) + + await transition_work_item( + store, + "wi-1", + target_phase=Phase.AWAITING_MANAGER_REVIEW, + reason="worker_completed", + ) + released = await transition_work_item( + store, + "wi-1", + target_phase=Phase.READY_FOR_REWORK, + reason="manager_rejected", + release_claim=True, + ) + assert released is not None + self.assertEqual(released.phase, Phase.READY_FOR_REWORK) + self._assert_claim_fully_released(released) + + reclaimed = await store.claim_delegation_work_item_if_dispatchable( + "wi-1", + expected_phase=Phase.READY_FOR_REWORK, + role_runtime_session_id="role-sess-2", + seat_id="seat-1", + task_id="task-2", + ) + self.assertIsNotNone(reclaimed, "CAS must re-claim after reject-release") + + async def test_children_approved_wake_clears_column_and_mirror_and_reclaims(self) -> None: + """Scenario 2: parent woken out of WAITING_FOR_CHILDREN. + + A parent holds a durable claim while WAITING_FOR_CHILDREN. Once the + child is APPROVED, refresh_dependents_for_run wakes the parent to a + non-terminal phase and releases the claim (clear_claim_on_wake). + Both the column and the mirror must be empty and the CAS must + re-claim the parent. + """ + store = await self._store() + await self._seed_item( + store, + work_item_id="parent", + metadata={"dependency_work_item_ids": ["child"]}, + ) + await self._seed_item( + store, + work_item_id="child", + phase=Phase.APPROVED, + ) + # Parent claims, starts running, then parks on its child. + parent_claimed = await self._claim(store, "parent", Phase.READY) + self._assert_claim_held(parent_claimed) + await transition_work_item( + store, + "parent", + target_phase=Phase.WAITING_FOR_CHILDREN, + reason="delegated_children", + ) + parked = await store.get_delegation_work_item("parent") + assert parked is not None + # The park keeps the durable claim (no release on this transition). + self._assert_claim_held(parked) + + changed = await refresh_dependents_for_run(store, run_id="run-1") + self.assertTrue(changed) + + woken = await store.get_delegation_work_item("parent") + assert woken is not None + self.assertIn(woken.phase, {Phase.READY, Phase.RUNNING}) + self.assertNotEqual(woken.phase, Phase.WAITING_FOR_CHILDREN) + self._assert_claim_fully_released(woken) + + reclaimed = await store.claim_delegation_work_item_if_dispatchable( + "parent", + expected_phase=woken.phase, + role_runtime_session_id="role-sess-2", + seat_id="seat-1", + task_id="task-2", + ) + self.assertIsNotNone(reclaimed, "CAS must re-claim the woken parent") + + async def test_reopen_approved_for_rework_clears_mirror_and_reclaims(self) -> None: + """Scenario 3: reopen an APPROVED card for rework. + + reopen_approved_delegation_work_item_for_rework clears the claim + column with its own SQL, bypassing update_delegation_work_item. It + must still clear the metadata mirror so the reopened card does not + deadlock on a stale audit claim. + """ + store = await self._store() + # An APPROVED card retains its claim as an audit record of the last + # executor (column + mirror). + await self._seed_item( + store, + phase=Phase.APPROVED, + metadata={ + "claimed_by_role_session_id": "role-sess-1", + "claimed_task_id": "task-1", + }, + ) + held = await store.get_delegation_work_item("wi-1") + assert held is not None + held.claimed_by_role_runtime_session_id = "role-sess-1" + held.claimed_by_seat_id = "seat-1" + await store.save_delegation_work_item(held) + self._assert_claim_held(await store.get_delegation_work_item("wi-1")) + + reopened = await store.reopen_approved_delegation_work_item_for_rework( + "wi-1", + target_phase=Phase.READY_FOR_REWORK, + release_claim=True, + ) + assert reopened is not None + self.assertEqual(reopened.phase, Phase.READY_FOR_REWORK) + self._assert_claim_fully_released(reopened) + + reclaimed = await store.claim_delegation_work_item_if_dispatchable( + "wi-1", + expected_phase=Phase.READY_FOR_REWORK, + role_runtime_session_id="role-sess-2", + seat_id="seat-1", + task_id="task-2", + ) + self.assertIsNotNone(reclaimed, "CAS must re-claim after reopen-for-rework") + + +if __name__ == "__main__": + unittest.main()