Fix two ways a company-mode session becomes permanently unusable (#52)

opc/database/store.py

@@ -19560,6 +19560,15 @@ async def settle_stale_delegation_run_claims_for_controller(
                         # persisted the claim and primary link, then
                         # terminalized the WorkItem/Task without controller
                         # owner/generation fields.
+                        # The same holds before a card is ever dispatched: an
+                        # empty envelope on *both* sides is symmetric, not
+                        # mixed.  The phase is therefore not required to be
+                        # terminal — `linked_task_status ==
+                        # expected_linked_task_status` already proves the pair
+                        # agrees under `task_status_for_phase`, and every other
+                        # conjunct pins identity.  Requiring DONE_PHASES here
+                        # made a consistent `waiting_dependencies`/`blocked`
+                        # pair raise and wedged the session for good.
                         # Under the winning run lease it is safe to release
                         # that exact terminal claim, but it is *not* safe to
                         # invent a Task attempt.  A later genuine rework claim
@@ -19604,7 +19613,7 @@ async def settle_stale_delegation_run_claims_for_controller(
                             == explicit_work_item_projection_id
                             and linked_task_projection_id
                             == explicit_work_item_projection_id
-                            and prior_phase in DONE_PHASES
+                            and prior_phase is not None
                             and linked_task_status
                             == expected_linked_task_status
                         )

opc/engine.py

@@ -142,6 +142,7 @@
     IN_PROGRESS_PHASES,
     IN_REVIEW_PHASES,
     InvalidPhaseTransition,
+    coerce_phase,
     task_status_for_phase,
 )
 from opc.layer2_organization.prompt_contract import (
@@ -10690,6 +10691,35 @@ async def _ignore_company_delivery_feedback_consumed(
         )
         return "Self-evolution review ignored."
 
+    async def _linked_work_item_phase_is_settled(self, task: Task) -> bool:
+        """True when this Task's linked WorkItem already reached a done phase.
+
+        Conservative by design: any missing link, missing store method, or
+        lookup failure reports False so the caller keeps its previous
+        behaviour.
+        """
+
+        if not self.store:
+            return False
+        work_item_id = str(linked_work_item_id_for_task(task) or "").strip()
+        if not work_item_id:
+            return False
+        getter = getattr(self.store, "get_delegation_work_item", None)
+        if not callable(getter):
+            return False
+        try:
+            work_item = await getter(work_item_id)
+        except Exception:
+            logger.opt(exception=True).debug(
+                "failed to inspect linked work item {} for delivery review task {}",
+                work_item_id,
+                getattr(task, "id", ""),
+            )
+            return False
+        if work_item is None:
+            return False
+        return coerce_phase(getattr(work_item, "phase", None)) in DONE_PHASES
+
     async def _ensure_open_final_delivery_review_checkpoints(
         self,
         plan: CompanyWorkItemRuntimePlan,
@@ -10710,6 +10740,18 @@ async def _ensure_open_final_delivery_review_checkpoints(
             and self._is_open_final_delivery_review_task(task)
             and not self._metadata_flag_true(dict(getattr(task, "metadata", {}) or {}).get("self_evolution_review_completed", False))
         ]
+        # The Task closure and the WorkItem approval are two writes on two
+        # connections, so a crash between them leaves an AWAITING_HUMAN Task
+        # next to an already settled WorkItem.  Republishing the card here
+        # would re-park the Task for good: the pair then fails the attempt
+        # envelope check in every later controller takeover and the session
+        # can never be resumed.  A settled WorkItem means the review is over,
+        # so leave the card closed and let the normal projection converge.
+        open_delivery_tasks = [
+            task
+            for task in open_delivery_tasks
+            if not await self._linked_work_item_phase_is_settled(task)
+        ]
         if not open_delivery_tasks:
             return
         try:

tests/test_claim_release_invariant.py

@@ -716,3 +716,71 @@ async def test_controller_takeover_rejects_mixed_linked_envelope_atomically(
             assert await store.get_task(task.id) == before_tasks[task.id]
     finally:
         await store.close()
+
+
+@_async_test
+async def test_controller_takeover_skips_never_dispatched_consistent_pair(
+    tmp_path: Path,
+) -> None:
+    """An empty envelope on both sides is symmetric, not mixed."""
+    store = OPCStore(tmp_path / "tasks.db")
+    await store.initialize()
+    try:
+        await store.save_delegation_run(
+            DelegationRun(
+                run_id="claim-invariant-run",
+                project_id="default",
+                session_id="claim-invariant-root",
+                execution_model="multi_team_org",
+                status="running",
+                lifecycle_status="active",
+            )
+        )
+        suffix = "never-dispatched"
+        task = Task(
+            id=f"{suffix}-task",
+            project_id="default",
+            session_id="claim-invariant-root",
+            title=suffix,
+            # task_status_for_phase(WAITING_DEPENDENCIES) is BLOCKED, so the
+            # pair agrees even though neither side carries a credential.
+            status=TaskStatus.BLOCKED,
+            metadata={
+                "delegation_run_id": "claim-invariant-run",
+                "work_item_projection_id": suffix,
+                "work_item_runtime": True,
+            },
+        )
+        item = _work_item(
+            suffix,
+            phase=Phase.WAITING_DEPENDENCIES,
+            claimed_session=f"role-runtime::{suffix}",
+            claimed_seat="seat::executor",
+            metadata={
+                "claimed_by_role_session_id": f"role-runtime::{suffix}",
+                "claimed_task_id": task.id,
+            },
+        )
+        await store.save_delegation_work_item(item)
+        await store.save_task(task)
+        assert await store.link_work_item_runtime_task(item.work_item_id, task.id)
+        lease = await store.acquire_delegation_run_controller_lease(
+            "claim-invariant-run",
+            project_id="default",
+            root_session_id="claim-invariant-root",
+            owner_token="claim-invariant-recovery",
+            lease_seconds=60,
+        )
+        assert lease.acquired
+
+        # Must not raise: a card that was never dispatched has no attempt
+        # credential to reconcile on either side.
+        await store.settle_stale_delegation_run_claims_for_controller(
+            "claim-invariant-run",
+            project_id="default",
+            root_session_id="claim-invariant-root",
+            owner_token="claim-invariant-recovery",
+            generation=lease.generation,
+        )
+    finally:
+        await store.close()