diff --git a/opc/database/store.py b/opc/database/store.py index 9eeaed5..367e73a 100644 --- a/opc/database/store.py +++ b/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, diff --git a/opc/layer2_organization/work_item_transition.py b/opc/layer2_organization/work_item_transition.py index d1c433d..c7d0c08 100644 --- a/opc/layer2_organization/work_item_transition.py +++ b/opc/layer2_organization/work_item_transition.py @@ -179,14 +179,13 @@ 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 - # the metadata claim mirror keys so the row can never satisfy a - # future claim-CAS predicate by accident. + # release, stranding a dead claim. The metadata mirror keys are + # blanked alongside the columns so the mirror never outlives the + # 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", "") + merged.setdefault("claimed_by_role_session_id", "") + merged.setdefault("claimed_task_id", "") try: result = await store.update_delegation_work_item(work_item_id, **kwargs) except InvalidPhaseTransition: @@ -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 - # WAITING_FOR_CHILDREN toward a non-terminal phase. The old - # condition ("only when all children approved AND target is - # RUNNING") left a gap: when a child went READY_FOR_REWORK, - # the refresh now fires (per _DEPENDENT_REFRESH_TARGETS) but - # the parent stayed in WAITING_FOR_CHILDREN with a stale claim, - # 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". + # Waking a parent out of WAITING_FOR_CHILDREN must orphan it so + # the dispatcher can re-pick it. For READY targets the store's + # phase-write invariant releases ownership; only the direct + # →RUNNING wake still needs the explicit clear (columns and + # mirror in the same write). Terminal targets keep the claim as + # 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 not in DONE_PHASES + and target_phase == Phase.RUNNING ) - 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, diff --git a/tests/test_claim_release_invariant.py b/tests/test_claim_release_invariant.py new file mode 100644 index 0000000..6c0423e --- /dev/null +++ b/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()