fix(engine): single live dispatcher per run + unified approval decision channel

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 <noreply@anthropic.com>

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))

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,

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)
 

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()

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()