Don't adopt a terminating Job on the event-loop respawn path (#3597)

docs/reference/agent-recovery.md

@@ -263,7 +263,12 @@ Restarts are allowed when the pipeline is in `RUNNING`, `AWAITING_HUMAN`, `FAILE
 8. The pipeline's `PhaseExecution` state is updated with the new container/agent entries
 9. Restart count is tracked per agent per phase — configurable maximum (default: 2) prevents infinite restart loops
 
-The response reports whether step 5 actually found a live loop to delegate the respawn to (`live_event_loop`, `arms_invalidated: <count>`) and adjusts the `respawn` field accordingly instead of unconditionally claiming success (#3548) — see the *Event-loop arm invalidation* deep-dive below for the exact `respawn` values and their conditions.
+The response reports whether step 5 actually found a live loop to delegate the respawn to (`live_event_loop`, `arms_invalidated: <count>`) and adjusts the `respawn` field accordingly instead of unconditionally claiming success (#3548) — see the *Event-loop arm invalidation* deep-dive below for the exact `respawn` values and their conditions. It also reports the teardown itself: `jobs_torn_down: <count>` (`0` means the role had already exited and there was nothing to kill) and `teardown_confirmed` (whether the deletion was observed to complete before the route returned) — see *Asynchronous Job deletion* below. The `restart_agent` MCP tool passes all four fields through, replacing the always-empty `container_id` it used to return (#3597).
+
+**Asynchronous Job deletion ([#3597](https://github.com/jwbron/egg/issues/3597)).** Deleting a Kubernetes Job is asynchronous: the API server accepts the request and the Job then sits in `Terminating` — still reporting `active > 0`, i.e. `RUNNING` — until its dependent pods finish terminating. Step 2's teardown returned immediately into that window, and the event loop polls every ~5s, so a poll landing inside it matched the still-terminating Job on its dedupe-key label, logged `Adopting existing live Job for event (dedupe hit)`, and declined to spawn a replacement. Because adoption also re-arms the key in the loop's live set — where a *missing* Job reads as "still running" — the role then stayed vanished indefinitely: no pod, no Job, `get_status` reporting `status: running` with `container_id: null`. An immediate second `restart_agent` call typically worked (the corpse had been reaped by then), which reads as a fluke rather than a race. Two changes close it:
+
+- **Terminating Jobs are not adoptable.** `ContainerInfo` carries the Job's `deletionTimestamp` (populated by `KubernetesClient.list_jobs`), and `_event_dedupe_key_live` (`orchestrator/kubernetes_spawner/_events.py`) counts a Job as live only when it is in `LIVE_POD_STATUSES` **and** unstamped. Deletion-in-progress is the third state the predicate models, alongside terminal Jobs ([#3181](https://github.com/jwbron/egg/issues/3181)) and the live ones adoption exists for. Since a one-shot Job's name is derived from its dedupe key, the replacement collides with the name of the Job being reaped, so `spawn_event_job` waits the corpse out (bounded by `_EVENT_JOB_TERMINATION_WAIT_S`, 15s) before creating — the event-loop twin of the `restart_agent_job` wait added in [#2655](https://github.com/jwbron/egg/issues/2655). Overrunning that budget is logged and the spawn proceeds: a 409 `AlreadyExists` is isolated per-role by the loop and retried on the next poll, which costs a poll interval rather than the role.
+- **The route waits for the teardown it requested.** `_restart_agent_body` waits (bounded by `_JOB_TEARDOWN_WAIT_SECONDS`, 20s shared across every Job it deleted, well under the MCP client's 60s restart timeout) for each deleted Job to be observed gone before returning, so the respawn it delegates starts from a clean slate. A timeout is reported as `teardown_confirmed: false`, never a failed restart.
 
 **Event-loop arm invalidation (#3548).** The consensus reset in step 4 makes the event loop re-derive the role's next event, but with the *same* identity — and therefore the same dedupe key — as before the restart, so loop-local state silently blocked the respawn: the key stayed in the loop's live-key set (the route deletes the Job by label, and Job observation maps a missing Job to "still running"), and any exhaustion / no-op-park latch for the key survived untouched. `_restart_agent_body` (`orchestrator/routes/pipelines/_routes_restart.py`) now reaches into the live event loop for the restarted role's `(pipeline_id, slice_id)` and calls `invalidate_role_arms(agent_role)`, which drops the role's keys from the loop's live-key/metadata tracking and retires their supervisor state (unioning `_key_meta` with the supervisor's own parked/exhausted key reports, since a parked key has already been popped from `_key_meta`) so the next poll re-derives the key as fresh and actually spawns. The route's JSON response now reports `live_event_loop` (bool) and `arms_invalidated` (count), and the `respawn` field is honest about whether a live loop exists to honor the delegation: `"delegated to orchestrator event loop"` when one was found, `"driver thread relaunched; event loop will respawn the role"` when the pipeline was inactive, or `"no live event loop for this slice — no respawn will occur; restart the phase if the agent must re-run"` otherwise.
 

orchestrator/kubernetes_client.py

@@ -874,6 +874,16 @@ def list_jobs(
 
                 started_at = _parse_k8s_datetime(job.status.start_time if job.status else None)
 
+                # A deleted Job keeps reporting its pre-delete status
+                # (``active > 0`` ⇒ RUNNING) for as long as its pods take to
+                # terminate, so status alone cannot tell a live Job from one
+                # on its way out. Surface the deletion stamp so callers that
+                # care — the event-loop dedupe predicate (#3597) — can tell
+                # the difference.
+                deletion_timestamp = _parse_k8s_datetime(
+                    getattr(job.metadata, "deletion_timestamp", None)
+                )
+
                 results.append(
                     ContainerInfo(
                         container_id=uid,
@@ -883,6 +893,7 @@ def list_jobs(
                         exited_at=exited_at,
                         namespace=namespace,
                         job_name=job_name,
+                        deletion_timestamp=deletion_timestamp,
                     )
                 )
 

orchestrator/kubernetes_spawner/__init__.py

@@ -202,6 +202,15 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 # already-hashed dedupe key is plenty of separation.
 _EVENT_JOB_NAME_DISCRIMINATOR_LEN = 8
 
+# Because that discriminator is deterministic, a re-derived event's Job name
+# collides with the one just deleted for the same key. Job deletion is
+# asynchronous, so the spawn path waits (at most this long, shared across every
+# matching Job) for the terminating object to actually go away before creating
+# its replacement — otherwise the create 409s ``AlreadyExists`` (#3597, the
+# event-loop twin of the #2655 restart-path wait). Kept short: this blocks the
+# event loop's poll thread, and overrunning it only costs a retry next poll.
+_EVENT_JOB_TERMINATION_WAIT_S = 15.0
+
 # Kubernetes caps label VALUES (and names) at 63 characters and rejects any
 # overflow at the API server. The dedupe key is a 64-char sha256 hexdigest, so
 # it must be shortened to a label-safe form before it can ride as a Job label
@@ -519,9 +528,15 @@ def get_kubernetes_spawner(
 KubernetesSpawner._teardown_session = _session._teardown_session
 KubernetesSpawner.sync_session_phases = _session.sync_session_phases
 KubernetesSpawner.spawn_agent_job = _spawn.spawn_agent_job
+KubernetesSpawner._list_event_jobs = _events._list_event_jobs
 KubernetesSpawner._event_dedupe_key_live = _events._event_dedupe_key_live
 KubernetesSpawner.create_event_job_status_view = _events.create_event_job_status_view
 KubernetesSpawner.spawn_event_job = _events.spawn_event_job
+# Module-level Job-state predicates shared by the adoption filter and the
+# terminating-Job wait (#3597); re-exported so they are addressable/testable
+# through the barrel like the rest of the private surface.
+_job_is_live = _events._job_is_live
+_job_is_terminating = _events._job_is_terminating
 KubernetesSpawner.stop_agent_job = _jobs.stop_agent_job
 KubernetesSpawner.remove_agent_job = _jobs.remove_agent_job
 # Module-level (not a class method) so ``remove_agent_job`` reaches it via the
@@ -570,6 +585,8 @@ def get_kubernetes_spawner(
     "_classify_spawn_error",
     "_fit_k8s_name",
     "_dedupe_label_value",
+    "_job_is_live",
+    "_job_is_terminating",
     "_forwarded_discipline_env",
     "_resolve_live_phase",
     "_resolve_wait_producer_allowlist",