fix(agent): keep claude_code resume across failed/cancelled runs

opc/engine.py

@@ -6603,7 +6603,8 @@ async def _prepare_company_runtime_tasks_for_resume(
                 has_resumable_external_session = bool(
                     gate_external_session
                 ) and external_session_status_allows_resume(
-                    gate_external_session.get("status")
+                    gate_external_session.get("status"),
+                    agent_type=gate_external_session.get("agent_type"),
                 )
                 if not has_resumable_external_session:
                     # The pin is a preference, not a fact: this work item has
@@ -6717,14 +6718,15 @@ async def _prepare_company_runtime_tasks_for_resume(
             task.metadata.pop("external_resume_checkpoint_session_updated_at", None)
             task.metadata.pop("external_resume_checkpoint_session_status", None)
             if isinstance(external_session, dict):
-                token_allowed = external_session_status_allows_resume(
-                    external_session.get("status")
-                )
                 agent_type = str(
                     external_session.get("agent_type")
                     or task.assigned_external_agent
                     or ""
                 ).strip()
+                token_allowed = external_session_status_allows_resume(
+                    external_session.get("status"),
+                    agent_type=agent_type,
+                )
                 assigned_agent_type = str(task.assigned_external_agent or "").strip()
                 token_candidates = [
                     str(external_session.get("resume_session_id") or "").strip(),
@@ -7100,7 +7102,7 @@ def _timestamp(value: Any) -> float:
             key=lambda session: _timestamp(getattr(session, "updated_at", None)),
         )
         latest_status = str(getattr(latest, "status", "") or "").strip().lower()
-        if external_session_status_allows_resume(latest_status):
+        if external_session_status_allows_resume(latest_status, agent_type=agent_type):
             return False
         latest_timestamp = _timestamp(getattr(latest, "updated_at", None))
         checkpoint_timestamp = _timestamp(checkpoint_updated_at)

opc/layer3_agent/external_broker.py

@@ -33,6 +33,8 @@
 from opc.layer2_organization.work_item_links import linked_work_item_id_for_task, set_linked_work_item_id
 from opc.layer3_agent.adapters.base import ExternalAgentAdapter
 from opc.layer3_agent.external_session_identity import (
+    NON_RESUMABLE_EXTERNAL_SESSION_STATUSES,
+    agent_resume_survives_run_failure,
     external_session_allows_resume,
     external_session_matches_provider_token,
     is_provider_session_token,
@@ -202,6 +204,13 @@ async def _stored_provider_token_allows_resume(
         status = str(getattr(selected, "status", "") or "").strip().lower()
         if status in {"done", "suspended"}:
             return True
+        if (
+            agent_resume_survives_run_failure(adapter.agent_type)
+            and status in NON_RESUMABLE_EXTERNAL_SESSION_STATUSES
+        ):
+            # The run outcome is terminal-but-failed, yet this provider keeps
+            # its transcript on disk — the token still resumes the thread.
+            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
@@ -2604,6 +2613,7 @@ async def _persist_session(
         elif (
             role_session_id
             and result.status in self._SESSION_INVALIDATING_RESULT_STATUSES
+            and not agent_resume_survives_run_failure(adapter.agent_type)
             and hasattr(self.store, "get_role_session_adapter_state")
             and hasattr(self.store, "update_role_session_adapter_state")
         ):
@@ -2647,7 +2657,10 @@ async def _persist_session(
                 logger.opt(exception=True).debug(
                     "Failed to clear provider-stream role state after terminal failure"
                 )
-        if result.status in self._SESSION_INVALIDATING_RESULT_STATUSES:
+        if (
+            result.status in self._SESSION_INVALIDATING_RESULT_STATUSES
+            and not agent_resume_survives_run_failure(adapter.agent_type)
+        ):
             failed_token = str(
                 resume_session_id
                 or provider_session_id

opc/layer3_agent/external_session_identity.py

@@ -20,15 +20,39 @@
     "startup_timeout",
 })
 
+# Agent types whose provider keeps the full session transcript on disk and can
+# replay it regardless of how the *launching run* ended.  Claude Code writes
+# every session to ``~/.claude/projects/<cwd>/<session-id>.jsonl`` and
+# ``--resume <id>`` restores it even after the spawning process failed, was
+# cancelled, or timed out — so a bad run outcome must not invalidate the
+# resume token and silently drop the whole conversation context.  Agents not
+# listed here keep the conservative status gate above.
+DURABLE_TRANSCRIPT_AGENT_TYPES: frozenset[str] = frozenset({
+    "claude_code",
+})
+
+
+def agent_resume_survives_run_failure(agent_type: Any) -> bool:
+    return (
+        str(agent_type or "").strip().lower() in DURABLE_TRANSCRIPT_AGENT_TYPES
+    )
 
-def external_session_status_allows_resume(status: Any) -> bool:
+
+def external_session_status_allows_resume(
+    status: Any,
+    *,
+    agent_type: Any = None,
+) -> bool:
+    if agent_resume_survives_run_failure(agent_type):
+        return True
     status = str(status or "").strip().lower()
     return status not in NON_RESUMABLE_EXTERNAL_SESSION_STATUSES
 
 
 def external_session_allows_resume(session: Any | None) -> bool:
     return session is not None and external_session_status_allows_resume(
-        getattr(session, "status", "")
+        getattr(session, "status", ""),
+        agent_type=getattr(session, "agent_type", None),
     )