fix: preserve company resume control and agent identity

opc/plugins/office_ui/snapshot_builder.py

@@ -2630,23 +2630,16 @@ async def _build_company_runtime_control_by_task(
         ]
         checkpoint = identity.checkpoint
 
-        def _task_status_value(task: Any) -> str:
-            status = getattr(task, "status", "")
-            if hasattr(status, "value"):
-                return str(status.value or "").strip().lower()
-            return str(status or "").strip().lower().removeprefix("taskstatus.")
-
-        non_terminal_group = [
-            task for task in group
-            if _task_status_value(task) not in {"done", "failed", "cancelled"}
-        ]
         # Persisted RUNNING is only a projection.  The controller-local
         # execution registry is the sole proof that this process still owns a
-        # coroutine capable of monitoring and persisting the run.
+        # coroutine capable of monitoring and persisting the run.  Driver
+        # ownership can deliberately sit on a terminal work-item envelope
+        # while the resumed scheduler is still active, so Task.status must not
+        # filter this lookup.
         runtime_is_live = getattr(engine, "_task_runtime_is_live", None)
         has_running_task = False
         if callable(runtime_is_live):
-            for task in non_terminal_group:
+            for task in group:
                 live_result = runtime_is_live(task)
                 if inspect.isawaitable(live_result):
                     live_result = await live_result
@@ -2657,19 +2650,31 @@ async def _build_company_runtime_control_by_task(
             str((getattr(task, "metadata", {}) or {}).get("company_runtime_stop_state", "") or "").strip()
             in {"suspending", "suspended", "resuming_after_suspending"}
             and bool(str((getattr(task, "metadata", {}) or {}).get("company_runtime_stop_marked_at", "") or "").strip())
-            for task in non_terminal_group
+            for task in group
         )
         any_held_suspended = any(
             str((getattr(task, "metadata", {}) or {}).get("dispatch_hold", "") or "").strip()
             == "company_runtime_suspended"
-            for task in non_terminal_group
+            for task in group
         )
         any_resuming = any(
             str((getattr(task, "metadata", {}) or {}).get("company_runtime_stop_state", "") or "").strip() == "resuming"
-            for task in non_terminal_group
+            for task in group
         )
         checkpoint_status = str(getattr(checkpoint, "status", "") or "").strip().lower() if checkpoint is not None else ""
-        if any_resuming or checkpoint_status == "resuming":
+        # ``resuming`` is the durable checkpoint claim for the whole resumed
+        # execution, not merely a short UI transition.  Once the controller
+        # registry proves that execution ownership exists, the runtime is
+        # stoppable and must project as running until that ownership ends.
+        if checkpoint_status == "pending":
+            state = "suspended"
+        elif checkpoint_status == "resuming" and (
+            any_held_suspended or any_stop_in_progress
+        ):
+            state = "suspending"
+        elif checkpoint_status == "resuming" and has_running_task:
+            state = "running"
+        elif any_resuming or checkpoint_status == "resuming":
             state = "resuming"
         elif checkpoint is not None:
             state = "suspended"
@@ -2678,7 +2683,7 @@ async def _build_company_runtime_control_by_task(
             and any(
                 str((getattr(task, "metadata", {}) or {}).get("company_runtime_stop_state", "") or "").strip()
                 in {"suspending", "resuming_after_suspending"}
-                for task in non_terminal_group
+                for task in group
             )
         ):
             state = "suspending"

opc/plugins/office_ui/tests/test_snapshot_builder_company_kanban.py

@@ -7,7 +7,7 @@ from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
-from opc.core.models import DelegationRun, DelegationWorkItem, Phase
+from opc.core.models import DelegationRun, DelegationWorkItem, ExecutionCheckpoint, Phase
 from opc.database.store import OPCStore
 from opc.plugins.office_ui.snapshot_builder import (
     _build_company_runtime_control_by_task,
@@ -145,6 +145,132 @@ class CompanyKanbanProjectionTests(unittest.IsolatedAsyncioTestCase):
         self.assertFalse(control["parent-task"]["can_stop"])
         engine._task_runtime_is_live.assert_awaited_once_with(parent_task)
 
+    async def test_registry_live_resume_checkpoint_projects_running_and_stoppable(self) -> None:
+        created_at = datetime.now(timezone.utc)
+        parent_task = SimpleNamespace(
+            id="parent-task",
+            session_id="session-root",
+            parent_session_id="",
+            title="Root runtime",
+            status="running",
+            created_at=created_at,
+            metadata={
+                "exec_mode": "company",
+                "company_profile": "corporate",
+            },
+        )
+        terminal_driver_task = SimpleNamespace(
+            id="terminal-driver-task",
+            session_id="driver-session",
+            parent_session_id="session-root",
+            title="Terminal driver envelope",
+            status="done",
+            created_at=created_at,
+            metadata={
+                "mode": "company",
+                "work_item_runtime": True,
+                "work_item_projection_id": "driver",
+            },
+        )
+        checkpoint = ExecutionCheckpoint(
+            checkpoint_id="resume-checkpoint",
+            project_id="proj1",
+            session_id="session-root",
+            checkpoint_type="company_runtime_suspended",
+            status="resuming",
+            task_id="terminal-driver-task",
+            payload={"parent_session_id": "session-root"},
+        )
+        store = MagicMock()
+        store.get_execution_checkpoints = AsyncMock(return_value=[checkpoint])
+        engine = SimpleNamespace(
+            store=store,
+            _task_runtime_is_live=AsyncMock(
+                side_effect=lambda task: task.id == "terminal-driver-task"
+            ),
+        )
+
+        control = await _build_company_runtime_control_by_task(
+            engine,
+            [parent_task, terminal_driver_task],
+            "proj1",
+        )
+
+        self.assertEqual(
+            control["parent-task"]["runtime_control_state"],
+            "running",
+        )
+        self.assertTrue(control["parent-task"]["can_stop"])
+        self.assertFalse(control["parent-task"]["can_resume"])
+
+    async def test_stop_hold_and_pending_checkpoint_override_live_resume_projection(self) -> None:
+        created_at = datetime.now(timezone.utc)
+        parent_task = SimpleNamespace(
+            id="parent-task",
+            session_id="session-root",
+            parent_session_id="",
+            title="Root runtime",
+            status="running",
+            created_at=created_at,
+            metadata={
+                "exec_mode": "company",
+                "company_profile": "corporate",
+            },
+        )
+        child_task = SimpleNamespace(
+            id="child-task",
+            session_id="child-session",
+            parent_session_id="session-root",
+            title="Live child",
+            status="done",
+            created_at=created_at,
+            metadata={
+                "mode": "company",
+                "work_item_runtime": True,
+                "work_item_projection_id": "child",
+                "dispatch_hold": "company_runtime_suspended",
+            },
+        )
+        checkpoint = ExecutionCheckpoint(
+            checkpoint_id="resume-checkpoint",
+            project_id="proj1",
+            session_id="session-root",
+            checkpoint_type="company_runtime_suspended",
+            status="resuming",
+            task_id="child-task",
+            payload={"parent_session_id": "session-root"},
+        )
+        store = MagicMock()
+        store.get_execution_checkpoints = AsyncMock(return_value=[checkpoint])
+        engine = SimpleNamespace(
+            store=store,
+            _task_runtime_is_live=AsyncMock(return_value=True),
+        )
+
+        suspending = await _build_company_runtime_control_by_task(
+            engine,
+            [parent_task, child_task],
+            "proj1",
+        )
+        self.assertEqual(
+            suspending["parent-task"]["runtime_control_state"],
+            "suspending",
+        )
+        self.assertFalse(suspending["parent-task"]["can_stop"])
+
+        checkpoint.status = "pending"
+        suspended = await _build_company_runtime_control_by_task(
+            engine,
+            [parent_task, child_task],
+            "proj1",
+        )
+        self.assertEqual(
+            suspended["parent-task"]["runtime_control_state"],
+            "suspended",
+        )
+        self.assertFalse(suspended["parent-task"]["can_stop"])
+        self.assertTrue(suspended["parent-task"]["can_resume"])
+
     async def test_runtime_control_treats_dispatch_hold_as_suspending_without_checkpoint(self) -> None:
         created_at = datetime.now(timezone.utc)
         parent_task = SimpleNamespace(

opc/plugins/office_ui/ws_handler.py

@@ -7985,7 +7985,13 @@ class WSHandler:
         async with lock:
             try:
                 try:
-                    await self._set_company_runtime_control(target, state="resuming")
+                    await self._set_company_runtime_control(
+                        target,
+                        state="resuming",
+                        checkpoint_id=str(
+                            getattr(checkpoint, "checkpoint_id", "") or ""
+                        ).strip(),
+                    )
                 except Exception:
                     logger.opt(exception=True).debug("failed to broadcast company suspend reply routing state")