fix: enforce task-wait checkpoint lifecycle and stop UI probe from rescanning tasks table

A company-mode review left a pending task_user_input checkpoint behind: the
runtime carried the paused work item forward via approval-card grants and a
fresh review attempt without ever replying through the engine checkpoint. The
orphan row then captured the user's next chat message and resumed through the
deprecated multi-agent path with an empty task list, returning an empty reply.

Invariants added:
- write side: when a task settles (done/failed/cancelled) its pending
  task_user_input/task_peer_wait checkpoints are superseded
- read side: checkpoint matching lazily resolves rows whose task settled or
  whose linked work item reached a terminal phase (heals existing dirty DBs)
- resume: the primary task is always part of the resumed set; the
  MULTI_AGENT/COMPANY_MODE value-alias no longer routes company checkpoints
  into _execute_multi_agent (which silently returned "" on empty task lists)

Perf: get_latest_pending_checkpoint_for_session is called per task on every
UI sync tick, and its parent-session resolution loaded and JSON-parsed the
entire tasks table each time (24MB with inline artifact blobs) — a full core
pegged at 100% and the event loop starved so replies never surfaced. Now a
no-live-checkpoints fast path returns immediately, and the resolution uses a
targeted session_id query backed by a new tasks(session_id) index.

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

opc/database/store.py

@@ -1620,6 +1620,7 @@ class OPCStore:
             CREATE INDEX IF NOT EXISTS idx_tasks_project_status_created ON tasks(project_id, status, created_at);
             CREATE INDEX IF NOT EXISTS idx_tasks_project_priority_created ON tasks(project_id, priority, created_at);
             CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_id);
+            CREATE INDEX IF NOT EXISTS idx_tasks_session ON tasks(session_id);
             CREATE INDEX IF NOT EXISTS idx_messages_task ON agent_messages(task_id);
             CREATE INDEX IF NOT EXISTS idx_messages_status ON agent_messages(status);
             CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON agent_messages(timestamp);
@@ -2792,6 +2793,24 @@ class OPCStore:
         await self.hydrate_task_work_item_links(tasks)
         return tasks
 
+    async def get_tasks_by_session_id(
+        self,
+        session_id: str,
+        project_id: str | None = None,
+    ) -> list[Task]:
+        assert self._db
+        query = "SELECT * FROM tasks WHERE session_id = ?"
+        params: list[Any] = [session_id]
+        if project_id:
+            query += " AND project_id = ?"
+            params.append(project_id)
+        query += " ORDER BY priority ASC, created_at ASC"
+        async with self._db.execute(query, params) as cursor:
+            rows = await cursor.fetchall()
+            tasks = [self._row_to_task(row, cursor.description) for row in rows]
+        await self.hydrate_task_work_item_links(tasks)
+        return tasks
+
     async def update_task_status(self, task_id: str, status: TaskStatus) -> None:
         assert self._db
         await self._db.execute("UPDATE tasks SET status = ? WHERE id = ?", (status.value, task_id))

opc/engine.py

@@ -8042,6 +8042,10 @@ class OPCEngine:
                         result_content=result.content,
                         project=bool(task.project_id and task.project_id != "default"),
                     )
+        if task.status in {TaskStatus.DONE, TaskStatus.FAILED, TaskStatus.CANCELLED}:
+            await self._supersede_stale_task_wait_checkpoints(
+                task.id, reason=f"task settled as {task.status.value}"
+            )
         return result
 
     async def _attempt_capability_recovery(self, task: Task, result: TaskResult) -> None:
@@ -8943,6 +8947,116 @@ class OPCEngine:
             preferred = {"inbox", "reply_message", "send_dm", "ask_peer_and_wait", "respond_meeting"}
         return allowed.intersection(preferred)
 
+    # Checkpoint types that represent "a task is parked waiting for user input".
+    # Invariant: a pending row of these types is only valid while its task is
+    # actually in a waiting status; every other path must terminate them.
+    _TASK_WAIT_CHECKPOINT_TYPES = ("task_user_input", "task_peer_wait")
+
+    async def _supersede_stale_task_wait_checkpoints(self, task_id: str, *, reason: str) -> None:
+        """Terminate pending task-wait checkpoints once their task moves on.
+
+        The company runtime can carry a paused work item forward through its own
+        machinery (approval-card grants, a fresh review attempt) without ever
+        replying through the engine checkpoint. If the checkpoint row stays
+        pending it will capture the user's next unrelated chat message and route
+        it into a resume of a task that is no longer waiting.
+        """
+        if not task_id or not self.store:
+            return
+        supersede = getattr(self.store, "supersede_pending_checkpoints", None)
+        if not callable(supersede):
+            return
+        try:
+            superseded = await supersede(
+                project_id=self.project_id or "default",
+                task_id=task_id,
+                checkpoint_types=list(self._TASK_WAIT_CHECKPOINT_TYPES),
+            )
+        except Exception:
+            logger.opt(exception=True).warning(
+                f"Failed to supersede stale task-wait checkpoints for task {task_id}"
+            )
+            return
+        if superseded:
+            logger.info(
+                f"Superseded {len(superseded)} stale task-wait checkpoint(s) for task {task_id} ({reason})"
+            )
+
+    async def _checkpoint_task_still_waiting(self, checkpoint: ExecutionCheckpoint) -> bool:
+        """Whether a task-wait checkpoint still matches a genuinely waiting task.
+
+        Lazily resolves orphaned rows (task finished, failed, superseded by a
+        new review attempt, or deleted) as ``stale`` so historical dirty data
+        self-heals the first time it is considered for a resume. Non-task-wait
+        checkpoint types are always considered live here.
+        """
+        if str(checkpoint.checkpoint_type or "").strip() not in self._TASK_WAIT_CHECKPOINT_TYPES:
+            return True
+        task_id = str(
+            checkpoint.task_id or dict(checkpoint.payload or {}).get("task_id") or ""
+        ).strip()
+        if not task_id or not self.store:
+            return True
+        try:
+            task = await self.store.get_task(task_id)
+        except Exception:
+            logger.opt(exception=True).debug(
+                f"Could not verify task {task_id} for checkpoint {checkpoint.checkpoint_id}; keeping it"
+            )
+            return True
+        if task is None:
+            stale_reason = f"task {task_id} no longer exists"
+        elif task.status in {TaskStatus.DONE, TaskStatus.FAILED, TaskStatus.CANCELLED}:
+            stale_reason = f"task {task_id} settled as {task.status.value}"
+        else:
+            # A non-terminal task status proves nothing on its own:
+            # suspend/restart flows legitimately park a waiting task back at
+            # PENDING or RUNNING. The linked delegation work item is the
+            # authoritative signal — once its phase is terminal (a later review
+            # attempt or the manager closed it) or the item is gone, no runtime
+            # will ever come back to consume this checkpoint.
+            stale_reason = await self._task_work_item_closed_reason(task)
+        if not stale_reason:
+            return True
+        try:
+            await self.store.resolve_execution_checkpoint(checkpoint.checkpoint_id, status="stale")
+            logger.info(
+                f"Resolved stale {checkpoint.checkpoint_type} checkpoint "
+                f"{checkpoint.checkpoint_id} ({stale_reason})"
+            )
+        except Exception:
+            logger.opt(exception=True).warning(
+                f"Failed to resolve stale checkpoint {checkpoint.checkpoint_id}"
+            )
+        return False
+
+    async def _task_work_item_closed_reason(self, task: Task) -> str:
+        """Non-empty reason when the task's delegation work item is closed.
+
+        Returns "" when the task has no linked work item, the item cannot be
+        loaded, or the item is still in a live phase — i.e. keep the checkpoint.
+        """
+        work_item_id = linked_work_item_id_for_task(task)
+        if not work_item_id or not self.store:
+            return ""
+        getter = getattr(self.store, "get_delegation_work_item", None)
+        if not callable(getter):
+            return ""
+        try:
+            work_item = await getter(work_item_id)
+        except Exception:
+            logger.opt(exception=True).debug(
+                f"Could not load work item {work_item_id} while validating a checkpoint; keeping it"
+            )
+            return ""
+        if work_item is None:
+            return f"work item {work_item_id} no longer exists"
+        phase_raw = getattr(work_item, "phase", "")
+        phase = str(getattr(phase_raw, "value", phase_raw) or "").strip()
+        if phase in {Phase.APPROVED.value, Phase.FAILED.value, Phase.CANCELLED.value}:
+            return f"work item {work_item_id} closed with phase={phase}"
+        return ""
+
     async def _save_execution_checkpoint(self, data: dict[str, Any]) -> None:
         assert self.store
         payload = dict(data.get("payload", {}))
@@ -9141,6 +9255,21 @@ class OPCEngine:
             return None
         project_id = self.project_id or "default"
         requested_session_id = str(session_id or "").strip()
+        # Fast path: with no live checkpoint rows in the project there is
+        # nothing to surface, so skip the parent-session resolution below.
+        # Snapshot builders call this once per task on every UI sync tick, and
+        # that resolution loads (and JSON-parses) task rows each time.
+        checkpoint_probe = getattr(self.store, "get_execution_checkpoints", None)
+        if callable(checkpoint_probe):
+            try:
+                live_checkpoints = await checkpoint_probe(
+                    project_id=project_id,
+                    statuses=["pending", "resuming"],
+                )
+            except Exception:
+                live_checkpoints = None
+            if live_checkpoints is not None and len(live_checkpoints) == 0:
+                return None
         company_parent_session_id = await self._company_runtime_parent_session_for_session_id(
             requested_session_id,
         )
@@ -9158,6 +9287,13 @@ class OPCEngine:
             project_id,
             session_id=requested_session_id or None,
         )
+        # Skip (and lazily resolve) orphaned task-wait checkpoints; each stale
+        # row is marked resolved before re-querying, so this terminates.
+        while checkpoint is not None and not await self._checkpoint_task_still_waiting(checkpoint):
+            checkpoint = await self.store.get_latest_pending_checkpoint(
+                project_id,
+                session_id=requested_session_id or None,
+            )
         deferred_suspend_checkpoint: ExecutionCheckpoint | None = None
         if checkpoint and self._checkpoint_is_user_visible(checkpoint):
             if not self._is_company_runtime_suspend_checkpoint(checkpoint.checkpoint_type):
@@ -9192,6 +9328,8 @@ class OPCEngine:
         for pending in checkpoints:
             if not self._checkpoint_is_user_visible(pending):
                 continue
+            if not await self._checkpoint_task_still_waiting(pending):
+                continue
             if self._is_company_runtime_suspend_checkpoint(pending.checkpoint_type):
                 if deferred_suspend_checkpoint is None and str(pending.session_id or "").strip() == session_id:
                     deferred_suspend_checkpoint = pending
@@ -9222,7 +9360,13 @@ class OPCEngine:
         if not sid:
             return ""
         try:
-            tasks = await self.store.get_tasks(project_id=self.project_id or "default")
+            get_by_session = getattr(self.store, "get_tasks_by_session_id", None)
+            if callable(get_by_session):
+                # Targeted lookup: this runs on every UI sync tick, and loading
+                # every task in the project rescans the whole tasks table.
+                tasks = await get_by_session(sid, project_id=self.project_id or "default")
+            else:
+                tasks = await self.store.get_tasks(project_id=self.project_id or "default")
         except Exception:
             logger.opt(exception=True).debug("failed to load tasks while resolving company parent session")
             return ""
@@ -9388,6 +9532,8 @@ class OPCEngine:
                 if str(getattr(checkpoint, "checkpoint_type", "") or "").strip() == "company_delivery_feedback":
                     return None
                 return "This request is no longer active."
+            if not await self._checkpoint_task_still_waiting(checkpoint):
+                return "This request is no longer active."
         else:
             checkpoint = await self.get_latest_pending_checkpoint_for_session(session_id)
             if not checkpoint:
@@ -9506,10 +9652,20 @@ class OPCEngine:
         task.metadata["progress_log"] = progress
         await self.store.save_task(task)
 
-        tasks: list[Task] = []
+        # Sibling ids persisted by older checkpoints can be work-item ids rather
+        # than task UUIDs; unresolvable entries are skipped, but the primary
+        # task must always be part of the resumed set so the resume can never
+        # degenerate into executing an empty task list (which used to return an
+        # empty reply and silently swallow the user's message).
+        tasks: list[Task] = [task]
         for sibling_id in payload.get("task_ids", [task_id]):
+            if str(sibling_id) == str(task_id):
+                continue
             sibling = await self.store.get_task(sibling_id)
             if not sibling:
+                logger.warning(
+                    f"Checkpoint {checkpoint.checkpoint_id} references unknown sibling task {sibling_id!r}; skipping it"
+                )
                 continue
             if sibling.status == TaskStatus.BLOCKED:
                 sibling.status = TaskStatus.PENDING
@@ -9518,21 +9674,25 @@ class OPCEngine:
 
         await self.store.resolve_execution_checkpoint(checkpoint.checkpoint_id, status="resolved")
 
-        execution_mode = str(payload.get("execution_mode", ExecutionMode.SINGLE_AGENT.value))
+        raw_execution_mode = str(payload.get("execution_mode", ExecutionMode.SINGLE_AGENT.value))
-        # Re-register child tasks so WSHandler can dual-route progress
+        try:
-        # events from child work items to the parent session channel.
+            # MULTI_AGENT is a value alias of COMPANY_MODE, so normalizing to the
-        if execution_mode in (ExecutionMode.MULTI_AGENT.value, ExecutionMode.COMPANY_MODE.value):
+            # enum collapses both onto one branch instead of letting the legacy
+            # multi-agent branch shadow the company-mode one.
+            execution_mode = ExecutionMode(raw_execution_mode)
+        except ValueError:
+            execution_mode = ExecutionMode.SINGLE_AGENT
+        if execution_mode == ExecutionMode.COMPANY_MODE:
+            # Re-register child tasks so WSHandler can dual-route progress
+            # events from child work items to the parent session channel.
             self._reregister_company_runtime_children(tasks, checkpoint_session_id=checkpoint.session_id)
-        if execution_mode == ExecutionMode.MULTI_AGENT.value:
-            logger.info("[compat] Resumed MULTI_AGENT checkpoint → routing through company mode parallel")
-            plan_data = payload.get("company_work_item_plan") or task.metadata.get("company_work_item_plan")
-            if isinstance(plan_data, dict) and plan_data:
-                return await self._execute_company_mode(tasks, deserialize_company_work_item_runtime_plan(plan_data))
-            return await self._execute_multi_agent(tasks)
-        if execution_mode == ExecutionMode.COMPANY_MODE.value:
             plan_data = payload.get("company_work_item_plan") or task.metadata.get("company_work_item_plan")
             if isinstance(plan_data, dict) and plan_data:
                 return await self._execute_company_mode(tasks, deserialize_company_work_item_runtime_plan(plan_data))
+            logger.info(
+                f"Resuming company-mode checkpoint {checkpoint.checkpoint_id} without a runtime plan; "
+                f"re-running the paused task {task.id} directly"
+            )
         return await self._execute_single_agent([task], task.assigned_external_agent)
 
     async def _resume_peer_checkpoint(self, checkpoint: ExecutionCheckpoint, user_reply: str) -> str:

tests/test_stale_checkpoint_selfheal.py

@@ -0,0 +1,214 @@
+"""Regression tests: orphaned task-wait checkpoints must never swallow user messages.
+
+Reproduces the production failure where a company-mode review left a pending
+``task_user_input`` checkpoint behind (the runtime carried the flow forward via
+approval-card grants and a fresh review attempt, never replying through the
+checkpoint). The user's next chat message in the primary session was captured
+by that orphan row, routed into the deprecated multi-agent resume path with an
+empty task list, and answered with an empty string.
+"""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+import uuid
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+from opc.core.models import (
+    DelegationWorkItem,
+    ExecutionCheckpoint,
+    ExecutionMode,
+    Phase,
+    Task,
+    TaskStatus,
+)
+from opc.engine import OPCEngine
+
+
+class StaleCheckpointSelfHealTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.engine = OPCEngine(opc_home=Path(self._tmp.name), project_id="default")
+        await self.engine.initialize()
+        self.store = self.engine.store
+
+    async def asyncTearDown(self) -> None:
+        await self.store.close()
+        self._tmp.cleanup()
+
+    async def _save_task(self, status: TaskStatus, session_id: str) -> Task:
+        task = Task(
+            title="Review #1: competitive analysis",
+            session_id=session_id,
+            project_id="default",
+            status=status,
+        )
+        await self.store.save_task(task)
+        return task
+
+    async def _save_wait_checkpoint(self, task: Task, session_id: str, **payload_extra) -> ExecutionCheckpoint:
+        checkpoint = ExecutionCheckpoint(
+            project_id="default",
+            session_id=session_id,
+            checkpoint_type="task_user_input",
+            task_id=task.id,
+            payload={
+                "task_id": task.id,
+                "session_id": session_id,
+                "execution_mode": ExecutionMode.COMPANY_MODE.value,
+                "task_ids": [task.id],
+                "prompt": "Tool execution blocked by autonomy policy",
+                "review_level": "human",
+                **payload_extra,
+            },
+        )
+        await self.store.save_execution_checkpoint(checkpoint)
+        return checkpoint
+
+    async def _checkpoint_status(self, checkpoint_id: str) -> str:
+        rows = await self.store.get_execution_checkpoints(project_id="default")
+        for row in rows:
+            if row.checkpoint_id == checkpoint_id:
+                return str(row.status or "")
+        return "<not found>"
+
+    async def test_orphan_checkpoint_is_lazily_resolved_and_not_returned(self) -> None:
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.DONE, session_id)
+        checkpoint = await self._save_wait_checkpoint(task, session_id)
+
+        found = await self.engine.get_latest_pending_checkpoint_for_session(session_id)
+
+        self.assertIsNone(found)
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "stale")
+
+    async def test_waiting_checkpoint_is_still_returned(self) -> None:
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.AWAITING_HUMAN, session_id)
+        checkpoint = await self._save_wait_checkpoint(task, session_id)
+
+        found = await self.engine.get_latest_pending_checkpoint_for_session(session_id)
+
+        self.assertIsNotNone(found)
+        self.assertEqual(found.checkpoint_id, checkpoint.checkpoint_id)
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "pending")
+
+    async def test_user_message_falls_through_to_normal_processing(self) -> None:
+        """A plain follow-up question must not be consumed by an orphan checkpoint."""
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.DONE, session_id)
+        checkpoint = await self._save_wait_checkpoint(task, session_id)
+
+        result = await self.engine._maybe_resume_checkpoint(
+            "你的交付文件在哪里？", session_id=session_id
+        )
+
+        self.assertIsNone(result)  # None → caller processes it as a fresh message
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "stale")
+
+    async def test_explicit_reply_to_dead_checkpoint_reports_inactive(self) -> None:
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.DONE, session_id)
+        checkpoint = await self._save_wait_checkpoint(task, session_id)
+
+        result = await self.engine._maybe_resume_checkpoint(
+            "approve",
+            session_id=session_id,
+            reply_metadata={
+                "response_to_checkpoint_id": checkpoint.checkpoint_id,
+                "response_to_checkpoint_type": "task_user_input",
+            },
+        )
+
+        self.assertEqual(result, "This request is no longer active.")
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "stale")
+
+    async def _link_work_item(self, task: Task, phase: Phase) -> None:
+        item = DelegationWorkItem(
+            work_item_id=f"review::{uuid.uuid4()}::v1",
+            title="Review #1",
+            phase=phase,
+        )
+        await self.store.save_delegation_work_item(item)
+        await self.store.link_work_item_runtime_task(item.work_item_id, task.id)
+
+    async def test_checkpoint_with_closed_work_item_is_stale_even_if_task_not_terminal(self) -> None:
+        """Production case: the swallowed resume left the task at PENDING, but the
+        review attempt's work item had long been approved — the checkpoint is dead."""
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.PENDING, session_id)
+        await self._link_work_item(task, Phase.APPROVED)
+        checkpoint = await self._save_wait_checkpoint(task, session_id)
+
+        found = await self.engine.get_latest_pending_checkpoint_for_session(session_id)
+
+        self.assertIsNone(found)
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "stale")
+
+    async def test_checkpoint_with_open_work_item_is_kept(self) -> None:
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.RUNNING, session_id)
+        await self._link_work_item(task, Phase.RUNNING)
+        checkpoint = await self._save_wait_checkpoint(task, session_id)
+
+        found = await self.engine.get_latest_pending_checkpoint_for_session(session_id)
+
+        self.assertIsNotNone(found)
+        self.assertEqual(found.checkpoint_id, checkpoint.checkpoint_id)
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "pending")
+
+    async def test_task_settling_supersedes_pending_wait_checkpoints(self) -> None:
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.AWAITING_HUMAN, session_id)
+        checkpoint = await self._save_wait_checkpoint(task, session_id)
+
+        await self.engine._supersede_stale_task_wait_checkpoints(task.id, reason="test settle")
+
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "superseded")
+
+    async def test_resume_with_unresolvable_siblings_never_returns_empty(self) -> None:
+        """Sibling ids that are work-item ids (not task UUIDs) must not empty the task list."""
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.AWAITING_HUMAN, session_id)
+        checkpoint = await self._save_wait_checkpoint(
+            task,
+            session_id,
+            task_ids=["review::467f36ff::v1"],  # work-item id, unresolvable as a task
+            company_work_item_plan=None,
+        )
+
+        self.engine._execute_single_agent = AsyncMock(return_value="resumed reply")
+        self.engine._execute_multi_agent = AsyncMock(return_value="")
+        self.engine._execute_company_mode = AsyncMock(return_value="")
+
+        result = await self.engine._resume_task_checkpoint(checkpoint, "继续")
+
+        self.assertEqual(result, "resumed reply")
+        self.engine._execute_multi_agent.assert_not_awaited()
+        self.engine._execute_company_mode.assert_not_awaited()
+        (called_tasks, _agent), _ = self.engine._execute_single_agent.await_args
+        self.assertEqual([t.id for t in called_tasks], [task.id])
+        self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "resolved")
+
+    async def test_resume_with_plan_routes_to_company_mode(self) -> None:
+        session_id = str(uuid.uuid4())
+        task = await self._save_task(TaskStatus.AWAITING_HUMAN, session_id)
+        checkpoint = await self._save_wait_checkpoint(
+            task,
+            session_id,
+            company_work_item_plan={"profile": "corporate"},
+        )
+
+        self.engine._execute_company_mode = AsyncMock(return_value="company resumed")
+        self.engine._execute_multi_agent = AsyncMock(return_value="")
+
+        result = await self.engine._resume_task_checkpoint(checkpoint, "继续")
+
+        self.assertEqual(result, "company resumed")
+        self.engine._execute_multi_agent.assert_not_awaited()
+
+
+if __name__ == "__main__":
+    unittest.main()