From 447516d93cd3a863bee65c49bb72f3a86599337b Mon Sep 17 00:00:00 2001 From: LZH-YS1998 Date: Wed, 8 Jul 2026 10:52:02 +0800 Subject: [PATCH] fix: converge approval park/resume through the work-item state machine Four coordinated fixes for company runs permanently stalling around tool-approval parks (project 1111 forensics + live reproduction): - company executor dispatch loop now exits convergently when nothing is in flight and every remaining task waits on a human: immediate parked exit when all waiters have pending checkpoints, bounded stall ticks otherwise. Previously it polled sleep(5) forever, hanging the turn (observed 7.5h), never answering the user, and holding claims that blocked any later rescue turn. - _resume_task_checkpoint routes work-item runtime tasks through the delegation state machine: release the human wait via the legal AWAITING_HUMAN -> READY recovery exit, clear stale claims, and hand the item back to the dispatcher (runtime-snapshot fallback when the checkpoint payload lacks a plan) instead of a detached single-agent re-run that never advances the work item phase. - startup recovery reverse self-heal: an awaiting_human task whose park checkpoint was already resolved (human answered, resume cut off before the phase write) is reopened for dispatch instead of being preserved as a wait nobody can end. Runs before the metadata-plan gate since modern runs no longer carry the plan in task metadata. - pause checkpoints record execution_mode from the durable work_item_runtime marker instead of volatile task metadata, which degraded to task_mode after a first resume and misrouted the next one. Validated end-to-end on a live native-agent minimal-org run (park -> approve -> re-park -> approve -> complete -> deliver, plus synthesized crash-between-resolve-and-phase-write healed on restart). Co-Authored-By: Claude Fable 5 --- opc/engine.py | 224 +++++++++++++++++++++++- opc/layer2_organization/company_mode.py | 75 ++++++++ 2 files changed, 296 insertions(+), 3 deletions(-) diff --git a/opc/engine.py b/opc/engine.py index ab97c7b..93c0e49 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -6274,6 +6274,117 @@ class OPCEngine: return True return bool(str(metadata.get("feedback_scope", "") or "").strip()) + async def _resolved_human_wait_checkpoint_task_ids(self) -> set[str]: + """Task ids whose human-wait checkpoint was already answered. + + Covers ``task_user_input`` / ``company_work_item_gate`` checkpoints in + ``resolved`` status: the human replied, so a task still parked in + ``awaiting_human`` for one of these ids is waiting on input that was + already given (the resume was interrupted before the state machine + advanced) and can be safely reopened. + """ + if not self.store: + return set() + try: + checkpoints = await self.store.get_execution_checkpoints( + project_id=self.project_id or "default", + checkpoint_types=["task_user_input", "company_work_item_gate"], + statuses=["resolved"], + ) + except Exception: + logger.opt(exception=True).debug( + "resolved_human_wait_checkpoint_task_ids: checkpoint load failed" + ) + return set() + task_ids: set[str] = set() + for checkpoint in checkpoints or []: + payload = dict(getattr(checkpoint, "payload", {}) or {}) + task_id = str( + checkpoint.task_id + or payload.get("waiting_task_id", "") + or payload.get("task_id", "") + or "" + ).strip() + if task_id: + task_ids.add(task_id) + return task_ids + + async def _reopen_answered_human_waits(self, tasks: list[Task]) -> int: + """Reverse self-heal for company tasks stuck on an answered human wait. + + A task parked in ``awaiting_human`` whose park checkpoint was already + RESOLVED means the human answered but the resume was cut off before the + work item advanced (process death between checkpoint resolution and + phase write). Reopen it for dispatch instead of preserving a wait + nobody can end. Mirror direction of the stale-checkpoint self-heal. + """ + if not self.store or not tasks: + return 0 + candidates = [ + task + for task in tasks + if task.status == TaskStatus.AWAITING_HUMAN + and str((task.metadata or {}).get("dispatch_hold", "") or "").strip() != "company_runtime_suspended" + and str((task.metadata or {}).get("company_runtime_stop_state", "") or "").strip() + not in {"suspending", "suspended"} + and not self._is_company_feedback_waiting_task(task) + ] + if not candidates: + return 0 + try: + pending_checkpoints = await self.store.get_pending_checkpoints( + project_id=self.project_id or "default" + ) + except Exception: + logger.opt(exception=True).debug( + "reopen_answered_human_waits: pending checkpoint load failed" + ) + return 0 + pending_task_ids = { + str( + checkpoint.task_id + or checkpoint.payload.get("waiting_task_id") + or checkpoint.payload.get("task_id") + or "" + ).strip() + for checkpoint in pending_checkpoints or [] + } + candidates = [task for task in candidates if task.id not in pending_task_ids] + if not candidates: + return 0 + resolved_task_ids = await self._resolved_human_wait_checkpoint_task_ids() + updated = 0 + for task in candidates: + if task.id not in resolved_task_ids: + continue + if await self._reopen_answered_human_wait_after_restart(task): + updated += 1 + return updated + + async def _reopen_answered_human_wait_after_restart(self, task: Task) -> bool: + """Reopen an ``awaiting_human`` task whose checkpoint was already resolved.""" + if not self.store: + return False + released = await self._release_work_item_human_wait( + task, reason="startup_resolved_checkpoint" + ) + task.metadata = dict(task.metadata or {}) + task.metadata["startup_reconcile_reopened_answered_wait"] = { + "detected_at": datetime.now().isoformat(), + "previous_status": getattr(task.status, "value", str(task.status)), + "work_item_phase_released": released, + } + task.status = TaskStatus.PENDING + task.result = None + await self.store.save_task(task) + logger.info( + "Startup recovery reopened task {} left in awaiting_human with an already-resolved " + "human-wait checkpoint (work_item_released={})", + task.id, + released, + ) + return True + async def _preserve_stable_waiting_task_after_restart( self, task: Task, @@ -6456,6 +6567,15 @@ class OPCEngine: if await self._mark_task_interrupted(task, reason=reason, session=session): updated += 1 + if runtime_groups: + # Reverse self-heal runs before the plan-gated reconcile below: + # modern runs keep the runtime plan in checkpoint payloads / + # snapshots rather than task metadata, so groups without a + # metadata plan would otherwise skip recovery entirely. + updated += await self._reopen_answered_human_waits( + [task for group in runtime_groups.values() for task in group] + ) + for parent_session_id, group in runtime_groups.items(): plan_data = None for task in sorted(group, key=lambda item: (item.created_at, item.id), reverse=True): @@ -9168,6 +9288,18 @@ class OPCEngine: } ) + def _checkpoint_execution_mode_for_task(self, task: Task) -> str: + """Execution mode to record on a pause checkpoint. + + Work-item runtime membership is the durable signal; the + ``execution_mode`` metadata field is volatile and has been observed to + degrade to ``task_mode`` after a resume, which then misroutes the next + resume away from the company state machine. + """ + if is_work_item_runtime_metadata(task.metadata): + return ExecutionMode.COMPANY_MODE.value + return str(task.metadata.get("execution_mode", ExecutionMode.SINGLE_AGENT.value)) + async def _save_task_pause_checkpoint(self, task: Task, result: TaskResult) -> None: pause_request = dict(result.artifacts.get("pause_request", {})) if result.artifacts else {} runtime_payload = self._build_runtime_checkpoint_payload(task, result) @@ -9221,7 +9353,7 @@ class OPCEngine: "payload": { "task_id": task.id, "session_id": task.session_id, - "execution_mode": task.metadata.get("execution_mode", ExecutionMode.SINGLE_AGENT.value), + "execution_mode": self._checkpoint_execution_mode_for_task(task), "task_ids": list(task.metadata.get("execution_task_ids", [task.id])), "org_version": task.metadata.get("org_version", 1), "runtime_topology_version": task.metadata.get("runtime_topology_version", 1), @@ -9249,7 +9381,7 @@ class OPCEngine: "payload": { "task_id": task.id, "session_id": task.session_id, - "execution_mode": task.metadata.get("execution_mode", ExecutionMode.SINGLE_AGENT.value), + "execution_mode": self._checkpoint_execution_mode_for_task(task), "task_ids": list(task.metadata.get("execution_task_ids", [task.id])), "org_version": task.metadata.get("org_version", 1), "runtime_topology_version": task.metadata.get("runtime_topology_version", 1), @@ -9646,6 +9778,67 @@ class OPCEngine: response = await self.message_bus.process_single(message) return response.content if response else "No response generated after resume." + async def _release_work_item_human_wait(self, task: Task, *, reason: str) -> bool: + """Push a work item parked in ``awaiting_human`` back to ``ready``. + + This is the state-machine half of resuming an answered human wait: the + phase moves through the legal ``AWAITING_HUMAN → READY`` recovery exit + and any stale claim is released so the company dispatcher can re-claim + the item on its next pass. Returns True when a phase write happened. + """ + if not self.store or not hasattr(self.store, "update_delegation_work_item"): + return False + work_item_id = linked_work_item_id_for_task(task) + if not work_item_id: + return False + try: + work_item = await self.store.get_delegation_work_item(work_item_id) + except Exception: + logger.opt(exception=True).debug( + "release_work_item_human_wait: work item load failed for task {}", task.id + ) + return False + if work_item is None or getattr(work_item, "phase", None) != Phase.AWAITING_HUMAN: + return False + metadata = dict(getattr(work_item, "metadata", {}) or {}) + metadata_unset: list[str] = [] + if str(metadata.get("dispatch_hold", "") or "").strip() == "company_runtime_suspended": + metadata_unset = ["dispatch_hold", "suspended_at", "suspend_reason", "suspended_phase"] + try: + await self.store.update_delegation_work_item( + work_item_id, + phase=Phase.READY, + blocked_reason="", + metadata_updates={ + "human_wait_released_at": datetime.now().isoformat(), + "human_wait_release_reason": reason, + "claimed_by_role_session_id": "", + "claimed_task_id": "", + }, + metadata_unset=metadata_unset or None, + claimed_by_role_runtime_session_id="", + claimed_by_seat_id="", + ) + except InvalidPhaseTransition: + logger.opt(exception=True).warning( + "release_work_item_human_wait: phase transition rejected for work item {}", + work_item_id, + ) + return False + except Exception: + logger.opt(exception=True).warning( + "release_work_item_human_wait: phase write failed for work item {}", + work_item_id, + ) + return False + logger.info( + "Released human wait on work item {} (task {}, reason={}): awaiting_human -> ready", + work_item_id, + task.id, + reason, + ) + return True + async def _resume_task_checkpoint(self, checkpoint: ExecutionCheckpoint, user_reply: str) -> str: assert self.store checkpoint = await self._ensure_checkpoint_runtime_v2_payload(checkpoint) @@ -9704,13 +9897,38 @@ class OPCEngine: execution_mode = ExecutionMode(raw_execution_mode) except ValueError: execution_mode = ExecutionMode.SINGLE_AGENT - if execution_mode == ExecutionMode.COMPANY_MODE: + # Work-item runtime tasks must resume through the delegation state + # machine, never through a detached single-agent re-run: the recorded + # execution_mode is volatile task metadata and has been observed to + # degrade to task_mode after a first resume, which detaches the re-run + # from the work item and leaves it parked in awaiting_human forever. + is_work_item_task = bool( + is_work_item_runtime_metadata(task.metadata) or linked_work_item_id_for_task(task) + ) + if execution_mode == ExecutionMode.COMPANY_MODE or is_work_item_task: + if is_work_item_task: + await self._release_work_item_human_wait(task, reason="approval_resume") # 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) 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)) + if is_work_item_task: + parent_session_id = str( + getattr(task, "parent_session_id", "") + or task.metadata.get("parent_session_id", "") + or checkpoint.session_id + or "" + ).strip() + snapshot = await self._load_company_runtime_snapshot(parent_session_id) + if snapshot is not None: + snapshot_plan, _snapshot_tasks = snapshot + logger.info( + f"Resuming company-mode checkpoint {checkpoint.checkpoint_id} via runtime " + f"snapshot for parent session {parent_session_id}" + ) + return await self._execute_company_mode(tasks, snapshot_plan) logger.info( f"Resuming company-mode checkpoint {checkpoint.checkpoint_id} without a runtime plan; " f"re-running the paused task {task.id} directly" diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 48a000c..8b69dc4 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -146,6 +146,13 @@ from opc.layer4_tools.output_budget import clip_text from opc.llm.retry import LLMRetryError, call_llm_json_with_retry +# Maximum consecutive idle dispatcher ticks (5s each) tolerated while every +# active task waits on a human but at least one waiter has no pending +# checkpoint on record yet (e.g. a park write racing this snapshot). Once +# exhausted the turn exits with a parked summary instead of spinning forever. +_HUMAN_WAIT_MAX_STALL_TICKS = 24 + + def review_work_item_id_for_attempt(worker_work_item_id: str, attempt: int) -> str: """Compute a per-attempt review work-item ID for a given worker. @@ -4305,12 +4312,43 @@ class CompanyWorkItemExecutor: if t.status in {TaskStatus.AWAITING_HUMAN, TaskStatus.AWAITING_MANAGER_REVIEW, TaskStatus.AWAITING_REVIEW} ] if human_waiting: + # Convergent exit: nothing is in flight, nothing is + # claimable, and every remaining active task waits on + # a human. The wait is resolved through a separate + # engine turn (checkpoint reply → phase ready → + # re-dispatch), never inside this loop, so polling + # here can only spin forever while the caller's turn + # hangs and its claims block the resuming turn. + pending_task_ids = await self._pending_checkpoint_task_ids( + str(human_waiting[0].project_id or "default") + ) + unparked = [t for t in human_waiting if t.id not in pending_task_ids] + self._stall_counter += 1 + if not unparked or self._stall_counter >= _HUMAN_WAIT_MAX_STALL_TICKS: + if unparked: + logger.warning( + "_execute_multi_team_org: exiting after {} stalled ticks with {} " + "human-waiting task(s) lacking a pending checkpoint: {}", + self._stall_counter, + len(unparked), + [t.id for t in unparked], + ) + summary = self._summarize_human_parked_exit(tasks, human_waiting) + await self._emit_progress( + "[Company] runtime turn parked: " + f"{len(human_waiting)} work item(s) awaiting human input; " + "answer the pending approval/review card(s) to continue." + ) + return summary await asyncio.sleep(5) continue for task in active_tasks: if task.status == TaskStatus.AWAITING_PEER: await self._save_peer_checkpoint(task) break + self._stall_counter = 0 + else: + self._stall_counter = 0 # Wait on: (a) any active work item completing, (b) a # dispatcher wake signaled by a delegation tool, or (c) a # short poll tick for external/DB-driven state changes. @@ -13271,6 +13309,43 @@ class CompanyWorkItemExecutor: return True return False + async def _pending_checkpoint_task_ids(self, project_id: str) -> set[str]: + """Task ids referenced by pending execution checkpoints for the project.""" + get_pending = getattr(self.store, "get_pending_checkpoints", None) + if not callable(get_pending) or not self._store_is_ready(self.store): + return set() + try: + rows = await get_pending(project_id=project_id) + except Exception: + logger.opt(exception=True).debug( + "_pending_checkpoint_task_ids: pending checkpoint load failed" + ) + return set() + task_ids: set[str] = set() + for row in rows or []: + payload = dict(getattr(row, "payload", {}) or {}) + task_id = str( + getattr(row, "task_id", "") + or payload.get("waiting_task_id", "") + or payload.get("task_id", "") + or "" + ).strip() + if task_id: + task_ids.add(task_id) + return task_ids + + def _summarize_human_parked_exit(self, tasks: list[Task], human_waiting: list[Task]) -> str: + lines = [ + "## Organization Runtime Parked", + "All remaining work items are waiting on human input. " + "Answer the pending approval/review card(s) and the run will continue from where it stopped.", + "", + ] + for task in sorted(human_waiting, key=lambda item: (item.created_at, item.id)): + status = str(task.status.value if isinstance(task.status, TaskStatus) else task.status) + lines.append(f"- {task.title}: {status}") + return "\n".join(lines) + def _summarize_multi_team_org_results(self, tasks: list[Task]) -> str: if not tasks: return "No organization runtime tasks were found."