From a24cbea4d84b4629c8c0056bb488c777b970284f Mon Sep 17 00:00:00 2001 From: LZH-YS1998 Date: Tue, 7 Jul 2026 22:38:33 +0800 Subject: [PATCH] fix: never let a parked approval prompt implicitly consume plain chat A task_user_input checkpoint created by an approval escalation (payload runtime_v2.permission_requests non-empty) used to capture the next plain chat message in the session as its answer. With deferred approval cards now staying pending indefinitely (c9018000), that implicit capture would swallow every later conversation message into the approval reply. Permission prompts are decided through their approval card, whose reply always carries an explicit response_to_checkpoint_id; a plain message now falls through to normal turn processing and the card stays pending and clickable. Waits without a permission request (agent asked the user a question) keep accepting typed answers unchanged. Co-Authored-By: Claude Fable 5 --- opc/engine.py | 25 ++++++++ tests/test_stale_checkpoint_selfheal.py | 77 +++++++++++++++++++++++++ 2 files changed, 102 insertions(+) diff --git a/opc/engine.py b/opc/engine.py index b8a70b8..ab97c7b 100644 --- a/opc/engine.py +++ b/opc/engine.py @@ -8979,6 +8979,24 @@ class OPCEngine: f"Superseded {len(superseded)} stale task-wait checkpoint(s) for task {task_id} ({reason})" ) + @staticmethod + def _checkpoint_awaits_approval_decision(checkpoint: ExecutionCheckpoint) -> bool: + """Whether a parked task-wait checkpoint is waiting on a permission decision. + + Approval escalations park the task with the pending permission request + recorded under ``payload.runtime_v2.permission_requests``. Those prompts + are decided through their approval card, whose reply always targets the + checkpoint explicitly; free-form chat text is never the decision. + """ + if str(checkpoint.checkpoint_type or "").strip() != "task_user_input": + return False + payload = dict(checkpoint.payload or {}) + runtime_state = payload.get("runtime_v2") + if not isinstance(runtime_state, dict): + return False + requests = runtime_state.get("permission_requests") + return isinstance(requests, list) and len(requests) > 0 + async def _checkpoint_task_still_waiting(self, checkpoint: ExecutionCheckpoint) -> bool: """Whether a task-wait checkpoint still matches a genuinely waiting task. @@ -9535,6 +9553,13 @@ class OPCEngine: checkpoint = await self.get_latest_pending_checkpoint_for_session(session_id) if not checkpoint: return None + if self._checkpoint_awaits_approval_decision(checkpoint): + # A parked permission prompt is answered by its approval card + # (the card reply carries an explicit response_to_checkpoint_id). + # Deferred cards stay pending indefinitely, so a plain chat + # message must not be consumed as the approval answer — let it + # continue as a normal conversation turn instead. + return None metadata_mode = str(dict(reply_metadata or {}).get("mode", "") or "").strip() inferred_mode = requested_mode or metadata_mode if not inferred_mode and self._checkpoint_is_company_scoped(checkpoint.checkpoint_type): diff --git a/tests/test_stale_checkpoint_selfheal.py b/tests/test_stale_checkpoint_selfheal.py index f1d0de8..2da4f24 100644 --- a/tests/test_stale_checkpoint_selfheal.py +++ b/tests/test_stale_checkpoint_selfheal.py @@ -209,6 +209,83 @@ class StaleCheckpointSelfHealTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(result, "company resumed") self.engine._execute_multi_agent.assert_not_awaited() + def _approval_runtime_payload(self) -> dict: + return { + "runtime_session_id": "rt-1", + "permission_requests": [ + { + "tool_name": "shell_exec", + "resolution": "ask", + "scope": "once", + "risk_level": "medium", + "source": "approval_engine", + } + ], + } + + async def test_plain_chat_is_not_consumed_by_parked_approval_checkpoint(self) -> None: + """A live permission prompt is decided via its approval card, never by free chat. + + Deferred approval cards stay pending indefinitely, so an implicit capture + here would swallow every later conversation message into the approval reply. + """ + session_id = str(uuid.uuid4()) + task = await self._save_task(TaskStatus.AWAITING_HUMAN, session_id) + checkpoint = await self._save_wait_checkpoint( + task, + session_id, + runtime_v2=self._approval_runtime_payload(), + ) + + result = await self.engine._maybe_resume_checkpoint( + "顺便问一下，进度怎么样了？", session_id=session_id + ) + + self.assertIsNone(result) # message continues as a normal turn + self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "pending") + + async def test_plain_chat_still_answers_agent_question_checkpoint(self) -> None: + """Waits without a permission request (agent asked the user a question) + keep accepting typed answers.""" + session_id = str(uuid.uuid4()) + task = await self._save_task(TaskStatus.AWAITING_HUMAN, session_id) + checkpoint = await self._save_wait_checkpoint(task, session_id) + + self.engine._execute_single_agent = AsyncMock(return_value="answered") + self.engine._execute_multi_agent = AsyncMock(return_value="") + self.engine._execute_company_mode = AsyncMock(return_value="") + + result = await self.engine._maybe_resume_checkpoint("用蓝色的方案", session_id=session_id) + + self.assertEqual(result, "answered") + self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "resolved") + + async def test_explicit_reply_still_resumes_parked_approval_checkpoint(self) -> None: + """The approval-card click path targets the checkpoint explicitly and must keep working.""" + session_id = str(uuid.uuid4()) + task = await self._save_task(TaskStatus.AWAITING_HUMAN, session_id) + checkpoint = await self._save_wait_checkpoint( + task, + session_id, + runtime_v2=self._approval_runtime_payload(), + ) + + self.engine._execute_single_agent = AsyncMock(return_value="approved and resumed") + self.engine._execute_multi_agent = AsyncMock(return_value="") + self.engine._execute_company_mode = AsyncMock(return_value="") + + result = await self.engine._maybe_resume_checkpoint( + "approve", + session_id=session_id, + reply_metadata={ + "response_to_checkpoint_id": checkpoint.checkpoint_id, + "response_to_checkpoint_type": "task_user_input", + }, + ) + + self.assertEqual(result, "approved and resumed") + self.assertEqual(await self._checkpoint_status(checkpoint.checkpoint_id), "resolved") + if __name__ == "__main__": unittest.main()