diff --git a/opc/layer3_agent/external_broker.py b/opc/layer3_agent/external_broker.py index 3fd7f16..086aab0 100644 --- a/opc/layer3_agent/external_broker.py +++ b/opc/layer3_agent/external_broker.py @@ -72,6 +72,11 @@ class ExternalAgentBroker: _STREAM_READ_SIZE = 8192 _MAX_PATH_HINT_TOKEN_LENGTH = 512 + # Result statuses that prove the provider attempt terminally failed, so + # its session token must not stay pinned. Parks (awaiting_human / + # awaiting_peer / awaiting_manager_review) and cancels keep the token: + # those runs resume the same provider thread once the gate clears. + _SESSION_INVALIDATING_RESULT_STATUSES = frozenset({TaskStatus.FAILED}) _STREAM_SESSION_UPDATE_MIN_SECONDS = 2.0 _STREAM_PROGRESS_MIN_SECONDS = 2.0 _STREAM_TRANSCRIPT_HEAD_LINES = 40 @@ -146,8 +151,16 @@ class ExternalAgentBroker: task: Task, role_session_id: str, token: str, + strict: bool = False, ) -> bool | None: - """Return the latest durable resumability verdict for one provider token.""" + """Return the latest durable resumability verdict for one provider token. + + ``True`` means the newest row for the token finalized resumable, + ``False`` means the token is dead and must be cleared, ``None`` means + there is no durable verdict either way (caller keeps the token). + ``strict`` treats an unfinalized newest row as ``False`` — required + for provider_stream tokens whose run may have crashed mid-stream. + """ list_sessions = getattr(self.store, "list_external_sessions", None) if not callable(list_sessions): @@ -179,13 +192,21 @@ class ExternalAgentBroker: agent_type=adapter.agent_type, project_id=project_id, ) - return bool( - selected is not None - and selected_token == token - and external_session_allows_resume(selected) - and str(getattr(selected, "status", "") or "").strip().lower() - in {"done", "suspended"} - ) + if ( + selected is None + or selected_token != token + or not external_session_allows_resume(selected) + ): + # The newest row for this token is terminally non-resumable. + return False + status = str(getattr(selected, "status", "") or "").strip().lower() + if status in {"done", "suspended"}: + return True + # The newest row is alive but not finalized (running / awaiting_human / + # awaiting_peer). An approval or peer park is not evidence the thread + # is dead, so a canonical token keeps its pin; a provider_stream token + # must not resume an attempt that never finalized. + return False if strict else None @classmethod def _task_explicitly_selected_external_agent(cls, task: Task, agent_type: str) -> bool: @@ -541,6 +562,8 @@ class ExternalAgentBroker: task=task, role_session_id=role_session_id, token=session_token, + strict=str(entry.get("source", "") or "").strip() + == "provider_stream", ) if session_token else None @@ -2580,13 +2603,15 @@ class ExternalAgentBroker: ) elif ( role_session_id - and result.status != TaskStatus.DONE + and result.status in self._SESSION_INVALIDATING_RESULT_STATUSES and hasattr(self.store, "get_role_session_adapter_state") and hasattr(self.store, "update_role_session_adapter_state") ): - # A stream token is durable early so Stop can retain it. A normal - # terminal failure must clear both a token discovered by this task - # and an older role token that this failed attempt resumed. + # A stream token is durable early so Stop can retain it, and a + # park (awaiting_human / awaiting_peer) keeps it so the run can + # resume the same thread once the gate clears. A terminal failure + # must clear both a token discovered by this task and an older + # role token that this failed attempt resumed. try: current = await self.store.get_role_session_adapter_state( role_session_id, @@ -2622,7 +2647,7 @@ class ExternalAgentBroker: logger.opt(exception=True).debug( "Failed to clear provider-stream role state after terminal failure" ) - if result.status != TaskStatus.DONE: + if result.status in self._SESSION_INVALIDATING_RESULT_STATUSES: failed_token = str( resume_session_id or provider_session_id diff --git a/tests/test_external_session_continuity.py b/tests/test_external_session_continuity.py index 5881117..284d83a 100644 --- a/tests/test_external_session_continuity.py +++ b/tests/test_external_session_continuity.py @@ -309,6 +309,54 @@ class BrokerPersistWritesRoleStateTests(unittest.IsolatedAsyncioTestCase): "provider_terminal_failure", ) + async def test_awaiting_human_park_keeps_role_token_and_task_metadata(self) -> None: + # An approval park is not a terminal failure: the run resumes the + # same provider thread once the human approves, so neither the role + # token nor the task's resume pin may be cleared. + await self._persist_run( + task_id="task-old", + agent_type="codex", + resume_session_id="thread-live", + ) + adapter = _MiniAdapter(agent_type="codex") + adapter.config.session_mode = "resume" + adapter.config.session_id = "thread-live" + task = self._task(task_id="task-parked") + task.metadata.update({ + "external_resume_session_id": "thread-live", + "external_resume_session_scope_id": "sess-a", + "external_resume_agent_type": "codex", + }) + await self.broker._persist_session( + adapter=adapter, + task=task, + workspace_path="/tmp/ws", + metadata={ + "command": "codex exec resume", + "model": "(cli default)", + "resume_session_id": "thread-live", + }, + result=TaskResult( + status=TaskStatus.AWAITING_HUMAN, + content="External action blocked by autonomy policy", + artifacts={ + "resume_session_id": "thread-live", + "requires_user_input": True, + }, + ), + ) + + entry = await self.store.get_role_session_adapter_state( + self.role_session_id, + "codex", + ) + self.assertIsNotNone(entry) + self.assertEqual(entry["resume_session_id"], "thread-live") + self.assertEqual( + task.metadata.get("external_resume_session_id"), "thread-live" + ) + self.assertNotIn("external_resume_fallback", task.metadata) + async def test_consecutive_tasks_overwrite_with_latest_token(self) -> None: await self._persist_run( task_id="task-1", agent_type="codex", resume_session_id="thread-1", @@ -571,6 +619,84 @@ class BrokerRestorePrefersRoleStateTests(unittest.IsolatedAsyncioTestCase): "provider_terminal_failure", ) + async def test_restore_keeps_role_token_when_newest_row_is_parked_awaiting_human(self) -> None: + # A canonical role token whose newest run parked on approval must + # keep its pin: awaiting_human is not evidence the thread is dead, + # and resume-after-approval depends on this token surviving. + token = "thread-parked" + await self.store.update_role_session_adapter_state( + self.role_session_id, + "codex", + { + "resume_session_id": token, + "provider_session_id": token, + "last_task_id": "task-old", + }, + ) + await self.store.save_external_session(ExternalSession( + agent_type="codex", + project_id="proj1", + session_id=token, + opc_session_id=self.role_session_id, + task_id="task-parked", + workspace_path="/tmp/ws", + run_mode="exec", + status="awaiting_human", + metadata={ + "resume_session_id": token, + "provider_session_id": token, + }, + )) + adapter = _MiniAdapter(agent_type="codex", can_resume_blank=False) + task = self._task() + + await self.broker._restore_session_resume_from_store(adapter, task) + + self.assertEqual(adapter.config.session_mode, "resume") + self.assertEqual(adapter.config.session_id, token) + self.assertIsNotNone(await self.store.get_role_session_adapter_state( + self.role_session_id, + "codex", + )) + + async def test_restore_still_rejects_unfinalized_provider_stream_token(self) -> None: + # provider_stream tokens keep the strict pre-existing rule: a row + # that never finalized (still "working") may belong to a crashed + # attempt and must not be resumed. + token = "thread-unfinalized" + await self.store.update_role_session_adapter_state( + self.role_session_id, + "codex", + { + "resume_session_id": token, + "provider_session_id": token, + "last_task_id": "task-new", + "source": "provider_stream", + "status": "working", + }, + ) + await self.store.save_external_session(ExternalSession( + agent_type="codex", + project_id="proj1", + session_id=token, + opc_session_id=self.role_session_id, + task_id="task-new", + workspace_path="/tmp/ws", + run_mode="exec", + status="working", + metadata={ + "resume_session_id": token, + "provider_session_id": token, + }, + )) + adapter = _MiniAdapter(agent_type="codex", can_resume_blank=False) + task = self._task() + + await self.broker._restore_session_resume_from_store(adapter, task) + + self.assertNotEqual(adapter.config.session_mode, "resume") + self.assertEqual(adapter.config.session_id, "") + if __name__ == "__main__": unittest.main()