From 57610e57a67994c33661df20dffdccc57864d250 Mon Sep 17 00:00:00 2001 From: LZH-YS1998 Date: Thu, 6 Aug 2026 15:37:00 +0800 Subject: [PATCH] fix(ui): keep late-approval fast path cross-channel; surface identity errors in chat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #27 scoped the lock-free parked-checkpoint answer to exact checkpoint task/session equality. Company gate cards are raised by role work-item tasks but answered from the run's anchor chat, whose task id only appears in payload["task_ids"] — the exact-match guard silently disabled the fast path for precisely the answers it exists for and re-opened the project-0012 late-approval lock wedge. Scope by the same linkage set _find_parked_checkpoint_for_deferred_resume uses (checkpoint task/session plus payload waiting_task_id/task_ids), keep rejecting unrelated channels, and keep legacy checkpoints without linkage deliverable. The new fail-closed identity errors in _process_session_message raised out of fire-and-forget background tasks (_track_session), where they are only logged and the user's message silently vanishes. Surface them as a visible system chat error and stop instead of raising. Co-Authored-By: Claude Fable 5 --- opc/plugins/office_ui/ws_handler.py | 110 ++++++++++++++-------- tests/test_lock_free_checkpoint_answer.py | 98 +++++++++++++++++++ tests/test_session_integration.py | 51 +++++++--- 3 files changed, 205 insertions(+), 54 deletions(-) diff --git a/opc/plugins/office_ui/ws_handler.py b/opc/plugins/office_ui/ws_handler.py index 51e9e7a..cee4ab1 100644 --- a/opc/plugins/office_ui/ws_handler.py +++ b/opc/plugins/office_ui/ws_handler.py @@ -8673,29 +8673,35 @@ class WSHandler: ) if checkpoint is None: return False + # Ownership scoping must accept every channel the card legitimately + # reaches, not just the task that raised the checkpoint: company gate + # cards raised by role work items are answered from the run's anchor + # chat, whose task id only appears in payload["task_ids"] (the same + # linkage set _find_parked_checkpoint_for_deferred_resume uses). An + # exact task/session equality check would silently disable this fast + # path for those answers and re-open the late-approval lock wedge. checkpoint_payload = dict(getattr(checkpoint, "payload", {}) or {}) - checkpoint_task_id = str( - getattr(checkpoint, "task_id", "") - or checkpoint_payload.get("waiting_task_id") - or checkpoint_payload.get("task_id") - or "" - ).strip() - checkpoint_session_id = str( - getattr(checkpoint, "session_id", "") - or checkpoint_payload.get("session_id") - or "" - ).strip() - payload_task_ids = { - str(item).strip() - for item in list(checkpoint_payload.get("task_ids", []) or []) - if str(item).strip() + requester_task_id = str(task_id or "").strip() + requester_session_id = str(session_id or "").strip() + linked_task_ids = { + str(checkpoint_payload.get("task_id") or "").strip(), + str(checkpoint_payload.get("waiting_task_id") or "").strip(), + str(getattr(checkpoint, "task_id", "") or "").strip(), } - if ( - not checkpoint_task_id - or checkpoint_task_id != str(task_id or "").strip() - or not checkpoint_session_id - or checkpoint_session_id != str(session_id or "").strip() - or (payload_task_ids and str(task_id or "").strip() not in payload_task_ids) + linked_task_ids.update( + str(item or "").strip() + for item in list(checkpoint_payload.get("task_ids", []) or []) + ) + linked_task_ids.discard("") + linked_session_ids = { + str(getattr(checkpoint, "session_id", "") or "").strip(), + str(checkpoint_payload.get("session_id") or "").strip(), + } + linked_session_ids.discard("") + has_linkage = bool(linked_task_ids or linked_session_ids) + if has_linkage and ( + requester_task_id not in linked_task_ids + and (not requester_session_id or requester_session_id not in linked_session_ids) ): return False logger.info( @@ -8801,26 +8807,50 @@ class WSHandler: from opc.core.models import TaskStatus task = await store.get_task(task_id) if task: - config_task = await self._resolve_session_runtime_config_task( - task_id, - task, - engine=engine, - ) - session_exec_mode, session_company_profile = self._resolve_task_session_config( - config_task - ) - session_org_id = self._resolve_task_org_id(config_task) - session_preferred_agent = self._resolve_task_preferred_agent(config_task) - if ( - session_exec_mode in {"org", "custom"} - and not session_org_id - and is_company_runtime_task(task) - ): - raise ServiceError( - "org_id_required", - "org_id_required", - {"project_id": pid, "task_id": task_id}, + # This coroutine usually runs as a fire-and-forget background + # task (_track_session), where a raised ServiceError is only + # logged and the user's message silently vanishes. Fail closed + # with a visible chat error instead of raising. + try: + config_task = await self._resolve_session_runtime_config_task( + task_id, + task, + engine=engine, ) + session_exec_mode, session_company_profile = self._resolve_task_session_config( + config_task + ) + session_org_id = self._resolve_task_org_id(config_task) + session_preferred_agent = self._resolve_task_preferred_agent(config_task) + if ( + session_exec_mode in {"org", "custom"} + and not session_org_id + and is_company_runtime_task(task) + ): + raise ServiceError( + "org_id_required", + "org_id_required", + {"project_id": pid, "task_id": task_id}, + ) + except ServiceError as exc: + logger.warning( + f"Session message for task {task_id} rejected during " + f"runtime identity resolution: {exc.code}" + ) + try: + msg = await self.chat_store.insert_message( + channel_id=channel_id, + sender="system", + sender_name="OPC", + content=f"Error: {exc.message}", + project_id=pid, + ) + await self.broadcast({"type": "session_message", "payload": msg}) + except Exception: + logger.opt(exception=True).debug( + "failed to surface session identity error" + ) + return if await self._try_lock_free_parked_checkpoint_answer( task_id=task_id, diff --git a/tests/test_lock_free_checkpoint_answer.py b/tests/test_lock_free_checkpoint_answer.py index 82663b9..43508db 100644 --- a/tests/test_lock_free_checkpoint_answer.py +++ b/tests/test_lock_free_checkpoint_answer.py @@ -263,6 +263,104 @@ class LockFreeCheckpointAnswerTests(unittest.IsolatedAsyncioTestCase): self.assertFalse(handled) self.assertEqual(engine.calls, []) + async def test_anchor_channel_answer_for_role_task_gate_is_handled(self) -> None: + # Project-0012 production shape: a company gate checkpoint is raised + # by a role work-item task, but the card is answered from the run's + # anchor chat channel. The anchor task id only appears in + # payload["task_ids"]; exact task/session equality would reject it + # and re-open the late-approval lock wedge. + checkpoint = SimpleNamespace( + checkpoint_id="ckpt-park", + checkpoint_type="company_work_item_gate", + status="pending", + task_id="role-task", + session_id="role-session", + payload={ + "waiting_task_id": "role-task", + "session_id": "role-session", + "task_ids": ["role-task", "chat-task"], + }, + ) + engine = _EngineStub(_StoreStub([checkpoint])) + 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_work_item_gate", + }, + ), + ) + self.assertTrue(handled) + self.assertEqual(len(engine.calls), 1) + finally: + holder.release_event.set() # type: ignore[attr-defined] + await holder + + async def test_legacy_checkpoint_without_linkage_is_still_handled(self) -> None: + # Checkpoints persisted before ownership fields existed carry no + # task/session linkage at all. They must keep the pre-scoping + # behavior (deliver by explicit checkpoint id) instead of silently + # falling back to the wedged serialized path. + checkpoint = SimpleNamespace( + checkpoint_id="ckpt-park", + checkpoint_type="task_user_input", + status="pending", + payload={}, + ) + engine = _EngineStub(_StoreStub([checkpoint])) + 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) + finally: + holder.release_event.set() # type: ignore[attr-defined] + await holder + + +class SessionIdentityErrorSurfacingTests(unittest.IsolatedAsyncioTestCase): + async def test_identity_service_error_surfaces_in_chat_instead_of_raising(self) -> None: + # _process_session_message mostly runs as a fire-and-forget background + # task; a ServiceError escaping it is only logged and the user's + # message silently vanishes. The pre-lock identity resolution must + # surface the failure as a visible chat error and stop. + from opc.plugins.office_ui.services.models import ServiceError + + class _Store: + async def get_task(self, task_id: str) -> Any: + return SimpleNamespace(id=task_id, session_id="sess-1", metadata={}) + + engine = _EngineStub(_Store()) + handler = _make_handler(engine) + handler.engine = engine + handler._session_to_task = {} + handler._exec_mode = "company" + handler._company_profile = "corporate" + handler._task_preferred_agent = "native" + + async def _raise_identity_error(*args: Any, **kwargs: Any) -> Any: + raise ServiceError( + "company_runtime_identity_mismatch", + "Company runtime identity could not be resolved", + {"task_id": "chat-task"}, + ) + + handler._resolve_session_runtime_config_task = _raise_identity_error + + await handler._process_session_message("chat-task", "please continue") + + self.assertEqual(engine.calls, []) + errors = [m for m in handler.chat_store.inserted if m.get("sender") == "system"] + self.assertEqual(len(errors), 1) + self.assertIn("Company runtime identity could not be resolved", errors[0]["content"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_session_integration.py b/tests/test_session_integration.py index 4d2e3b0..e23ca15 100644 --- a/tests/test_session_integration.py +++ b/tests/test_session_integration.py @@ -2061,16 +2061,28 @@ class TestWSHandlerSessionSend(unittest.IsolatedAsyncioTestCase): return_value="wrong-active-org" ) - with self.assertRaises(ServiceError) as context: - await self.handler._process_session_message( - self.task_id, - "approve", - session_id=self.session_id, - ) + # Fail closed without raising: this coroutine usually runs as a + # fire-and-forget background task where an escaping ServiceError is + # only logged and the user's message silently vanishes. The rejection + # must instead surface as a visible chat error. + await self.handler._process_session_message( + self.task_id, + "approve", + session_id=self.session_id, + ) self.engine.process_message.assert_not_called() self.handler.services_context.get_active_saved_org_name.assert_not_awaited() - self.assertEqual(context.exception.code, "company_runtime_identity_mismatch") + errors = [ + msg["payload"].get("content", "") + for msg in self.broadcasts + if msg.get("type") == "session_message" + and str(msg.get("payload", {}).get("sender", "")) == "system" + ] + self.assertTrue( + any("Company runtime identity could not be resolved" in text for text in errors), + errors, + ) async def test_process_session_message_rejects_runtime_org_without_durable_org_id(self) -> None: runtime_session_id = "runtime-org-missing-id-session" @@ -2104,16 +2116,27 @@ class TestWSHandlerSessionSend(unittest.IsolatedAsyncioTestCase): return_value="wrong-active-org" ) - with self.assertRaises(ServiceError) as context: - await self.handler._process_session_message( - role_task.id, - "approve", - session_id=role_task.session_id, - ) + # Same fail-closed-without-raising contract as the identity-mismatch + # case above: reject visibly instead of raising out of a background + # task. + await self.handler._process_session_message( + role_task.id, + "approve", + session_id=role_task.session_id, + ) self.engine.process_message.assert_not_called() self.handler.services_context.get_active_saved_org_name.assert_not_awaited() - self.assertEqual(context.exception.code, "org_id_required") + errors = [ + msg["payload"].get("content", "") + for msg in self.broadcasts + if msg.get("type") == "session_message" + and str(msg.get("payload", {}).get("sender", "")) == "system" + ] + self.assertTrue( + any("org_id_required" in text for text in errors), + errors, + ) async def test_lock_free_process_session_message_uses_durable_org_for_role_task(self) -> None: runtime_session_id = "runtime-org-lock-free-session"