diff --git a/opc/database/store.py b/opc/database/store.py index d5ec522..4d53d35 100644 --- a/opc/database/store.py +++ b/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)) diff --git a/opc/engine.py b/opc/engine.py index 540a1fd..a546434 100644 --- a/opc/engine.py +++ b/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)) - # Re-register child tasks so WSHandler can dual-route progress - # events from child work items to the parent session channel. - if execution_mode in (ExecutionMode.MULTI_AGENT.value, ExecutionMode.COMPANY_MODE.value): + raw_execution_mode = str(payload.get("execution_mode", ExecutionMode.SINGLE_AGENT.value)) + try: + # MULTI_AGENT is a value alias of COMPANY_MODE, so normalizing to the + # 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: diff --git a/tests/test_stale_checkpoint_selfheal.py b/tests/test_stale_checkpoint_selfheal.py new file mode 100644 index 0000000..f1d0de8 --- /dev/null +++ b/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 "" + + 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()