From e1c28c3889d08b93afb1cef8bf22626abf179c0d Mon Sep 17 00:00:00 2001 From: LZH-YS1998 Date: Tue, 7 Jul 2026 20:30:17 +0800 Subject: [PATCH] Revert "fix: enforce task-wait checkpoint lifecycle and stop UI probe from rescanning tasks table" This reverts commit d502d66c60e71ff5ef1a49ef5f22ff4956e2784a. --- opc/database/store.py | 19 --- opc/engine.py | 186 ++------------------ tests/test_stale_checkpoint_selfheal.py | 214 ------------------------ 3 files changed, 13 insertions(+), 406 deletions(-) delete mode 100644 tests/test_stale_checkpoint_selfheal.py diff --git a/opc/database/store.py b/opc/database/store.py index 4d53d35..d5ec522 100644 --- a/opc/database/store.py +++ b/opc/database/store.py @@ -1620,7 +1620,6 @@ 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); @@ -2793,24 +2792,6 @@ 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 a546434..540a1fd 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -8042,10 +8042,6 @@ 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: @@ -8947,116 +8943,6 @@ 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", {})) @@ -9255,21 +9141,6 @@ 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, ) @@ -9287,13 +9158,6 @@ 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): @@ -9328,8 +9192,6 @@ 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 @@ -9360,13 +9222,7 @@ class OPCEngine: if not sid: return "" try: - 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") + 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 "" @@ -9532,8 +9388,6 @@ 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: @@ -9652,20 +9506,10 @@ class OPCEngine: task.metadata["progress_log"] = progress await self.store.save_task(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] + tasks: list[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 @@ -9674,25 +9518,21 @@ class OPCEngine: await self.store.resolve_execution_checkpoint(checkpoint.checkpoint_id, status="resolved") - 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. + 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): 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 deleted file mode 100644 index f1d0de8..0000000 --- a/tests/test_stale_checkpoint_selfheal.py +++ /dev/null @@ -1,214 +0,0 @@ -"""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()