diff --git a/opc/engine.py b/opc/engine.py index 65b5f4d..12d0c99 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -7,6 +7,7 @@ import copy import hashlib import inspect import json +import os import re import shutil import time @@ -639,9 +640,21 @@ class OPCEngine: self.org_engine = OrgEngine(self.config, self.opc_home, store=self.store) self.talent_market = TalentMarket(self.opc_home, self.config) self.task_scheduler = TaskGraphScheduler(self.store, self.event_bus) + escalation_timeout_seconds = self.config.system.escalation_timeout_seconds + # Test/ops override: lets a harness shrink the inline approval wait + # (e.g. to seconds) so the late-click park/resume cycle can be + # exercised without waiting out the production timeout. + raw_escalation_timeout = str(os.environ.get("OPC_ESCALATION_TIMEOUT_SECONDS", "") or "").strip() + if raw_escalation_timeout: + try: + escalation_timeout_seconds = max(1, int(raw_escalation_timeout)) + except ValueError: + logger.warning( + f"Ignoring invalid OPC_ESCALATION_TIMEOUT_SECONDS={raw_escalation_timeout!r}" + ) self.escalation = EscalationEngine( self.event_bus, - timeout_seconds=self.config.system.escalation_timeout_seconds, + timeout_seconds=escalation_timeout_seconds, user_reply_callback=self.on_escalation, ) self.communication = CommunicationManager(self.store, self.event_bus, self.llm, self.org_engine) @@ -4303,9 +4316,17 @@ class OPCEngine: "created_at": latest_compaction.created_at.isoformat(), }) - permission_requests: list[dict[str, Any]] = [] + # Preserve any permission_requests already recorded on the payload — + # they carry the blocked call's tool_args, which are the only source + # of the command text for a late allowlist grant. Rebuilding from the + # legacy approval/pause_request keys is a fallback, not a replacement. + permission_requests: list[dict[str, Any]] = [ + dict(item) + for item in list(payload_data.get("permission_requests", []) or []) + if isinstance(item, dict) + ] approval = dict(payload_data.get("approval", {}) or {}) - if approval: + if approval and not permission_requests: permission_requests.append({ "tool_name": str(payload_data.get("tool_name", "") or ""), "resolution": "ask", @@ -11357,10 +11378,36 @@ class OPCEngine: injected_reply = user_reply.strip() permission_context = dict(pause_request.get("permission_context", {}) or {}) blocked_tool_name = str(permission_context.get("tool_name", "") or "").strip() + blocked_tool_args: dict[str, Any] = {} + if not blocked_tool_name: + # Company runtime parks persist the blocked call as a + # permission_requests entry (runtime_v2 artifacts), not as + # pause_request.permission_context. Without this fallback a late + # approval reply resumes the task but records no allowlist grant, + # so the identical command re-blocks and re-parks on a fresh card + # every cycle (the project-0012 approve treadmill). + for request in reversed(list(payload.get("permission_requests", []) or [])): + if not isinstance(request, dict): + continue + if str(request.get("resolution", "") or "").strip() != "ask": + continue + candidate_tool = str(request.get("tool_name", "") or "").strip() + if not candidate_tool: + continue + blocked_tool_name = candidate_tool + raw_request_args = request.get("tool_args") + if isinstance(raw_request_args, dict): + blocked_tool_args = dict(raw_request_args) + break decision_token = normalize_escalation_reply(user_reply) if blocked_tool_name and decision_token and self.approval_engine is not None: arguments: dict[str, Any] = {} - raw_args = payload.get("tool_args") or pause_request.get("tool_args") or {} + raw_args = ( + payload.get("tool_args") + or pause_request.get("tool_args") + or blocked_tool_args + or {} + ) if isinstance(raw_args, dict): arguments = dict(raw_args) candidate = str(permission_context.get("candidate", "") or "").strip() diff --git a/opc/layer3_agent/runtime_v2/runtime.py b/opc/layer3_agent/runtime_v2/runtime.py index fe8e6e8..5df1989 100644 --- a/opc/layer3_agent/runtime_v2/runtime.py +++ b/opc/layer3_agent/runtime_v2/runtime.py @@ -4029,8 +4029,15 @@ class NativeRuntimeV2: resolution_value = getattr(resolution, "value", str(resolution)) if resolution_value not in {"ask", "deny"}: continue + raw_arguments = call.get("arguments") requests.append({ "tool_name": str(call.get("function", "") or ""), + # The blocked call's arguments must survive into the park + # checkpoint: a late approval reply rebuilds the allowlist + # context from them, and without the command text no grant can + # be recorded — the task resumes, retries, re-blocks, and + # re-parks on an identical card forever. + "tool_args": dict(raw_arguments) if isinstance(raw_arguments, dict) else {}, "resolution": resolution_value, "scope": getattr(getattr(decision, "scope", None), "value", str(getattr(decision, "scope", ""))), "risk_level": getattr(getattr(decision, "risk_level", None), "value", str(getattr(decision, "risk_level", ""))), diff --git a/opc/plugins/office_ui/ws_handler.py b/opc/plugins/office_ui/ws_handler.py index 368e866..b3faa54 100644 --- a/opc/plugins/office_ui/ws_handler.py +++ b/opc/plugins/office_ui/ws_handler.py @@ -8525,6 +8525,143 @@ class WSHandler: logger.opt(exception=True).debug("failed to persist checkpoint card terminal state") return None + _LOCK_FREE_CHECKPOINT_ANSWER_TYPES = frozenset({ + "task_user_input", + "company_work_item_gate", + }) + + async def _try_lock_free_parked_checkpoint_answer( + self, + *, + task_id: str, + content: str, + session_id: str | None, + message_metadata: dict[str, Any] | None, + user_message_id: str | None, + user_message_created_at: float | None, + engine: Any, + pid: str, + channel_id: str, + session_exec_mode: str, + session_company_profile: str | None, + session_org_id: str, + attachment_refs: list[dict] | None, + ) -> bool: + """Answer a pending park checkpoint without taking the per-task turn lock. + + A company goal turn can hold the per-task lock for hours while its live + dispatcher waits on AWAITING_HUMAN approval cards. The card answers are + themselves session messages, so they queue behind that same lock — a + circular wait: dispatcher waits for the answer, the answer waits for the + lock, the lock waits for the dispatcher (project-0012 late-approval + wedge). When the reply explicitly targets a pending park checkpoint and + the lock is currently held, deliver it straight through the engine's + checkpoint-resume channel instead: with a live dispatcher the engine + only persists the input, applies the approval decision, releases the + human wait, and wakes the loop — no second dispatcher, no re-entry. + + Returns True when the reply was fully handled here. + """ + metadata = dict(message_metadata or {}) + checkpoint_id = str(metadata.get("response_to_checkpoint_id", "") or "").strip() + checkpoint_type = str(metadata.get("response_to_checkpoint_type", "") or "").strip() + if not checkpoint_id or checkpoint_type not in self._LOCK_FREE_CHECKPOINT_ANSWER_TYPES: + return False + lock = self._get_task_lock(task_id) + if not lock.locked(): + # No turn in flight: the serialized path works and preserves + # ordering, so keep the existing behavior. + return False + store = getattr(engine, "store", None) + if not self._store_is_ready(store): + return False + getter = getattr(store, "get_pending_checkpoints", None) + if not callable(getter): + return False + try: + pending = await getter(project_id=pid) + except Exception: + logger.opt(exception=True).debug( + "Lock-free checkpoint answer: failed to load pending checkpoints" + ) + return False + checkpoint = next( + ( + item + for item in pending or [] + if str(getattr(item, "checkpoint_id", "") or "").strip() == checkpoint_id + and str(getattr(item, "checkpoint_type", "") or "").strip() + in self._LOCK_FREE_CHECKPOINT_ANSWER_TYPES + ), + None, + ) + if checkpoint is None: + return False + logger.info( + f"Lock-free checkpoint answer: task lock for {task_id} is held by a " + f"live turn; delivering reply to pending checkpoint {checkpoint_id} " + "through the engine resume channel" + ) + try: + engine_mode, company_profile = self._resolve_engine_mode( + session_exec_mode, + session_company_profile, + ) + engine_message_metadata = dict(metadata) + engine_message_metadata.update(_ui_message_identity_metadata( + message_id=user_message_id, + conversation_turn_id=_ui_conversation_turn_id(user_message_id), + created_at=user_message_created_at, + )) + response = await engine.process_message( + content, + project_id=pid, + session_id=session_id, + mode=engine_mode, + org_id=session_org_id or None, + company_profile=company_profile, + origin_task_id=task_id, + attachment_refs=attachment_refs, + message_metadata=engine_message_metadata or None, + ) + updated_checkpoint_msg = await self._mark_checkpoint_card_after_engine_response( + channel_id=channel_id, + project_id=pid, + engine=engine, + message_metadata=engine_message_metadata, + response_message_id=user_message_id, + ) + if updated_checkpoint_msg is not None: + await self.broadcast({"type": "session_message", "payload": updated_checkpoint_msg}) + reply_text = str(response or "").strip() or "Input received." + reply_msg = await self.chat_store.insert_message( + channel_id=channel_id, + sender="assistant", + sender_name="OPC", + content=reply_text, + project_id=pid, + metadata={"type": "system", "checkpoint_answer_lock_free": True}, + ) + await self.broadcast({"type": "session_message", "payload": reply_msg}) + except asyncio.CancelledError: + raise + except Exception as exc: + # Do NOT fall back to the locked path: it would silently queue + # behind the in-flight turn — exactly the wedge this path exists + # to break. Surface the failure so the user can retry. + logger.opt(exception=True).warning( + f"Lock-free checkpoint answer failed for {checkpoint_id}" + ) + helper = await self.chat_store.insert_message( + channel_id=channel_id, + sender="system", + sender_name="OPC", + content=f"Failed to deliver the approval reply: {exc}. Please click the card again.", + project_id=pid, + ) + await self.broadcast({"type": "session_message", "payload": helper}) + return True + async def _process_session_message( self, task_id: str, content: str, *, session_id: str | None = None, @@ -8566,6 +8703,23 @@ class WSHandler: session_org_id = self._resolve_task_org_id(task) session_preferred_agent = self._resolve_task_preferred_agent(task) + if await self._try_lock_free_parked_checkpoint_answer( + task_id=task_id, + content=content, + session_id=session_id, + message_metadata=message_metadata, + user_message_id=user_message_id, + user_message_created_at=user_message_created_at, + engine=engine, + pid=pid, + channel_id=channel_id, + session_exec_mode=session_exec_mode, + session_company_profile=session_company_profile, + session_org_id=session_org_id, + attachment_refs=attachment_refs, + ): + return + # Per-task lock: same session serialized, different sessions concurrent async with self._get_task_lock(task_id): current_task = asyncio.current_task() diff --git a/tests/test_checkpoint_answer_live_dispatcher.py b/tests/test_checkpoint_answer_live_dispatcher.py index 702486d..4d2e9db 100644 --- a/tests/test_checkpoint_answer_live_dispatcher.py +++ b/tests/test_checkpoint_answer_live_dispatcher.py @@ -136,6 +136,68 @@ class CheckpointAnswerLiveDispatcherTests(unittest.IsolatedAsyncioTestCase): self.assertIn("shell_exec", injected) self.assertEqual(saved.status, TaskStatus.PENDING) + async def test_runtime_v2_park_shape_applies_approval_via_permission_requests(self) -> None: + """Company runtime parks carry the blocked call in permission_requests + (empty pause_request). The decision bridge must fall back to that shape + or a late approval resumes without recording any allowlist grant and + the identical command re-parks forever (project-0012 treadmill).""" + checkpoint = await self._seed() + payload = dict(checkpoint.payload) + payload["pause_request"] = {} + payload["permission_requests"] = [ + { + "tool_name": "shell_exec", + "tool_args": {"command": "pip install pandas"}, + "resolution": "ask", + "scope": "once", + "risk_level": "medium", + "rationale": "Command is not in the low-risk allowlist.", + "source": "approval_engine", + } + ] + checkpoint.payload = payload + await self.store.save_execution_checkpoint(checkpoint) + self.executor._live_run_dispatchers["run-1"] = 1 + + reply = await self.engine._resume_task_checkpoint(checkpoint, "approve_session") + + self.assertIn("live", reply) + saved = await self.store.get_task("task-1") + injected = str(saved.context_snapshot.get("user_supplied_input", "")) + self.assertIn("Approval decision applied", injected) + self.assertIn("shell_exec", injected) + + async def test_permission_requests_artifact_preserves_tool_args(self) -> None: + """The runtime park artifact must persist the blocked call's arguments; + they are the only source of the command text for late allowlist grants.""" + from opc.layer3_agent.runtime_v2.runtime import NativeRuntimeV2 + + class _Decision: + resolution = type("R", (), {"value": "ask"})() + scope = type("S", (), {"value": "once"})() + risk_level = type("L", (), {"value": "medium"})() + rationale = "blocked" + source = "approval_engine" + + runtime = object.__new__(NativeRuntimeV2) + requests = NativeRuntimeV2._permission_requests_from_results( + runtime, + [ + { + "permission_decision": _Decision(), + "tool_call": { + "function": "shell_exec", + "arguments": {"command": "curl -sI https://example.com"}, + }, + } + ], + ) + self.assertEqual(len(requests), 1) + self.assertEqual(requests[0]["tool_name"], "shell_exec") + self.assertEqual( + requests[0]["tool_args"], {"command": "curl -sI https://example.com"} + ) + async def test_no_live_dispatcher_falls_through_to_reentry_path(self) -> None: checkpoint = await self._seed() diff --git a/tests/test_lock_free_checkpoint_answer.py b/tests/test_lock_free_checkpoint_answer.py new file mode 100644 index 0000000..09f0f35 --- /dev/null +++ b/tests/test_lock_free_checkpoint_answer.py @@ -0,0 +1,235 @@ +"""Regression tests for the lock-free parked-checkpoint answer path. + +Project-0012 forensics: a company goal turn holds the per-task session lock +for hours while its live dispatcher waits on AWAITING_HUMAN approval cards. +The card answers are session messages, so they queued behind that same lock — +a circular wait (dispatcher -> answer -> lock -> dispatcher) that left the +approval clicks undelivered forever. The fix routes a reply that explicitly +targets a pending park checkpoint through the engine's checkpoint-resume +channel without acquiring the turn lock. +""" +from __future__ import annotations + +import asyncio +import unittest +from types import SimpleNamespace +from typing import Any + +from opc.plugins.office_ui.ws_handler import WSHandler + + +class _ChatStoreStub: + def __init__(self) -> None: + self.inserted: list[dict[str, Any]] = [] + + async def insert_message(self, **kwargs: Any) -> dict[str, Any]: + self.inserted.append(kwargs) + return {"message_id": f"msg-{len(self.inserted)}", **kwargs} + + +class _StoreStub: + def __init__(self, pending: list[Any]) -> None: + self._pending = pending + + async def get_pending_checkpoints(self, project_id: str = "default") -> list[Any]: + return list(self._pending) + + +class _EngineStub: + def __init__(self, store: Any, *, reply: str = "Input received.", error: Exception | None = None) -> None: + self.store = store + self.reply = reply + self.error = error + self.calls: list[dict[str, Any]] = [] + + async def process_message(self, content: str, **kwargs: Any) -> str: + self.calls.append({"content": content, **kwargs}) + if self.error is not None: + raise self.error + return self.reply + + +def _pending_checkpoint(checkpoint_id: str, checkpoint_type: str = "task_user_input") -> Any: + return SimpleNamespace( + checkpoint_id=checkpoint_id, + checkpoint_type=checkpoint_type, + status="pending", + ) + + +def _make_handler(engine: _EngineStub) -> WSHandler: + handler = object.__new__(WSHandler) + handler._task_locks = {} + handler._task_lock_holders = {} + handler.chat_store = _ChatStoreStub() + handler._store_is_ready = lambda store: store is not None + handler.broadcast = _async_noop + handler._mark_checkpoint_card_after_engine_response = _async_none_kwargs + return handler + + +async def _async_noop(*args: Any, **kwargs: Any) -> None: + return None + + +async def _async_none_kwargs(**kwargs: Any) -> None: + return None + + +def _answer_kwargs(**overrides: Any) -> dict[str, Any]: + kwargs: dict[str, Any] = { + "task_id": "chat-task", + "content": "Approval decision: approve_session. Re-run it and continue the task.", + "session_id": "session-1", + "message_metadata": { + "response_to_checkpoint_id": "ckpt-park", + "response_to_checkpoint_type": "task_user_input", + }, + "user_message_id": "ui-msg-1", + "user_message_created_at": None, + "pid": "0012", + "channel_id": "session:chat-task", + "session_exec_mode": "company", + "session_company_profile": "corporate", + "session_org_id": "", + "attachment_refs": None, + } + kwargs.update(overrides) + return kwargs + + +class LockFreeCheckpointAnswerTests(unittest.IsolatedAsyncioTestCase): + async def _hold_lock(self, handler: WSHandler, task_id: str) -> asyncio.Task: + lock = handler._get_task_lock(task_id) + acquired = asyncio.Event() + release = asyncio.Event() + + async def _holder() -> None: + async with lock: + acquired.set() + await release.wait() + + holder = asyncio.create_task(_holder()) + await acquired.wait() + handler._task_lock_holders[task_id] = holder + holder.release_event = release # type: ignore[attr-defined] + return holder + + async def test_lock_held_delivers_through_resume_channel(self) -> None: + engine = _EngineStub( + _StoreStub([_pending_checkpoint("ckpt-park")]), + reply="Input received. The company runtime is live and will pick it up on its next dispatch tick.", + ) + handler = _make_handler(engine) + holder = await self._hold_lock(handler, "chat-task") + try: + handled = await handler._try_lock_free_parked_checkpoint_answer( + engine=engine, **_answer_kwargs() + ) + self.assertTrue(handled) + self.assertEqual(len(engine.calls), 1) + call = engine.calls[0] + self.assertEqual(call["mode"], "company") + self.assertEqual(call["project_id"], "0012") + self.assertEqual( + call["message_metadata"]["response_to_checkpoint_id"], "ckpt-park" + ) + # The turn lock must remain untouched — still held by the live turn. + self.assertTrue(handler._get_task_lock("chat-task").locked()) + # The engine reply is surfaced to the session channel. + replies = [m for m in handler.chat_store.inserted if m.get("sender") == "assistant"] + self.assertEqual(len(replies), 1) + self.assertIn("Input received", replies[0]["content"]) + finally: + holder.release_event.set() # type: ignore[attr-defined] + await holder + + async def test_lock_free_session_keeps_serialized_path(self) -> None: + engine = _EngineStub(_StoreStub([_pending_checkpoint("ckpt-park")])) + handler = _make_handler(engine) + handled = await handler._try_lock_free_parked_checkpoint_answer( + engine=engine, **_answer_kwargs() + ) + self.assertFalse(handled) + self.assertEqual(engine.calls, []) + + async def test_unknown_or_resolved_checkpoint_declines(self) -> None: + engine = _EngineStub(_StoreStub([])) + handler = _make_handler(engine) + holder = await self._hold_lock(handler, "chat-task") + try: + handled = await handler._try_lock_free_parked_checkpoint_answer( + engine=engine, **_answer_kwargs() + ) + self.assertFalse(handled) + self.assertEqual(engine.calls, []) + finally: + holder.release_event.set() # type: ignore[attr-defined] + await holder + + async def test_non_park_checkpoint_type_declines(self) -> None: + engine = _EngineStub( + _StoreStub([_pending_checkpoint("ckpt-park", "company_delivery_feedback")]) + ) + handler = _make_handler(engine) + holder = await self._hold_lock(handler, "chat-task") + try: + handled = await handler._try_lock_free_parked_checkpoint_answer( + engine=engine, + **_answer_kwargs( + message_metadata={ + "response_to_checkpoint_id": "ckpt-park", + "response_to_checkpoint_type": "company_delivery_feedback", + } + ), + ) + self.assertFalse(handled) + self.assertEqual(engine.calls, []) + finally: + holder.release_event.set() # type: ignore[attr-defined] + await holder + + async def test_engine_failure_surfaces_error_without_queueing(self) -> None: + engine = _EngineStub( + _StoreStub([_pending_checkpoint("ckpt-park")]), + error=RuntimeError("resume blew up"), + ) + handler = _make_handler(engine) + holder = await self._hold_lock(handler, "chat-task") + try: + handled = await handler._try_lock_free_parked_checkpoint_answer( + engine=engine, **_answer_kwargs() + ) + # Handled=True: the reply must NOT fall through to the locked path, + # which would silently queue behind the wedged turn again. + self.assertTrue(handled) + errors = [m for m in handler.chat_store.inserted if m.get("sender") == "system"] + self.assertEqual(len(errors), 1) + self.assertIn("resume blew up", errors[0]["content"]) + finally: + holder.release_event.set() # type: ignore[attr-defined] + await holder + + async def test_stale_done_holder_lock_self_heals_and_declines(self) -> None: + engine = _EngineStub(_StoreStub([_pending_checkpoint("ckpt-park")])) + handler = _make_handler(engine) + lock = handler._get_task_lock("chat-task") + await lock.acquire() + + async def _finished() -> None: + return None + + done_holder = asyncio.create_task(_finished()) + await done_holder + handler._task_lock_holders["chat-task"] = done_holder + handled = await handler._try_lock_free_parked_checkpoint_answer( + engine=engine, **_answer_kwargs() + ) + # _get_task_lock replaces the stale lock, so the fresh lock is free and + # the normal serialized path is the right route. + self.assertFalse(handled) + self.assertEqual(engine.calls, []) + + +if __name__ == "__main__": + unittest.main()