From 14ee8806dec116804b8f88fbfcfd69b290219893 Mon Sep 17 00:00:00 2001 From: LZH-YS1998 Date: Tue, 28 Jul 2026 10:55:55 +0800 Subject: [PATCH] fix(engine): single live dispatcher per run + unified approval decision channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OBS-4: checkpoint answers for a run whose dispatcher is still live no longer re-enter _execute_company_mode (re-entry reset live claim registries and the attempt ledger stamped in-flight cards interrupted until the streak limit killed them). The executor keeps a _live_run_dispatchers refcount; task and peer checkpoint resumes deliver the input in place and wake the dispatcher. Runs without a live dispatcher keep the original re-entry resume semantics. OBS-7: approval decisions expressed through the chat/checkpoint route now reach the approval engine instead of being parsed as plain task input (which _ask_user treated as deny, re-escalating until the card died). normalize_escalation_reply maps decision tokens/synonyms (never silently denies free text), escalation_context_for_blocked_tool rebuilds the allowlist context from pause_request.permission_context, and _resume_task_checkpoint applies the grant via apply_deferred_escalation_decision — the same engine path as the UI card. Co-Authored-By: Claude Fable 5 --- opc/engine.py | 95 ++++++++++- opc/layer2_organization/approval.py | 74 +++++++++ opc/layer2_organization/company_mode.py | 44 +++++ tests/test_approval_escalation_reply.py | 149 +++++++++++++++++ .../test_checkpoint_answer_live_dispatcher.py | 155 ++++++++++++++++++ 5 files changed, 514 insertions(+), 3 deletions(-) create mode 100644 tests/test_approval_escalation_reply.py create mode 100644 tests/test_checkpoint_answer_live_dispatcher.py diff --git a/opc/engine.py b/opc/engine.py index 1e5a17e..9fd1c88 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -79,7 +79,7 @@ from opc.layer2_organization.org_engine import ( TASK_MODE_COMPANY_ONLY_TOOLS, ) from opc.layer2_organization.task_graph import TaskGraphScheduler -from opc.layer2_organization.approval import ApprovalEngine +from opc.layer2_organization.approval import ApprovalEngine, normalize_escalation_reply from opc.layer2_organization.escalation import EscalationEngine from opc.layer2_organization.communication import CommunicationManager from opc.layer2_organization.collaboration_policy import ownership_guard_violation @@ -11137,9 +11137,63 @@ class OPCEngine: await self.store.resolve_execution_checkpoint(checkpoint.checkpoint_id, status="invalid") return "Could not resume the pending task because it no longer exists." - task.context_snapshot = dict(task.context_snapshot) - task.context_snapshot["user_supplied_input"] = user_reply.strip() + # Unified approval channel (OBS-7): when the pause was a blocked tool + # and the reply is an explicit decision, apply it through the same + # engine the Office UI card click uses. Without this bridge the chat + # route is input-only: the worker retries the still-blocked command + # and re-escalates until the attempt ledger terminalizes the card. pause_request = dict(payload.get("pause_request", {})) + 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() + 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 {} + if isinstance(raw_args, dict): + arguments = dict(raw_args) + candidate = str(permission_context.get("candidate", "") or "").strip() + if candidate and not str(arguments.get("command", "") or "").strip(): + arguments["command"] = candidate + try: + context = self.approval_engine.escalation_context_for_blocked_tool( + task, + tool_name=blocked_tool_name, + arguments=arguments, + ) + outcome = self.approval_engine.apply_deferred_escalation_decision( + decision_token, + context, + ) + except Exception: + logger.opt(exception=True).warning( + "Checkpoint {} approval reply {} failed to apply; degrading to plain input", + checkpoint.checkpoint_id, + decision_token, + ) + else: + approved = bool(outcome.get("approved")) + logger.info( + "Checkpoint {} approval decision applied via reply: {} -> " + "approved={} scope={} patterns={}", + checkpoint.checkpoint_id, + decision_token, + approved, + outcome.get("scope"), + outcome.get("patterns"), + ) + injected_reply = ( + f"Approval decision applied: {decision_token}. The blocked " + f"`{blocked_tool_name}` action is now allowlisted — retry it " + "and continue the task." + if approved + else + f"Approval decision applied: deny. Do not retry the blocked " + f"`{blocked_tool_name}` action; take an alternative approach " + "or report the limitation in your handoff." + ) + task.context_snapshot = dict(task.context_snapshot) + task.context_snapshot["user_supplied_input"] = injected_reply if pause_request: task.context_snapshot["requested_user_input"] = pause_request self._restore_runtime_state_from_checkpoint(task, payload) @@ -11195,6 +11249,26 @@ class OPCEngine: # 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) + # Single-dispatcher-per-run: when the run's dispatcher is live in + # this process, the answer is already persisted (task input + + # released human wait) — wake the loop and let it claim the work. + # Re-entering _execute_company_mode here would overlay a second + # dispatcher on the live run: the resume reset cleared live claim + # registries and the attempt ledger stamped every in-flight card + # as interrupted (project t2 forensics, OBS-4). + run_id = str((task.metadata or {}).get("delegation_run_id", "") or "").strip() + wake = getattr(getattr(self, "company_executor", None), "wake_live_run_dispatcher", None) + if run_id and callable(wake) and wake(run_id): + logger.info( + "Checkpoint {} answered with live dispatcher for run {}; " + "input delivered in place without re-entry", + checkpoint.checkpoint_id, + run_id, + ) + return ( + "Input received. The company runtime is live and will pick " + "it up on its next dispatch tick." + ) 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)) @@ -11272,6 +11346,21 @@ class OPCEngine: if execution_mode == ExecutionMode.COMPANY_MODE.value: # Re-register child tasks for WSHandler dual-routing self._reregister_company_runtime_children(tasks, checkpoint_session_id=checkpoint.session_id) + # Single-dispatcher-per-run: deliver + wake instead of re-entry + # when the run's dispatcher is live (see _resume_task_checkpoint). + run_id = str((task.metadata or {}).get("delegation_run_id", "") or "").strip() + wake = getattr(getattr(self, "company_executor", None), "wake_live_run_dispatcher", None) + if run_id and callable(wake) and wake(run_id): + logger.info( + "Peer checkpoint {} answered with live dispatcher for run {}; " + "input delivered in place without re-entry", + checkpoint.checkpoint_id, + run_id, + ) + return ( + "Input received. The company runtime is live and will pick " + "it up on its next dispatch tick." + ) 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)) diff --git a/opc/layer2_organization/approval.py b/opc/layer2_organization/approval.py index 4da2cab..dc36e10 100644 --- a/opc/layer2_organization/approval.py +++ b/opc/layer2_organization/approval.py @@ -108,6 +108,36 @@ _SHELL_COMMAND_PREFIX_ARITY = { } +_ESCALATION_DECISION_TOKENS: frozenset[str] = frozenset({ + "approve_once", + "approve_session", + "always_project", + "always_global", + "deny", +}) + +_ESCALATION_APPROVE_SYNONYMS: frozenset[str] = frozenset({ + "approve", "approved", "yes", "y", "ok", "allow", "同意", "批准", "允许", +}) + +_ESCALATION_DENY_SYNONYMS: frozenset[str] = frozenset({ + "no", "n", "denied", "reject", "rejected", "拒绝", "不允许", +}) + + +def normalize_escalation_reply(reply: str) -> str: + """Map a human reply to an approval decision token; ``""`` when the text + is not a decision (it is then ordinary task input, never a silent deny).""" + text = str(reply or "").strip().lower() + if text in _ESCALATION_DECISION_TOKENS: + return text + if text in _ESCALATION_APPROVE_SYNONYMS: + return "approve_once" + if text in _ESCALATION_DENY_SYNONYMS: + return "deny" + return "" + + class ApprovalEngine: """Bounded-autonomy policy engine.""" @@ -1945,6 +1975,50 @@ class ApprovalEngine: metadata=result_metadata, ) + def escalation_context_for_blocked_tool( + self, + task: Task | None, + *, + tool_name: str, + arguments: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Rebuild the ``approval_context`` an inline escalation would carry + for a runtime-blocked tool. + + A block that outlived its inline wait survives only as checkpoint + payload (``pause_request.permission_context``), which does not carry + the allowlist context an approval card stores. This builder lets any + reply surface (chat, CLI, headless API) route its decision through + ``apply_deferred_escalation_decision`` — the same channel as the + Office UI card click — instead of degrading to plain task input. + """ + action_kind = "tool" + action_name = str(tool_name or "").strip() + metadata: dict[str, Any] = {"arguments": dict(arguments or {})} + allowlist_enabled = self._allowlist_enabled_for_action(action_kind, metadata) + patterns = ( + self._build_allowlist_patterns( + action_kind=action_kind, + action_name=action_name, + metadata=metadata, + ) + if allowlist_enabled + else [] + ) + return { + "action_kind": action_kind, + "action_name": action_name, + "project_id": str(getattr(task, "project_id", "") or "") if task else "", + "session_scope_id": self._approval_session_scope_id(task), + "allowlist_enabled": allowlist_enabled, + "allowlist_patterns": list(patterns), + "candidates": self._build_allowlist_candidates( + action_kind=action_kind, + action_name=action_name, + metadata=metadata, + ), + } + def apply_deferred_escalation_decision( self, reply: str, diff --git a/opc/layer2_organization/company_mode.py b/opc/layer2_organization/company_mode.py index 6a91c99..83706fa 100644 --- a/opc/layer2_organization/company_mode.py +++ b/opc/layer2_organization/company_mode.py @@ -1423,6 +1423,14 @@ class CompanyWorkItemExecutor: # waits on this Event so children are claimed+spawned without # waiting for the parent turn's gather batch to drain. self._dispatcher_wake = asyncio.Event() + # Single-dispatcher-per-run invariant: refcount of live + # _execute_multi_team_org loops keyed by delegation run id. + # Checkpoint answers consult this via wake_live_run_dispatcher — + # a live run receives input in place instead of a second + # _execute_company_mode entry (re-entry reset live claim + # registries and the attempt ledger stamped every in-flight + # card as interrupted). + self._live_run_dispatchers: dict[str, int] = {} if communication is not None and getattr(communication, "on_work_items_created", None) is None: communication.on_work_items_created = self._signal_dispatcher_wake # D2: register the wake callback with the phase-transition hook @@ -4580,6 +4588,31 @@ class CompanyWorkItemExecutor: if ownership is not None: ownership.release() + def wake_live_run_dispatcher(self, run_id: str) -> bool: + """Wake the live dispatcher for ``run_id``; False when none is live. + + Callers must have already persisted whatever the dispatcher should + pick up (task status, work-item phase, user input) — the wake is + only a scheduling nudge. This is how checkpoint answers reach a + live run without starting a second ``_execute_company_mode`` over + it (the single-dispatcher-per-run invariant). + """ + clean = str(run_id or "").strip() + if not clean or self._live_run_dispatchers.get(clean, 0) <= 0: + return False + self._signal_dispatcher_wake() + return True + + @staticmethod + def _delegation_run_id_for_tasks(tasks: list[Task]) -> str: + for task in tasks: + run_id = str( + (getattr(task, "metadata", {}) or {}).get("delegation_run_id", "") or "" + ).strip() + if run_id: + return run_id + return "" + async def _execute_multi_team_org( self, plan: CompanyWorkItemRuntimePlan, @@ -4589,9 +4622,20 @@ class CompanyWorkItemExecutor: CompanyExecutorRunState(active_plan=plan, active_tasks=list(tasks)) ) runtime_token = self.runtime.use_state(self.runtime.create_state()) + run_id = self._delegation_run_id_for_tasks(tasks) + if run_id: + self._live_run_dispatchers[run_id] = ( + self._live_run_dispatchers.get(run_id, 0) + 1 + ) try: return await self._execute_multi_team_org_scoped(plan, tasks) finally: + if run_id: + remaining = self._live_run_dispatchers.get(run_id, 0) - 1 + if remaining > 0: + self._live_run_dispatchers[run_id] = remaining + else: + self._live_run_dispatchers.pop(run_id, None) self.runtime.reset_state(runtime_token) self._reset_run_state(run_token) diff --git a/tests/test_approval_escalation_reply.py b/tests/test_approval_escalation_reply.py new file mode 100644 index 0000000..35fc7a7 --- /dev/null +++ b/tests/test_approval_escalation_reply.py @@ -0,0 +1,149 @@ +"""Regression: the unified approval decision channel (OBS-7). + +An approval escalation that outlives its inline wait survives as a +``task_user_input`` checkpoint whose payload carries the runtime block's +``permission_context``. A reply that expresses a decision must reach the +approval engine — ``normalize_escalation_reply`` maps human phrasing to a +decision token, ``escalation_context_for_blocked_tool`` rebuilds the +allowlist context, and ``apply_deferred_escalation_decision`` persists the +grant exactly like the Office UI card click. Non-decision text stays plain +task input (never a silent deny). +""" +from __future__ import annotations + +import unittest + +from opc.core.config import AutonomyConfig +from opc.core.models import CompanyMemberSession, Task +from opc.layer2_organization.approval import ( + ApprovalEngine, + normalize_escalation_reply, +) +from opc.layer2_organization.company_mode import CompanyWorkItemExecutor + + +class _PreferencesStub: + def get_autonomy_preferences(self, project_id=None): + _ = project_id + return {"learned_actions": {}} + + def record_autonomy_feedback(self, **kwargs): + _ = kwargs + + +class _StoreStub: + async def record_approval(self, **kwargs): + _ = kwargs + + +class _MemoryStub: + def append_autonomy_event(self, event, project=False): + _ = (event, project) + + +def _engine() -> ApprovalEngine: + return ApprovalEngine( + llm=object(), + store=_StoreStub(), + preferences=_PreferencesStub(), + memory=_MemoryStub(), + escalation=None, + config=AutonomyConfig(), + ) + + +class NormalizeEscalationReplyTests(unittest.TestCase): + def test_exact_tokens_pass_through(self) -> None: + for token in ("approve_once", "approve_session", "always_project", + "always_global", "deny"): + self.assertEqual(normalize_escalation_reply(token), token) + + def test_approve_synonyms_map_to_approve_once(self) -> None: + for text in ("approve", "Yes", " y ", "同意", "允许"): + self.assertEqual(normalize_escalation_reply(text), "approve_once") + + def test_deny_synonyms_map_to_deny(self) -> None: + for text in ("no", "Reject", "拒绝"): + self.assertEqual(normalize_escalation_reply(text), "deny") + + def test_plain_content_is_not_a_decision(self) -> None: + for text in ("", "please use conda instead", "proceed with best judgment"): + self.assertEqual(normalize_escalation_reply(text), "") + + +class EscalationContextForBlockedToolTests(unittest.TestCase): + def test_shell_exec_context_carries_command_patterns(self) -> None: + engine = _engine() + task = Task(id="t1", title="t", project_id="p1", session_id="s1") + context = engine.escalation_context_for_blocked_tool( + task, + tool_name="shell_exec", + arguments={"command": "pip install pandas"}, + ) + self.assertEqual(context["action_kind"], "tool") + self.assertEqual(context["action_name"], "shell_exec") + self.assertEqual(context["project_id"], "p1") + self.assertTrue(context["allowlist_enabled"]) + self.assertTrue(context["candidates"]) + self.assertTrue( + any("pip" in str(item) for item in context["candidates"]), + context["candidates"], + ) + + def test_decision_applies_through_deferred_channel(self) -> None: + engine = _engine() + task = Task(id="t2", title="t", project_id="p1", session_id="s1") + task.metadata["session_scope_id"] = "scope-1" + context = engine.escalation_context_for_blocked_tool( + task, + tool_name="shell_exec", + arguments={"command": "pip install pandas"}, + ) + outcome = engine.apply_deferred_escalation_decision("approve_session", context) + self.assertTrue(outcome.get("approved")) + deny = engine.apply_deferred_escalation_decision("deny", context) + self.assertFalse(deny.get("approved")) + + +class LiveRunDispatcherRegistryTests(unittest.TestCase): + """OBS-4: checkpoint answers wake a live dispatcher instead of re-entry.""" + + def _executor(self) -> CompanyWorkItemExecutor: + import asyncio + + executor = CompanyWorkItemExecutor.__new__(CompanyWorkItemExecutor) + executor._live_run_dispatchers = {} + executor._dispatcher_wake = asyncio.Event() + return executor + + def test_wake_returns_false_when_no_live_dispatcher(self) -> None: + executor = self._executor() + self.assertFalse(executor.wake_live_run_dispatcher("run-1")) + self.assertFalse(executor._dispatcher_wake.is_set()) + + def test_wake_signals_live_dispatcher(self) -> None: + executor = self._executor() + executor._live_run_dispatchers["run-1"] = 1 + self.assertTrue(executor.wake_live_run_dispatcher("run-1")) + self.assertTrue(executor._dispatcher_wake.is_set()) + + def test_run_id_extraction_prefers_first_tagged_task(self) -> None: + t1 = Task(id="a", title="a", project_id="p", session_id="s") + t2 = Task(id="b", title="b", project_id="p", session_id="s") + t2.metadata["delegation_run_id"] = "run-9" + self.assertEqual( + CompanyWorkItemExecutor._delegation_run_id_for_tasks([t1, t2]), + "run-9", + ) + self.assertEqual( + CompanyWorkItemExecutor._delegation_run_id_for_tasks([t1]), + "", + ) + + def test_member_session_import_smoke(self) -> None: + # Guard against accidental import regressions in the test module. + self.assertTrue(CompanyMemberSession) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_checkpoint_answer_live_dispatcher.py b/tests/test_checkpoint_answer_live_dispatcher.py new file mode 100644 index 0000000..702486d --- /dev/null +++ b/tests/test_checkpoint_answer_live_dispatcher.py @@ -0,0 +1,155 @@ +"""Engine-integration regression for OBS-4 + OBS-7. + +A ``task_user_input`` answer for a run whose dispatcher is live must: + 1. apply an explicit approval decision through the approval engine + (OBS-7 — the chat route was input-only, so the blocked tool + re-escalated until the attempt ledger killed the card), and + 2. deliver the input in place and wake the live dispatcher instead of + re-entering ``_execute_company_mode`` (OBS-4 — re-entry reset live + claim registries and the ledger stamped in-flight cards interrupted). + +When no dispatcher is live the resume must fall through to the legacy +re-entry path unchanged. +""" +from __future__ import annotations + +import asyncio +import unittest +from datetime import datetime +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import AsyncMock + +from opc.core.config import AutonomyConfig +from opc.core.models import ExecutionCheckpoint, Task, TaskStatus +from opc.database.store import OPCStore +from opc.engine import OPCEngine +from opc.layer2_organization.approval import ApprovalEngine +from opc.layer2_organization.company_mode import CompanyWorkItemExecutor +from opc.layer2_organization.work_item_links import set_linked_work_item_id + + +class _PreferencesStub: + def get_autonomy_preferences(self, project_id=None): + _ = project_id + return {"learned_actions": {}} + + def record_autonomy_feedback(self, **kwargs): + _ = kwargs + + +class _StoreStub: + async def record_approval(self, **kwargs): + _ = kwargs + + +class _MemoryStub: + def append_autonomy_event(self, event, project=False): + _ = (event, project) + + +def _approval_engine() -> ApprovalEngine: + return ApprovalEngine( + llm=object(), + store=_StoreStub(), + preferences=_PreferencesStub(), + memory=_MemoryStub(), + escalation=None, + config=AutonomyConfig(), + ) + + +class CheckpointAnswerLiveDispatcherTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self._tmp = TemporaryDirectory() + self.store = OPCStore(Path(self._tmp.name) / "tasks.db") + await self.store.initialize() + self.engine = OPCEngine(project_id="p") + self.engine.store = self.store + self.engine.approval_engine = _approval_engine() + self.executor = CompanyWorkItemExecutor.__new__(CompanyWorkItemExecutor) + self.executor._live_run_dispatchers = {} + self.executor._dispatcher_wake = asyncio.Event() + self.engine.company_executor = self.executor + self.engine._execute_company_mode = AsyncMock(return_value="re-entered") + self.engine._execute_single_agent = AsyncMock(return_value="single-agent") + + async def asyncTearDown(self) -> None: + await self.store.close() + self._tmp.cleanup() + + async def _seed(self) -> ExecutionCheckpoint: + task = Task( + id="task-1", + title="blocked worker", + project_id="p", + session_id="s", + status=TaskStatus.AWAITING_HUMAN, + metadata={ + "delegation_run_id": "run-1", + "execution_mode": "company_mode", + "work_item_runtime": True, + }, + ) + set_linked_work_item_id(task, "wi-1") + await self.store.save_task(task) + checkpoint = ExecutionCheckpoint( + checkpoint_id="ckpt-1", + project_id="p", + session_id="s", + checkpoint_type="task_user_input", + task_id="task-1", + status="pending", + payload={ + "task_id": "task-1", + "session_id": "s", + "execution_mode": "company_mode", + "task_ids": ["task-1"], + "prompt": "Tool execution blocked by autonomy policy", + "pause_request": { + "requires_user_input": True, + "permission_context": { + "tool_name": "shell_exec", + "candidate": "pip install pandas", + "resolution": "ask", + }, + }, + }, + created_at=datetime.now(), + ) + await self.store.save_execution_checkpoint(checkpoint) + return checkpoint + + async def test_live_dispatcher_gets_wake_and_approval_applies(self) -> None: + checkpoint = await self._seed() + self.executor._live_run_dispatchers["run-1"] = 1 + + reply = await self.engine._resume_task_checkpoint(checkpoint, "approve_session") + + self.assertIn("live", reply) + self.assertTrue(self.executor._dispatcher_wake.is_set()) + self.engine._execute_company_mode.assert_not_awaited() + self.engine._execute_single_agent.assert_not_awaited() + 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) + self.assertEqual(saved.status, TaskStatus.PENDING) + + async def test_no_live_dispatcher_falls_through_to_reentry_path(self) -> None: + checkpoint = await self._seed() + + reply = await self.engine._resume_task_checkpoint(checkpoint, "please continue") + + self.assertFalse(self.executor._dispatcher_wake.is_set()) + self.assertEqual(reply, "single-agent") + self.engine._execute_single_agent.assert_awaited() + saved = await self.store.get_task("task-1") + self.assertEqual( + saved.context_snapshot.get("user_supplied_input"), + "please continue", + ) + + +if __name__ == "__main__": + unittest.main()