fix(k8s): surface the real asset failure cause, not a bare "Job failed"

packages/interloper-k8s/src/interloper_k8s/runner.py

@@ -130,6 +130,11 @@ def _submit_asset(
             args=cmd[1:],
             env=env if env else None,
             resources=resources,
+            # On a non-zero exit with an empty termination-log, Kubernetes copies
+            # the tail of the container logs into the pod's termination state, so
+            # the host can recover a real cause even when the live log stream that
+            # carries child events has already dropped.
+            termination_message_policy="FallbackToLogsOnError",
         )
 
         pod_spec = client.V1PodSpec(
@@ -243,9 +248,10 @@ def _handle_completed(self, future: Future[Any], asset: Asset) -> None:
         try:
             future.result()
         except Exception as e:
-            self.state.mark_asset_failed(asset, str(e), emit=True)
+            error, tb = self._recover_asset_failure(job_name, asset.id, fallback=str(e))
+            self.state.mark_asset_failed(asset, error, tb=tb, emit=True)
             if self.fail_fast or self.reraise:
-                raise RunnerError(f"Asset '{type(asset).key}' failed: {e}") from e
+                raise RunnerError(f"Asset '{type(asset).key}' failed: {error}") from e
         else:
             self.state.mark_asset_completed(asset, emit=True)
 
@@ -265,7 +271,8 @@ def _handle_flushed_future(self, future: Future[Any], asset: Asset) -> None:
             try:
                 future.result()
             except Exception as e:  # noqa: BLE001
-                self.state.mark_asset_failed(asset, str(e), emit=True)
+                error, tb = self._recover_asset_failure(job_name, asset.id, fallback=str(e))
+                self.state.mark_asset_failed(asset, error, tb=tb, emit=True)
             else:
                 self.state.mark_asset_completed(asset, emit=True)
 
@@ -299,6 +306,141 @@ def _poll_job(self, job_name: str) -> None:
 
         raise RunnerError(f"Job {job_name} stopped (runner shutting down)")
 
+    # ------------------------------------------------------------------
+    # Failure recovery
+    # ------------------------------------------------------------------
+
+    def _recover_asset_failure(
+        self,
+        job_name: str | None,
+        target_asset_id: str,
+        *,
+        fallback: str,
+    ) -> tuple[str, str | None]:
+        """Recover the real failure cause from a failed child Job's pod.
+
+        The child streams its rich terminal (error + traceback) over the live
+        pod-log stream, but that follow-stream can drop while the pod is still
+        running (long retries, API-server idle timeouts), leaving the host with
+        only the Job status — a bare ``"Job ... failed"`` and no traceback.
+
+        On failure we re-read the pod's final logs (non-follow) and, failing
+        that, its container termination state, to recover the real error rather
+        than the Job status alone.
+
+        Returns:
+            ``(error, traceback)`` — the richest available; falls back to
+            ``fallback`` with no traceback when nothing can be recovered.
+        """
+        if job_name is None or self._core_v1 is None:
+            return fallback, None
+
+        pod = self._find_pod(job_name)
+        if pod is None:
+            return fallback, None
+
+        pod_name = pod.metadata.name if pod.metadata else None
+
+        # 1. Rich path: the child's own terminal event, re-read from the final
+        #    logs the live stream missed.
+        if pod_name is not None:
+            error, tb = self._terminal_from_pod_logs(pod_name, target_asset_id)
+            if error is not None:
+                return error, tb
+
+        # 2. Fallback: the container's termination state. With
+        #    terminationMessagePolicy=FallbackToLogsOnError, ``message`` carries
+        #    the tail of the logs; ``reason``/``exit_code`` flag OOMKills, signals.
+        return self._terminal_from_termination_state(pod, fallback)
+
+    def _find_pod(self, job_name: str) -> Any | None:
+        """Return the (first) pod owned by ``job_name``, or ``None``."""
+        if self._core_v1 is None:
+            return None
+        try:
+            pods = cast(
+                client.V1PodList,
+                self._core_v1.list_namespaced_pod(
+                    namespace=self.namespace,
+                    label_selector=f"job-name={job_name}",
+                ),
+            )
+        except Exception:  # noqa: BLE001
+            return None
+        return pods.items[0] if pods.items else None
+
+    def _terminal_from_pod_logs(
+        self,
+        pod_name: str,
+        target_asset_id: str,
+    ) -> tuple[str | None, str | None]:
+        """Re-read the pod's final logs and parse the child's terminal event.
+
+        Returns the error/traceback of the last ``asset_failed`` /
+        ``asset_exec_failed`` event for the target asset, or ``(None, None)``
+        when the logs hold no parseable terminal (e.g. the pod was SIGKILLed).
+        """
+        if self._core_v1 is None:
+            return None, None
+        try:
+            logs = cast(
+                str,
+                self._core_v1.read_namespaced_pod_log(
+                    name=pod_name,
+                    namespace=self.namespace,
+                    container="interloper",
+                    tail_lines=2000,
+                ),
+            )
+        except Exception:  # noqa: BLE001
+            return None, None
+
+        error: str | None = None
+        tb: str | None = None
+        for line in logs.splitlines():
+            line = line.rstrip()
+            if not line:
+                continue
+            try:
+                event = parse_event_from_log_line(line)
+            except Exception:  # noqa: BLE001
+                continue
+            if event is None or event.type not in (EventType.ASSET_FAILED, EventType.ASSET_EXEC_FAILED):
+                continue
+            event_asset_id = event.metadata.get("asset_id")
+            if event_asset_id and event_asset_id != target_asset_id:
+                continue
+            err = event.metadata.get("error")
+            if err:
+                error = err
+                tb = event.metadata.get("traceback") or tb
+        return error, tb
+
+    def _terminal_from_termination_state(self, pod: Any, fallback: str) -> tuple[str, str | None]:
+        """Build a cause from the pod's terminated container state.
+
+        ``reason``/``exit_code`` are appended to ``fallback`` (so an OOMKill or
+        signal is no longer hidden behind a bare "Job failed"); the termination
+        ``message`` — the log tail under FallbackToLogsOnError — becomes the
+        traceback shown in the UI.
+        """
+        status = getattr(pod, "status", None)
+        if status is None:
+            return fallback, None
+        for cs in status.container_statuses or []:
+            term = cs.state.terminated if cs.state else None
+            if term is None:
+                continue
+            bits: list[str] = []
+            if term.reason:
+                bits.append(f"reason={term.reason}")
+            if term.exit_code is not None:
+                bits.append(f"exit_code={term.exit_code}")
+            error = f"{fallback} ({', '.join(bits)})" if bits else fallback
+            tb = term.message or None
+            return error, tb
+        return fallback, None
+
     # ------------------------------------------------------------------
     # Job helpers
     # ------------------------------------------------------------------

packages/interloper-k8s/tests/test_runner_terminal_events.py

@@ -11,6 +11,8 @@
 from collections.abc import Iterator
 from concurrent.futures import Future
 from contextlib import contextmanager
+from types import SimpleNamespace
+from unittest.mock import MagicMock
 
 import interloper as il
 from interloper.errors import RunnerError
@@ -135,3 +137,97 @@ def test_no_fail_fast_keeps_quiet_on_child_reported_failure() -> None:
         runner._handle_completed(future, asset)  # must not raise
 
     assert not [e for e in events if e.type == EventType.ASSET_FAILED]
+
+
+# ----------------------------------------------------------------------
+# Failure recovery: the host-authored terminal carries the pod's real cause
+# (not a bare "Job ... failed") when the live event stream dropped before the
+# child reported its terminal.
+# ----------------------------------------------------------------------
+
+
+def _pod(name: str = "pod-x", *, terminated: SimpleNamespace | None = None) -> SimpleNamespace:
+    container_statuses = None
+    if terminated is not None:
+        container_statuses = [SimpleNamespace(state=SimpleNamespace(terminated=terminated))]
+    return SimpleNamespace(
+        metadata=SimpleNamespace(name=name),
+        status=SimpleNamespace(container_statuses=container_statuses),
+    )
+
+
+def _mock_core_v1(*, pod: SimpleNamespace | None = None, logs: str = "") -> MagicMock:
+    core = MagicMock()
+    core.list_namespaced_pod.return_value = SimpleNamespace(items=[pod] if pod is not None else [])
+    core.read_namespaced_pod_log.return_value = logs
+    return core
+
+
+def _failed_event_line(asset_id: str, error: str, traceback: str | None = None) -> str:
+    metadata: dict[str, str] = {"asset_id": asset_id, "error": error}
+    if traceback is not None:
+        metadata["traceback"] = traceback
+    return Event(type=EventType.ASSET_FAILED, metadata=metadata).to_json()
+
+
+def test_recover_returns_fallback_without_job_or_client() -> None:
+    """No job name (or no k8s client) → the bare fallback, no traceback."""
+    runner, _ = _runner_with_asset("asset-1", "run-1")
+    assert runner._recover_asset_failure(None, "asset-1", fallback="Job x failed") == ("Job x failed", None)
+
+    runner._core_v1 = _mock_core_v1()
+    assert runner._recover_asset_failure(None, "asset-1", fallback="Job x failed") == ("Job x failed", None)
+
+
+def test_recover_reads_rich_terminal_from_pod_logs() -> None:
+    """The child's own error + traceback are recovered from the final logs."""
+    runner, _ = _runner_with_asset("asset-1", "run-1")
+    line = _failed_event_line("asset-1", "429 Too Many Requests", "Traceback...\nHTTPStatusError")
+    runner._core_v1 = _mock_core_v1(pod=_pod(), logs=f"some debug log\n{line}\n")
+
+    error, tb = runner._recover_asset_failure("job-x", "asset-1", fallback="Job job-x failed")
+
+    assert error == "429 Too Many Requests"
+    assert tb is not None and "HTTPStatusError" in tb
+
+
+def test_recover_ignores_other_assets_terminal_in_logs() -> None:
+    """A terminal for a different asset must not be attributed to this one."""
+    runner, _ = _runner_with_asset("asset-1", "run-1")
+    term = SimpleNamespace(reason="Error", exit_code=1, message="log tail")
+    runner._core_v1 = _mock_core_v1(pod=_pod(terminated=term), logs=_failed_event_line("other-asset", "not mine"))
+
+    error, tb = runner._recover_asset_failure("job-x", "asset-1", fallback="Job job-x failed")
+
+    # No matching log terminal → falls through to the termination state.
+    assert error == "Job job-x failed (reason=Error, exit_code=1)"
+    assert tb == "log tail"
+
+
+def test_recover_falls_back_to_termination_state_on_oomkill() -> None:
+    """A SIGKILLed pod has no terminal event; reason/exit_code surface instead."""
+    runner, _ = _runner_with_asset("asset-1", "run-1")
+    term = SimpleNamespace(reason="OOMKilled", exit_code=137, message="killed log tail")
+    runner._core_v1 = _mock_core_v1(pod=_pod(terminated=term), logs="no events here\n")
+
+    error, tb = runner._recover_asset_failure("job-x", "asset-1", fallback="Job job-x failed")
+
+    assert error == "Job job-x failed (reason=OOMKilled, exit_code=137)"
+    assert tb == "killed log tail"
+
+
+def test_handle_completed_emits_recovered_error_and_traceback() -> None:
+    """End to end: the host-authored ``asset_failed`` carries the recovered cause."""
+    runner, asset = _runner_with_asset("asset-1", "run-1")
+    runner._core_v1 = _mock_core_v1(pod=_pod(), logs=_failed_event_line("asset-1", "429 Too Many Requests", "rich-tb"))
+    future: Future[None] = Future()
+    future.set_exception(RunnerError("Job interloper-run-x failed"))
+    runner._job_map[future] = "interloper-run-x"
+
+    with _capture() as events:
+        runner._handle_completed(future, asset)
+
+    failed = [e for e in events if e.type == EventType.ASSET_FAILED]
+    assert len(failed) == 1
+    assert failed[0].metadata.get("error") == "429 Too Many Requests"
+    assert failed[0].metadata.get("traceback") == "rich-tb"