fix(runner): classify failures from current terminal diagnostics

argus_skill/adapters/agent_cli_backend/_exec_spawn.py

@@ -29,6 +29,7 @@
     is_execution_host_startup_error,
     is_model_catalog_startup_error,
     result_has_pre_provider_refusal,
+    terminal_failure_diagnostic,
 )
 from ...core.runner_receipts import is_provider_turn_cap_receipt
 from ...core.secret_guard import redact_secrets_text
@@ -283,9 +284,8 @@ def spawn_and_finish(ctx: "_ExecContext", cli_options: Any) -> RunnerResult:
         or getattr(cli_result, "fatal_error", None)
         or int(getattr(cli_result, "exit_code", 0) or 0) != 0
     )
-    stderr_lines = list(getattr(cli_result, "stderr_lines", None) or [])
     fatal_error = str(getattr(cli_result, "fatal_error", "") or "")
-    failure_text = "\n".join([fatal_error, *map(str, stderr_lines)]).strip()
+    failure_text = terminal_failure_diagnostic(cli_result)
     safe_failure_text = redact_secrets_text(
         failure_text,
         known_values=backend._known_secret_values,

argus_skill/adapters/agent_cli_backend/_result.py

@@ -19,7 +19,7 @@
 from ...core.role_decision import extract_role_decisions
 from ...core.runner_errors import (
     is_execution_host_startup_error,
-    is_model_catalog_startup_error,
+    terminal_failure_diagnostic,
 )
 from ...core.runner_receipts import is_provider_turn_cap_receipt
 from ...core.stop_kinds import (
@@ -363,35 +363,25 @@ def translate_result(
             {**row, "model": authoritative_usage_model}
             for row in model_usage
         ]
-    raw_fatal_error = str(getattr(cli_result, "fatal_error", "") or "").strip()
-    fatal_error = _normalize_fatal_error(cli_result.fatal_error)
-    if (
-        getattr(cli_result, "turn_failed", False)
-        and not fatal_error
-    ):
-        fatal_error = "\n".join(
-            map(str, getattr(cli_result, "stderr_lines", None) or [])
-        ).strip() or "backend reported a failed turn"
-    failure_diagnostic = raw_fatal_error
-    authoritative_local_stop = (
-        is_provider_turn_cap_receipt(raw_fatal_error)
-        or is_execution_host_startup_error(raw_fatal_error)
+    failure_diagnostic = terminal_failure_diagnostic(cli_result)
+    fatal_error = _normalize_fatal_error(failure_diagnostic) or None
+    stop_kind = normalize_stop_kind(getattr(cli_result, "stop_kind", None)) or _raw_backend_stop_kind(
+        fatal_error=failure_diagnostic, exit_code=cli_result.exit_code,
     )
-    if not authoritative_local_stop and (
-        getattr(cli_result, "turn_failed", False)
-        or int(getattr(cli_result, "exit_code", 0) or 0) != 0
+    # Preserve the established Manager timeout backoff for its final explicit
+    # reconnect/429 notice. This narrow scheduling hint must not participate
+    # in auth replay, overwrite the receipt, or override an operator stop.
+    if (
+        stop_kind == "transient_error"
+        and failure_diagnostic.casefold().startswith(
+            "external interrupt: manager turn wall-clock limit reached"
+        )
     ):
-        stderr_diagnostic = "\n".join(
-            map(str, getattr(cli_result, "stderr_lines", None) or [])
-        ).strip()
-        if is_model_catalog_startup_error(stderr_diagnostic):
-            # CLI startup can provide only a generic terminal receipt while
-            # stderr explains that model discovery failed before any turn.
-            fatal_error = stderr_diagnostic
-        if stderr_diagnostic and stderr_diagnostic not in failure_diagnostic:
-            failure_diagnostic = "\n".join(
-                part for part in (failure_diagnostic, stderr_diagnostic) if part
-            )
+        latest = next((str(line).strip() for line in reversed(
+            getattr(cli_result, "stderr_lines", None) or []
+        ) if str(line).strip()), "")
+        if latest.casefold().startswith("reconnecting") and has_http_status(latest, {429}):
+            stop_kind = "provider_cooldown"
     return RunnerResult(
         exit_code=cli_result.exit_code,
         agent_messages=list(cli_result.agent_messages or []),
@@ -400,13 +390,7 @@ def translate_result(
         stderr_lines=list(cli_result.stderr_lines or []),
         thread_id=cli_result.thread_id or resume_thread_id,
         fatal_error=fatal_error,
-        stop_kind=(
-            normalize_stop_kind(getattr(cli_result, "stop_kind", None))
-            or _raw_backend_stop_kind(
-                fatal_error=failure_diagnostic,
-                exit_code=cli_result.exit_code,
-            )
-        ),
+        stop_kind=stop_kind,
         input_tokens=input_tokens,
         cached_input_tokens=cached_input_tokens,
         cache_write_tokens=raw_usage.cache_write_tokens,

argus_skill/agent_cli/_event_consumers.py

@@ -315,7 +315,11 @@ def _consume_codex_event(
         event_type = event.get("type")
         if event_type == "thread.started":
             thread_id = event.get("thread_id", thread_id)
+        elif event_type == "turn.started" and not turn_failed:
+            fatal_error = None
         elif event_type == "item.completed":
+            if not turn_failed:
+                fatal_error = None
             item = event.get("item", {})
             if not isinstance(item, dict):
                 return thread_id, turn_completed, turn_failed, fatal_error
@@ -335,12 +339,14 @@ def _consume_codex_event(
             turn_completed = True
         elif event_type == "turn.failed":
             turn_failed = True
+            if not is_execution_host_startup_error(fatal_error):
+                fatal_error = "Backend reported a failed turn."
             err = event.get("error", {})
             if isinstance(err, dict):
                 maybe_msg = err.get("message")
-                if isinstance(maybe_msg, str) and not is_execution_host_startup_error(fatal_error):
+                if isinstance(maybe_msg, str) and maybe_msg.strip() and not is_execution_host_startup_error(fatal_error):
                     fatal_error = maybe_msg
-        elif event_type == "error" and fatal_error is None:
+        elif event_type == "error" and not turn_failed:
             maybe_msg = event.get("message")
             if isinstance(maybe_msg, str):
                 fatal_error = maybe_msg

argus_skill/agent_cli/_run_exec.py

@@ -75,6 +75,7 @@ class _StreamState:
     fatal_error: str | None = None
     provider_turns: int = 0
     provider_turn_cap_hit: bool = False
+    model_progress_observed: bool = False
     tool_activity_observed: bool = False
     usage_model: str = ""
     watchdog_terminated: bool = False
@@ -560,6 +561,11 @@ def check_wall_clock_limit() -> bool:
                 if event is None:
                     continue
                 state.json_event_count += 1
+                if event.get("type") in {
+                    "turn.started", "turn.completed", "turn.failed",
+                    "item.started", "item.updated", "item.completed",
+                }:
+                    state.model_progress_observed = True
                 if (
                     provider_turn_cap > 0
                     and not state.watchdog_terminated
@@ -769,7 +775,7 @@ def _finalize_turn_result(
 
         if state.watchdog_terminated:
             state.turn_failed = True
-            if state.watchdog_reason and state.fatal_error is None:
+            if state.watchdog_reason:
                 state.fatal_error = state.watchdog_reason
         elif state.turn_completed and not state.turn_failed:
             state.fatal_error = None
@@ -787,7 +793,11 @@ def _finalize_turn_result(
             # authoritative completion event. Preserve stderr when available
             # so configuration failures still retain their concrete diagnosis.
             state.turn_failed = True
-            state.fatal_error = _incomplete_turn_error(state.stderr_lines)
+            state.fatal_error = (
+                "Agent CLI exited without completing a model turn."
+                if state.model_progress_observed
+                else _incomplete_turn_error(state.stderr_lines)
+            )
 
         return AgentRunResult(
             command=command,
@@ -805,6 +815,7 @@ def _finalize_turn_result(
             fatal_error=state.fatal_error,
             provider_turns=state.provider_turns,
             provider_turn_cap_hit=state.provider_turn_cap_hit,
+            model_progress_observed=state.model_progress_observed,
             tool_activity_observed=state.tool_activity_observed,
             usage_model=state.usage_model,
             orphan_process_group_id=state.orphan_process_group_id,

argus_skill/agent_cli/models.py

@@ -26,6 +26,7 @@ class AgentRunResult:
     # keeps the work and continues in a fresh session from the checkpoint.
     provider_turns: int = 0
     provider_turn_cap_hit: bool = False
+    model_progress_observed: bool = False
     tool_activity_observed: bool = False
     usage_model: str = ""
     orphan_process_group_id: int = 0

argus_skill/core/runner_errors.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Any
 
 _MISSING_RESUME_TARGET = "No session, task, or name matched"
@@ -30,6 +31,44 @@
 )
 
 
+def terminal_failure_diagnostic(result: Any) -> str:
+    """Select one current diagnostic; stderr remains a separate history.
+
+    Concrete terminal receipts win. Generic process-exit receipts may use the
+    latest startup diagnostic only when no model/tool progress was observed.
+    This also supports older/external runners without diagnostic provenance.
+    Never combine old stderr with the current failure for control decisions.
+    """
+    fatal = str(getattr(result, "fatal_error", None) or "").strip()
+    failed = bool(getattr(result, "turn_failed", False)
+                  or int(getattr(result, "exit_code", 0) or 0) != 0 or fatal)
+    if getattr(result, "turn_completed", False) and not getattr(result, "turn_failed", False) and int(getattr(result, "exit_code", 0) or 0) == 0:
+        return ""
+    generic = (not fatal or fatal.casefold() == "process exited" or fatal == "Agent CLI exited without completing a model turn."
+               or bool(re.fullmatch(r"Process exited with code -?\d+ before turn completion\.", fatal)))
+    if not generic:
+        return fatal
+    progress = bool(
+        getattr(result, "provider_turns", 0)
+        or getattr(result, "model_progress_observed", False)
+        or getattr(result, "tool_activity_observed", False)
+        or getattr(result, "agent_messages", None)
+        or any(e.get("type") in {"turn.started", "turn.completed", "turn.failed",
+                                 "item.started", "item.updated", "item.completed"}
+               for e in (getattr(result, "json_events", None) or []) if isinstance(e, dict))
+    )
+    if progress:
+        return (fatal or "Backend exited after progress without a terminal diagnostic.") if failed else ""
+    # Startup has no model turn to recover within. The last nonempty line is
+    # the best available evidence; do not search backwards for a desired code.
+    for line in reversed(getattr(result, "stderr_lines", None) or []):
+        if str(line).strip():
+            text = str(line).strip()
+            # Some providers reject startup with exit 0 and no turn receipt.
+            return text if failed or is_pre_provider_refusal_error(text) else ""
+    return fatal
+
+
 def is_execution_host_startup_error(value: object) -> bool:
     """Recognize the Codex runtime receipt for an unavailable execution host.
 
@@ -93,30 +132,19 @@ def is_unrecoverable_resume_error(value: object) -> bool:
 
 
 def result_has_missing_resume_target(result: Any) -> bool:
-    parts = [
-        getattr(result, "fatal_error", ""),
-        *(getattr(result, "stderr_lines", None) or []),
-    ]
-    return is_missing_resume_target_error("\n".join(map(str, parts)))
+    return is_missing_resume_target_error(terminal_failure_diagnostic(result))
 
 
 def result_has_unrecoverable_resume_state(result: Any) -> bool:
-    parts = [
-        getattr(result, "fatal_error", ""),
-        *(getattr(result, "stderr_lines", None) or []),
-    ]
-    return is_unrecoverable_resume_error("\n".join(map(str, parts)))
+    return is_unrecoverable_resume_error(terminal_failure_diagnostic(result))
 
 
 def result_has_pre_provider_refusal(result: Any) -> bool:
-    parts = [
-        getattr(result, "fatal_error", ""),
-        *(getattr(result, "stderr_lines", None) or []),
-    ]
-    return is_pre_provider_refusal_error("\n".join(map(str, parts)))
+    return is_pre_provider_refusal_error(terminal_failure_diagnostic(result))
 
 
 __all__ = [
+    "terminal_failure_diagnostic",
     "is_execution_host_startup_error",
     "is_missing_resume_target_error",
     "is_model_catalog_startup_error",

argus_skill/provider_integrations/authorization_retry.py

@@ -9,7 +9,7 @@
 from typing import Any
 
 from ..core.http_status import has_http_status
-from ..core.runner_errors import is_execution_host_startup_error
+from ..core.runner_errors import is_execution_host_startup_error, terminal_failure_diagnostic
 from ..core.runner_receipts import is_provider_turn_cap_receipt
 from ..core.secret_guard import redact_secrets_text
 from ..tools.capability_vault import read_codex_provider_config
@@ -160,15 +160,9 @@ def _unauthorized_cause(result: Any) -> str:
     )
     if not failed:
         return ""
-    candidates = [
-        getattr(result, "fatal_error", None),
-        *reversed(list(getattr(result, "stderr_lines", None) or [])),
-    ]
-    for candidate in candidates:
-        text = str(candidate or "").strip()
-        if has_http_status(text, {401}):
-            return text
-    return ""
+    text = terminal_failure_diagnostic(result)
+    return text if has_http_status(text, {401}) else ""
+
 
 
 AUTHORIZATION_RETRY_OWNER = AuthorizationRetryOwner()