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