diff --git a/docs/reference/agent-recovery.md b/docs/reference/agent-recovery.md index 92a025352..ec5577050 100644 --- a/docs/reference/agent-recovery.md +++ b/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: `) 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: `) 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: ` (`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. diff --git a/orchestrator/kubernetes_client.py b/orchestrator/kubernetes_client.py index 8bc9080af..7ae4b4f94 100644 --- a/orchestrator/kubernetes_client.py +++ b/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, ) ) diff --git a/orchestrator/kubernetes_spawner/__init__.py b/orchestrator/kubernetes_spawner/__init__.py index 0e086f0d0..8854f8708 100644 --- a/orchestrator/kubernetes_spawner/__init__.py +++ b/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", diff --git a/orchestrator/kubernetes_spawner/_events.py b/orchestrator/kubernetes_spawner/_events.py index b9afb7242..603981660 100644 --- a/orchestrator/kubernetes_spawner/_events.py +++ b/orchestrator/kubernetes_spawner/_events.py @@ -4,11 +4,13 @@ the barrel (``from kubernetes_spawner import ...``), not directly. """ +from datetime import datetime from typing import Any import kubernetes_spawner as _pkg from kubernetes_spawner import ( _EVENT_JOB_NAME_DISCRIMINATOR_LEN, + _EVENT_JOB_TERMINATION_WAIT_S, ENV_EVENT_ACTION, ENV_EVENT_DEDUPE_KEY, ENV_EVENT_PAYLOAD_REFS, @@ -19,24 +21,13 @@ from models import LIVE_POD_STATUSES, AgentRole -def _event_dedupe_key_live(self, dedupe_key: str) -> bool: - """Return True iff a Job already carries this dedupe-key label. +def _list_event_jobs(self, dedupe_key: str) -> list[Any]: + """Return every Job carrying ``dedupe_key``'s label, or ``[]``. - The reconciliation handle: a fresh orchestrator process re-derives - every event and the spawner asks this before creating a Job, so an - in-flight Job from a prior process (or a racing duplicate request) is - adopted rather than duplicated. No spawn state is persisted — the - label IS the state. Queried via a label selector so the API returns - only matching Jobs; best-effort (a list failure ⇒ "not live" ⇒ spawn - proceeds rather than wedging). - - Only Jobs in a non-terminal status (``PENDING``/``RUNNING``) count as - live. ``list_jobs`` returns *all* label-matching Jobs regardless of - status, and one-shot event Jobs linger for ``ttl_seconds_after_finished`` - (10 min) after completing, so a terminated Job (``EXITED``/``FAILED``) - must NOT adopt a re-derived identical event — otherwise an event whose - pod failed without advancing the tracker would be silently swallowed - for the TTL window instead of respawned. + Queried via a label selector so the API returns only matching Jobs; + best-effort (a list failure ⇒ ``[]`` ⇒ callers treat the key as + un-owned and spawn rather than wedging). A non-sequence (e.g. an + unconfigured mock) is normalized to ``[]`` for the same reason. """ # The selector value MUST use the same label-safe shortening applied # to the label on the spawn side, or it can never match the live Job. @@ -49,21 +40,217 @@ def _event_dedupe_key_live(self, dedupe_key: str) -> bool: dedupe_key=dedupe_key, error=str(exc), ) - return False - # Count only Jobs whose pod is still doing work (PENDING / CREATING / - # RUNNING). A *terminal* Job — FAILED (crashed) or EXITED (clean rc=0) - # — lingers for the ~600s ``ttlSecondsAfterFinished`` window, and - # adopting one would dead-end the supervisor's bounded respawn: a - # crashed propose arm would be "adopted" (no new pod) for the whole TTL - # while its FAILED status keeps re-incrementing the abort streak, so a - # transient crash falsely escalates to AGENT_FAILED without ever - # retrying (#3181). Mirrors ``LIVE_POD_STATUSES`` — the single - # source of truth shared with ``_count_live_pods_for_pipeline`` / - # startup reconciliation. A non-sequence (e.g. an unconfigured mock) is - # treated as "no live Job" so the spawn proceeds. + return [] if not isinstance(jobs, (list, tuple)): - return False - return any(getattr(j, "status", None) in LIVE_POD_STATUSES for j in jobs) + return [] + return list(jobs) + + +def _job_is_terminating(job: Any) -> bool: + """Return True iff *job* has been deleted but has not gone away yet. + + Kubernetes Job deletion is asynchronous: the API server stamps + ``metadata.deletionTimestamp`` and the object lingers — still + reporting ``active > 0`` ⇒ ``RUNNING`` — until its dependent pods + finish terminating. + + Tested with ``isinstance`` rather than ``is not None`` because + ``ContainerInfo.deletion_timestamp`` is a ``datetime | None`` and + anything else is not a real stamp — most notably an auto-attribute on + an unconfigured mock, which would otherwise read as "terminating" and + silently disable adoption. Same "a mock is not evidence" convention + the live-Job list already applies to a non-sequence ``list_jobs``. + """ + return isinstance(getattr(job, "deletion_timestamp", None), datetime) + + +def _job_is_live(job: Any) -> bool: + """Return True iff *job* still has a pod that will do the event's work.""" + return getattr(job, "status", None) in LIVE_POD_STATUSES and not _job_is_terminating(job) + + +def _event_dedupe_key_live(self, dedupe_key: str) -> bool: + """Return True iff a *live* Job already carries this dedupe-key label. + + The reconciliation handle: a fresh orchestrator process re-derives + every event and the spawner asks this before creating a Job, so an + in-flight Job from a prior process (or a racing duplicate request) is + adopted rather than duplicated. No spawn state is persisted — the + label IS the state. + + "Live" means a Job that still has a pod which will do this event's + work. Three states must NOT qualify: + + * *terminal* — ``list_jobs`` returns all label-matching Jobs regardless + of status, and one-shot event Jobs linger for + ``ttl_seconds_after_finished`` (10 min) after completing. Adopting a + ``FAILED``/``EXITED`` Job would dead-end the supervisor's bounded + respawn: a crashed propose arm would be "adopted" (no new pod) for + the whole TTL while its FAILED status keeps re-incrementing the abort + streak, so a transient crash falsely escalates to AGENT_FAILED + without ever retrying (#3181); + * *terminating* — a deleted Job keeps reporting ``RUNNING``/``PENDING`` + until its pod actually terminates, so status alone cannot see that it + is on its way out. ``restart_agent`` deletes the role's Job and + delegates the respawn to the event loop; if the next poll landed + inside that deletion window the loop adopted the corpse, declined to + spawn, and the role silently vanished — no pod, no Job, state still + ``running`` (#3597). Adoption is only ever correct for a Job that + will still run the event, which a terminating one never will; + * *unknown* — a list failure yields no Jobs, which spawns (a duplicate + Job is recoverable; a swallowed event is not). + + The live status set mirrors ``LIVE_POD_STATUSES`` — the single source + of truth shared with ``_count_live_pods_for_pipeline`` / startup + reconciliation. + """ + return any(_job_is_live(j) for j in self._list_event_jobs(dedupe_key)) + + +def _await_terminating_event_jobs( + self, + jobs: list[Any], + *, + pipeline_id: str, + role: str, + action: str, + dedupe_key: str, +) -> None: + """Block until the terminating Jobs among *jobs* are gone (bounded) (#3597). + + A one-shot event Job's name is derived from the dedupe key, so the + replacement this spawn is about to create carries the *same* name as + the Job that was just deleted. ``delete_job`` returns as soon as the + deletion is accepted, and the object then lingers with its finalizer + until its pods finish terminating — creating into that window returns + 409 ``AlreadyExists`` (the #2655 race, which ``restart_agent_job`` + already waits out on its own path). + + Best-effort and bounded by ``_EVENT_JOB_TERMINATION_WAIT_S`` shared + across all matching Jobs: on timeout (or a k8s client without the wait + helper) we log and let the spawn proceed — a 409 there is raised as a + ``KubernetesSpawnError``, which the event loop isolates per-role and + retries on the next poll, so a slow reap costs a poll interval rather + than the silent vanish this whole path exists to prevent. + + Each of those outcomes is logged for what it actually is, matching the + taxonomy the restart route applies: "still present" is only claimed + after a wait ran and observed the Job, and every path that skips the + wait entirely — a Job the listing did not name, no helper on the + client, budget already spent — says so instead of borrowing an + observation it never made. + + The wait runs on the event-loop poll thread, so it delays the roles + handled later in the same ``poll_once`` pass by up to the budget. That + is the accepted tradeoff: the wait is bounded, only reachable when a + matching Job is mid-deletion, and the alternative is the role + vanishing outright. Mid-deletion is usually a restart, but not only: + ``_job_is_terminating`` does not filter on status, so the TTL + controller reaping a finished prior Job with the same dedupe key + (one-shot Jobs carry ``ttl_seconds_after_finished``) also enters the + wait. That case is harmless and wanted — its pods are already gone so + the wait returns near-instantly, and the 409 protection still applies + to the recycled name. + """ + terminating = [j for j in jobs if _job_is_terminating(j)] + if not terminating: + return + # Partition before anything reports a count: a Job the listing did not + # name is not one we could ever have waited on, so it gets its own line + # rather than being folded into the counts the skip-paths below report. + # + # Defensive: ``KubernetesClient.list_jobs`` always populates ``job_name`` + # (and mirrors it onto ``container_name``), so the unnamed branch is + # unreachable against the production lister — it guards a future/alternate + # backend whose listing is thinner, mirroring the restart route's + # ``addressable=False`` branch. + pending: list[str] = [] + unnamed: list[str] = [] + for job in terminating: + job_name = getattr(job, "job_name", None) or getattr(job, "container_name", None) + if job_name: + pending.append(job_name) + else: + # No name to wait on, but the listing still carries a container id + # (``KubernetesClient.list_jobs`` sets it from the Job's uid). Carry + # it through so this line hands the operator the same actionable + # handle the restart route's counterpart does, rather than a bare + # count they cannot trace back to an object. + unnamed.append(str(getattr(job, "container_id", None) or "")) + if unnamed: + logger.warning( + "Event spawn: teardown wait not performed; terminating Job(s) unobserved", + pipeline_id=pipeline_id, + role=role, + action=action, + dedupe_key=dedupe_key, + terminating=len(unnamed), + container_ids=unnamed, + reason="unaddressable", + ) + if not pending: + return + waiter = getattr(self.k8s, "wait_for_job_gone", None) + if waiter is None: + # Defensive: ``KubernetesClient`` implements ``wait_for_job_gone``, + # so this guards a future/alternate backend. Logged rather than + # returned silently — the spawn proceeds into a window that may 409, + # and the operator should be able to see why nothing waited. + logger.warning( + "Event spawn: teardown wait not performed; terminating Job(s) unobserved", + pipeline_id=pipeline_id, + role=role, + action=action, + dedupe_key=dedupe_key, + terminating=len(pending), + reason="no_wait_helper", + ) + return + logger.info( + "Event spawn: waiting for terminating Job(s) to be reaped before respawn", + pipeline_id=pipeline_id, + role=role, + action=action, + dedupe_key=dedupe_key, + terminating=len(pending), + ) + deadline = _pkg.time.monotonic() + _EVENT_JOB_TERMINATION_WAIT_S + for job_name in pending: + remaining = deadline - _pkg.time.monotonic() + if remaining <= 0: + # No wait ran for this Job, so "still present" is not ours to + # claim — the snapshot said terminating, nothing observed since. + logger.warning( + "Event spawn: teardown wait not performed; terminating Job unobserved", + pipeline_id=pipeline_id, + role=role, + action=action, + dedupe_key=dedupe_key, + job_name=job_name, + reason="budget_exhausted", + ) + continue + try: + gone = bool(waiter(job_name, self._namespace, timeout_s=remaining)) + except Exception as exc: # noqa: BLE001 — the wait is best-effort + logger.warning( + "Failed to wait out a terminating event Job; spawning anyway", + pipeline_id=pipeline_id, + role=role, + dedupe_key=dedupe_key, + job_name=job_name, + error=str(exc), + ) + continue + if not gone: + logger.warning( + "Terminating event Job still present; spawn may 409 and retry next poll", + pipeline_id=pipeline_id, + role=role, + action=action, + dedupe_key=dedupe_key, + job_name=job_name, + ) def create_event_job_status_view(self) -> _pkg._EventJobStatusView: @@ -113,7 +300,9 @@ def spawn_event_job( **Adoption**: requesting a spawn for an already-live dedupe key returns ``None`` (the existing Job is adopted) rather than creating a duplicate — the defense-in-depth backstop for the loop's own dedupe - set racing a restart. + set racing a restart. A Job that is merely *terminating* is not live + (see :meth:`_event_dedupe_key_live`): adopting one produced a role + with no pod at all (#3597), so we wait it out and spawn instead. Everything else (worktree create-with-retry, gateway-session registration) flows through :meth:`spawn_agent_job` unchanged; this @@ -131,7 +320,8 @@ def spawn_event_job( "agent-free, wait is a no-op)." ) - if self._event_dedupe_key_live(dedupe_key): + existing_jobs = self._list_event_jobs(dedupe_key) + if any(_job_is_live(j) for j in existing_jobs): logger.info( "Adopting existing live Job for event (dedupe hit)", pipeline_id=pipeline_id, @@ -141,6 +331,20 @@ def spawn_event_job( ) return None + # #3597: no live Job owns this key, but a *terminating* one may still be + # holding its name. The Job name is deterministic in the dedupe key, so + # creating now would 409 ``AlreadyExists`` against the corpse. Wait for + # the API server to actually reap it (same reasoning as the #2655 + # restart-path wait) before spawning the replacement. + _await_terminating_event_jobs( + self, + existing_jobs, + pipeline_id=pipeline_id, + role=agent_role.value, + action=action, + dedupe_key=dedupe_key, + ) + # --- Attempt worktree re-attach + session reuse --- reuse_worktree_id: str | None = None reuse_repo_volumes: dict[str, str] | None = None diff --git a/orchestrator/mcp_tools/_lifecycle.py b/orchestrator/mcp_tools/_lifecycle.py index 1d0b21aa2..85bb24442 100644 --- a/orchestrator/mcp_tools/_lifecycle.py +++ b/orchestrator/mcp_tools/_lifecycle.py @@ -45,10 +45,22 @@ def _handle_restart_agent(self, args: dict[str, Any]) -> dict[str, Any]: data=data, timeout=60, ) + # Surface the route's teardown/delegation reporting instead of a + # ``container_id`` that is always "" (#3164 deliberately spawns no + # resident pod here, but an empty id reads as a failure — #3597). + # ``respawn`` / ``live_event_loop`` say whether anything will actually + # respawn the role; ``jobs_torn_down`` / ``teardown_confirmed`` + # distinguish "killed a stuck pod and watched it go" from "there was + # nothing to tear down" and from "the delete has not landed yet". + payload = result.get("data", {}) return { "restarted": True, "agent_role": args["agent_role"], - "container_id": result.get("data", {}).get("container_id", ""), + "respawn": payload.get("respawn"), + "live_event_loop": payload.get("live_event_loop"), + "jobs_torn_down": payload.get("jobs_torn_down"), + "teardown_confirmed": payload.get("teardown_confirmed"), + "restart_count": payload.get("restart_count"), "message": f"Agent {args['agent_role']} restarted successfully", } except (TimeoutError, OSError) as e: diff --git a/orchestrator/models/_execution.py b/orchestrator/models/_execution.py index 70ece797b..677a244f7 100644 --- a/orchestrator/models/_execution.py +++ b/orchestrator/models/_execution.py @@ -70,6 +70,15 @@ class ContainerInfo(BaseModel): pod_name: str | None = Field(default=None, description="Kubernetes pod name") namespace: str | None = Field(default=None, description="Kubernetes namespace") job_name: str | None = Field(default=None, description="Kubernetes Job name") + # Set once the object has been asked to go away. Job deletion is + # ASYNCHRONOUS: the API server stamps ``metadata.deletionTimestamp`` and + # the object lingers (status still PENDING/RUNNING) until its dependent + # pods finish terminating. Without this field a Job on its way out is + # indistinguishable from a healthy live one, and the event-loop dedupe + # predicate adopts the corpse instead of respawning (#3597). + deletion_timestamp: datetime | None = Field( + default=None, description="Kubernetes deletionTimestamp (set iff the object is terminating)" + ) @model_validator(mode="before") @classmethod diff --git a/orchestrator/routes/pipelines/_routes_restart.py b/orchestrator/routes/pipelines/_routes_restart.py index ebf47eca8..475e30b3d 100644 --- a/orchestrator/routes/pipelines/_routes_restart.py +++ b/orchestrator/routes/pipelines/_routes_restart.py @@ -9,6 +9,12 @@ import routes.pipelines as _pkg # noqa: E402,F401 +# Total budget for waiting out the asynchronous teardown of the Job(s) +# ``restart_agent`` deletes, shared across all of them (#3597). Deliberately +# well under the MCP client's 60s restart timeout: overrunning it degrades to a +# reported ``teardown_confirmed: false``, never a failed restart. +_JOB_TEARDOWN_WAIT_SECONDS = 20.0 + def _restart_agent_body(pipeline_id: str, agent_role: str) -> tuple[_pkg.Response, int]: """Restart a single agent in a pipeline (orchestrator-native). @@ -24,10 +30,15 @@ def _restart_agent_body(pipeline_id: str, agent_role: str) -> tuple[_pkg.Respons (``check_and_increment_restart_count``); a request over budget is rejected with HTTP 429 before any state is mutated (#3244). 1. Best-effort deletes the role's live one-shot Job(s) (to kill a - stuck pod). One-shot Jobs carry an event-discriminator suffix - in their name, so they are found by label - (``LABEL_PIPELINE_ID`` + ``LABEL_AGENT_ROLE`` [+ ``LABEL_SLICE_ID`` - when slice-scoped]), not by name. + stuck pod) and waits (bounded) for the deletion to be *observed*. + One-shot Jobs carry an event-discriminator suffix in their name, + so they are found by label (``LABEL_PIPELINE_ID`` + + ``LABEL_AGENT_ROLE`` [+ ``LABEL_SLICE_ID`` when slice-scoped]), + not by name. The wait matters because Job deletion is + asynchronous: a Job in ``Terminating`` still reports ``RUNNING``, + and an event-loop poll landing inside that window used to adopt + the corpse on its dedupe-key label and decline to respawn, so the + role silently vanished (#3597). 2. Resets the role's consensus state and health-monitor anchor. 3. Marks the agent record RUNNING with ``container_id = None``. @@ -87,6 +98,10 @@ def _restart_agent_body(pipeline_id: str, agent_role: str) -> tuple[_pkg.Respons "agent_role": "coder", "slice_id": "slice-2", "respawn": "delegated to orchestrator event loop", + "live_event_loop": true, + "arms_invalidated": 1, + "jobs_torn_down": 1, + "teardown_confirmed": true, "restart_count": 1 } } @@ -368,18 +383,40 @@ def _restart_agent_body(pipeline_id: str, agent_role: str) -> tuple[_pkg.Respons } if slice_id is not None: job_labels[_pkg.LABEL_SLICE_ID] = slice_id + removed_jobs = 0 + teardown_confirmed = True try: live_jobs = spawner.k8s.list_containers(labels=job_labels) - removed_jobs = 0 + # ``(name, addressable_by_job_name)`` — the flag is what makes the + # wait below honest; see the fallback branch there. + deleted_names: list[tuple[str, bool]] = [] for job in live_jobs: try: # Mirror the cleanup call sites: prefer the explicit # ``job_name`` (already Job-prefixed), fall back to the # container id which ``remove_agent_job`` -> ``remove_container`` # resolves to a Job name. - spawner.remove_agent_job(job.job_name or job.container_id, force=True) + job_name = job.job_name + target = job_name or job.container_id + spawner.remove_agent_job(target, force=True) removed_jobs += 1 + deleted_names.append((target, bool(job_name))) except Exception as job_err: # noqa: BLE001 - best-effort teardown + # The delete never landed, so this Job is neither torn down nor + # waited on: it is still live (and NOT terminating), which means + # the next event-loop poll adopts it on its dedupe-key label and + # no respawn happens. Leaving the flag ``True`` here would report + # the operator's no-op restart as the payload's "there was + # nothing to tear down" case — the one branch in this route that + # OVER-claims. Clear it: we did not observe the Job gone. + # + # A benign already-gone race (the TTL controller reaps the Job + # between ``list_containers`` and the delete) lands here too, as + # a 404 for which ``True`` would have been correct. We cannot + # discriminate at this layer — ``remove_container`` re-wraps + # every failure into ``JobOperationError`` — so we under-claim + # on both, consistent with the rest of the field. + teardown_confirmed = False _pkg.logger.warning( "Failed to delete live one-shot Job during restart (best-effort)", pipeline_id=pipeline_id, @@ -388,14 +425,133 @@ def _restart_agent_body(pipeline_id: str, agent_role: str) -> tuple[_pkg.Respons job_name=getattr(job, "job_name", None), error=str(job_err), ) + # #3597: the delete above is ASYNCHRONOUS — it returns as soon as the + # API server accepts it, and the Job then lingers in ``Terminating`` + # (still reporting RUNNING) until its pods are gone. The event loop + # polls every ~5s, so a poll landing inside that window used to match + # the corpse on its dedupe-key label, adopt it, and decline to spawn a + # replacement: the role silently vanished with the pipeline still + # reporting ``running``. The adoption filter now excludes terminating + # Jobs, but we also wait for the teardown we requested to be OBSERVED + # before returning, so the respawn we are delegating starts from a + # clean slate and cannot 409 on the recycled Job name. Bounded by a + # deadline shared across every deleted Job (the MCP client's restart + # timeout is 60s) and best-effort: a timeout is reported, not fatal. + deadline = _pkg.time.monotonic() + _JOB_TEARDOWN_WAIT_SECONDS + # Partition first, so each report covers only the Jobs it is actually + # about: an entry the listing gave us no ``job_name`` for could never + # have been waited on, whatever the backend, so it must not be folded + # into the helper-less count below. + # + # The listing carried no ``job_name``, so the only handle we have is + # the container id. ``wait_for_job_gone`` normalizes it into a Job name + # that never existed, 404s on the first read, and reports "gone" + # without having observed the real teardown. Report it unconfirmed + # instead of claiming an observation we never made. + # + # Defensive: ``KubernetesClient.list_containers`` always populates + # ``job_name`` (it derives the name from ``LABEL_CONTAINER_NAME``, + # which ``create_container`` applies to the pod template), so this is + # unreachable against the production lister — it guards a + # future/alternate backend whose listing is thinner. + pending_waits = [name for name, addressable in deleted_names if addressable] + for name, addressable in deleted_names: + if addressable: + continue + teardown_confirmed = False + _pkg.logger.warning( + "restart_agent: deleted Job carried no job_name; its teardown cannot be observed", + pipeline_id=pipeline_id, + agent_role=agent_role, + slice_id=slice_id, + container_id=name, + reason="unaddressable", + ) + # ``getattr`` so a backend without the wait helper reports an + # unconfirmed teardown rather than raising into the list-failure + # handler below (which would log a misleading "failed to list"). + # + # Defensive, and loop-invariant: ``KubernetesClient`` implements + # ``wait_for_job_gone``, so this is unreachable against the production + # client for the same reason as the ``addressable=False`` branch above + # — it guards a future/alternate backend. Checked ONCE, above the loop: + # a helper-less backend is a single backend-capability fact, not N + # per-Job teardown failures, so it earns one log line. + waiter = getattr(spawner.k8s, "wait_for_job_gone", None) + if pending_waits and waiter is None: + teardown_confirmed = False + _pkg.logger.warning( + "restart_agent: teardown wait not performed; teardown unobserved", + pipeline_id=pipeline_id, + agent_role=agent_role, + slice_id=slice_id, + jobs=len(pending_waits), + reason="no_wait_helper", + ) + pending_waits = [] + for name in pending_waits: + # Each branch below that does NOT complete a wait clears the flag + # and ``continue``s, mirroring ``_await_terminating_event_jobs``. + # Falling through to the "still terminating" warning would claim an + # observation the code never made — that message is only supportable + # after a wait actually ran and reported the Job still present. + remaining = deadline - _pkg.time.monotonic() + if remaining <= 0: + teardown_confirmed = False + _pkg.logger.warning( + "restart_agent: teardown wait not performed; teardown unobserved", + pipeline_id=pipeline_id, + agent_role=agent_role, + slice_id=slice_id, + job_name=name, + reason="budget_exhausted", + ) + continue + try: + gone = bool(waiter(name, spawner.k8s.namespace, timeout_s=remaining)) + except Exception as wait_err: # noqa: BLE001 - the wait is best-effort + # Handled locally, mirroring the event-loop path's + # ``_await_terminating_event_jobs``: a raising waiter is an + # unconfirmed teardown, not a listing failure. Letting it + # reach the outer handler would log the misleading + # "Failed to list live one-shot Jobs". (``wait_for_job_gone`` + # swallows its own exceptions today, so this is future-proofing.) + teardown_confirmed = False + _pkg.logger.warning( + "restart_agent: teardown wait raised; treating the teardown as unconfirmed", + pipeline_id=pipeline_id, + agent_role=agent_role, + slice_id=slice_id, + job_name=name, + error=str(wait_err), + ) + continue + if not gone: + # This also fires under a benign race: once the corpse is + # reaped, the event loop can recreate the SAME deterministic + # Job name inside this window, so the wait sees a live Job and + # times out even though the respawn already succeeded. + # ``teardown_confirmed: false`` therefore means "not observed + # gone", never "the restart failed" — it under-claims by design. + teardown_confirmed = False + _pkg.logger.warning( + "restart_agent: Job still terminating after teardown wait; " + "the event loop's respawn may be delayed a poll", + pipeline_id=pipeline_id, + agent_role=agent_role, + slice_id=slice_id, + job_name=name, + ) _pkg.logger.info( "restart_agent: deleted live one-shot Job(s) for role", pipeline_id=pipeline_id, agent_role=agent_role, slice_id=slice_id, removed=removed_jobs, + teardown_confirmed=teardown_confirmed, ) except Exception as list_err: # noqa: BLE001 - best-effort teardown + teardown_confirmed = False _pkg.logger.warning( "Failed to list live one-shot Jobs during restart (best-effort)", pipeline_id=pipeline_id, @@ -632,6 +788,20 @@ def _restart_agent_body(pipeline_id: str, agent_role: str) -> tuple[_pkg.Respons "respawn": respawn_note, "live_event_loop": live_loop_found, "arms_invalidated": invalidated_keys, + # Legible teardown (#3597). ``jobs_torn_down: 0`` with + # ``teardown_confirmed: true`` says "there was nothing to kill" (the + # role had already exited); a non-zero count with + # ``teardown_confirmed: true`` says "the pod is gone and the respawn + # starts clean". ``teardown_confirmed: false`` means the teardown was + # not observed to complete — either a delete failed outright (so the + # Job may still be live and no respawn will follow) or the delete + # landed but did not finish within the route's budget (the respawn may + # take an extra poll). It is a deliberate under-claim, not by itself a + # failure signal: it also fires when the event loop recreates the same + # deterministic Job name inside the wait window, i.e. when the respawn + # has in fact already succeeded. + "jobs_torn_down": removed_jobs, + "teardown_confirmed": teardown_confirmed, "restart_count": new_restart_count, } if fresh_session: diff --git a/orchestrator/tests/test_ble001_narrowing_audit.py b/orchestrator/tests/test_ble001_narrowing_audit.py index e29635e14..3768dbcf0 100644 --- a/orchestrator/tests/test_ble001_narrowing_audit.py +++ b/orchestrator/tests/test_ble001_narrowing_audit.py @@ -168,7 +168,12 @@ def test_audit_window_retains_documented_ble001_population() -> None: # Upper bound guards against a future PR re-introducing swallow-all handlers # en masse without re-running the audit. The file has carried ~80 documented # swallows through the overhaul; a jump well past that needs an audit pass. - assert len(noqa_lines) <= 120, ( + # Raised 120 -> 121 by #3597, which added one audited site: the restart + # route's bounded teardown wait catches a raising ``wait_for_job_gone`` so + # it degrades to ``teardown_confirmed: false`` instead of falling through to + # the outer handler and logging a misleading "failed to list" — mirroring + # ``_await_terminating_event_jobs`` on the event-loop path. + assert len(noqa_lines) <= 121, ( f"Found {len(noqa_lines)} ``# noqa: BLE001`` swallows in " f"routes/pipelines.py, well past the documented population — a future " f"PR appears to have re-introduced swallow-all handlers without " diff --git a/orchestrator/tests/test_kubernetes_client.py b/orchestrator/tests/test_kubernetes_client.py index d2fbe5d1a..011287b99 100644 --- a/orchestrator/tests/test_kubernetes_client.py +++ b/orchestrator/tests/test_kubernetes_client.py @@ -84,12 +84,14 @@ def _make_mock_job( active: int | None = None, start_time: datetime | None = None, completion_time: datetime | None = None, + deletion_timestamp: datetime | None = None, ) -> MagicMock: """Create a mock Job object matching the k8s SDK shape.""" job = MagicMock() job.metadata.name = name job.metadata.uid = uid job.metadata.labels = labels or {LABEL_ORCHESTRATOR: "true"} + job.metadata.deletion_timestamp = deletion_timestamp job.status.succeeded = succeeded job.status.failed = failed job.status.active = active @@ -1572,6 +1574,38 @@ def test_list_jobs_with_status_mapping( assert jobs[2].status == ContainerStatus.RUNNING assert jobs[3].status == ContainerStatus.PENDING + def test_list_jobs_reports_deletion_timestamp( + self, + k8s_client: KubernetesClient, + mock_batch_api: MagicMock, + ): + """A Terminating Job is reported as such, not just as RUNNING (#3597). + + Job deletion is asynchronous: the API server stamps + ``metadata.deletionTimestamp`` and the object keeps reporting + ``active > 0`` (⇒ RUNNING) until its pods finish terminating. Status + alone therefore cannot distinguish a live Job from one on its way + out, which is what let the event loop adopt a corpse and silently + drop the role. Surfacing the stamp is what makes that distinction + possible for callers. + """ + stamp = datetime(2026, 7, 25, 1, 49, 8, tzinfo=UTC) + terminating = _make_mock_job( + name="j-terminating", uid="uid-t", active=1, deletion_timestamp=stamp + ) + healthy = _make_mock_job(name="j-live", uid="uid-l", active=1) + + mock_result = MagicMock() + mock_result.items = [terminating, healthy] + mock_batch_api.list_namespaced_job.return_value = mock_result + + jobs = k8s_client.list_jobs("test-ns") + + # Both still report RUNNING — that is exactly the ambiguity. + assert [j.status for j in jobs] == [ContainerStatus.RUNNING, ContainerStatus.RUNNING] + assert jobs[0].deletion_timestamp == stamp + assert jobs[1].deletion_timestamp is None + def test_list_jobs_with_label_selector( self, k8s_client: KubernetesClient, diff --git a/orchestrator/tests/test_kubernetes_spawner.py b/orchestrator/tests/test_kubernetes_spawner.py index 249a895bc..ceecd3c63 100644 --- a/orchestrator/tests/test_kubernetes_spawner.py +++ b/orchestrator/tests/test_kubernetes_spawner.py @@ -2389,6 +2389,255 @@ def test_same_dedupe_key_adopts_existing_job(self, spawner, mock_k8s_client, moc # create_container was NOT called a second time — the Job was adopted. assert mock_k8s_client.create_container.call_count == 1 + @staticmethod + def _terminating_job(name="egg-agent-pipe-1-slice-2-coder-ev"): + """A Job that has been deleted but has not gone away yet. + + Deliberately ``RUNNING``: that is the whole point — a Job under + deletion keeps reporting its pre-delete status until its pods finish + terminating, so only ``deletion_timestamp`` distinguishes it. + """ + return ContainerInfo( + container_id="uid-terminating", + container_name=name, + job_name=name, + namespace="test-ns", + status=ContainerStatus.RUNNING, + deletion_timestamp=datetime(2026, 7, 25, 1, 49, 8, tzinfo=UTC), + ) + + def test_terminating_job_is_not_adopted(self, spawner, mock_k8s_client): + """A Job under deletion must NOT be adopted (#3597). + + The incident: ``restart_agent`` deletes the role's one-shot Job and + delegates the respawn to the event loop. Deletion is asynchronous, so + for a few seconds the Job sits in ``Terminating`` still reporting + ``RUNNING``. The next poll (~5s) matched it on the dedupe-key label, + adopted it, and declined to spawn a replacement — then the adopted Job + finished terminating. Net result: no pod, no Job, and because + adoption re-arms the key in the loop's live set (where a missing Job + reads as "still running"), the role stayed vanished indefinitely with + the pipeline still reporting ``status: running``. + """ + mock_k8s_client.list_jobs.return_value = [self._terminating_job()] + mock_k8s_client.wait_for_job_gone.return_value = True + + spawner.spawn_event_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + action="propose", + dedupe_key=self._KEY, + slice_id="slice-2", + phase="implement", + repos=["owner/repo"], + ) + + assert mock_k8s_client.create_container.call_count == 1, ( + "a terminating Job must not be adopted — the role would vanish" + ) + + def test_terminating_job_is_waited_out_before_respawn(self, spawner, mock_k8s_client): + """The replacement waits for the deleted Job's name to be free (#3597). + + A one-shot Job's name is derived from the dedupe key, so the + replacement carries the SAME name as the Job being torn down. + Creating into the deletion window returns 409 ``AlreadyExists`` (the + #2655 race), so the spawn waits the corpse out first. + """ + mock_k8s_client.list_jobs.return_value = [self._terminating_job()] + mock_k8s_client.wait_for_job_gone.return_value = True + call_order: list[str] = [] + mock_k8s_client.wait_for_job_gone.side_effect = lambda *a, **k: ( + call_order.append("wait"), + True, + )[1] + created = mock_k8s_client.create_container.return_value + mock_k8s_client.create_container.side_effect = lambda **kw: ( + call_order.append("create"), + created, + )[1] + + spawner.spawn_event_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + action="propose", + dedupe_key=self._KEY, + slice_id="slice-2", + phase="implement", + repos=["owner/repo"], + ) + + assert call_order == ["wait", "create"] + waited_name = mock_k8s_client.wait_for_job_gone.call_args.args[0] + assert waited_name == "egg-agent-pipe-1-slice-2-coder-ev" + + def test_spawn_proceeds_when_terminating_job_outlives_the_wait(self, spawner, mock_k8s_client): + """A wait timeout is reported, never fatal (#3597). + + Overrunning the bounded wait degrades to "the create may 409 and the + event loop retries next poll", which is recoverable; refusing to + spawn would reproduce the silent-vanish this path exists to prevent. + """ + mock_k8s_client.list_jobs.return_value = [self._terminating_job()] + mock_k8s_client.wait_for_job_gone.return_value = False + + spawner.spawn_event_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + action="propose", + dedupe_key=self._KEY, + slice_id="slice-2", + phase="implement", + repos=["owner/repo"], + ) + + assert mock_k8s_client.create_container.call_count == 1 + + def test_spent_wait_budget_is_not_reported_as_an_observation(self, spawner, mock_k8s_client): + """A Job the wait never ran for is "unobserved", not "still present" (#3597). + + "Terminating event Job still present" asserts a wait ran and found + the Job there. When the shared budget is already spent, no wait ran + at all, so that message is unsupportable — the same taxonomy the + restart route applies on its side of this fix. + """ + mock_k8s_client.list_jobs.return_value = [self._terminating_job()] + # A waiter that WOULD confirm, to prove the budget check short-circuits + # before it rather than the wait quietly succeeding. + mock_k8s_client.wait_for_job_gone.return_value = True + + with ( + patch("kubernetes_spawner._events._EVENT_JOB_TERMINATION_WAIT_S", 0.0), + patch("kubernetes_spawner._events.logger") as mock_logger, + ): + spawner.spawn_event_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + action="propose", + dedupe_key=self._KEY, + slice_id="slice-2", + phase="implement", + repos=["owner/repo"], + ) + + # Exhausting the budget never blocks the spawn — that is the whole point. + assert mock_k8s_client.create_container.call_count == 1 + mock_k8s_client.wait_for_job_gone.assert_not_called() + calls = mock_logger.warning.call_args_list + messages = [c.args[0] for c in calls if c.args] + assert any("teardown wait not performed" in m for m in messages) + assert [c.kwargs.get("reason") for c in calls] == ["budget_exhausted"] + assert not any("still present" in m for m in messages) + + def test_missing_wait_helper_is_logged_not_silently_skipped(self, spawner, mock_k8s_client): + """A backend without the wait helper says so (#3597). + + The docstring promised "on timeout (or a k8s client without the wait + helper) we log and let the spawn proceed"; the no-helper arm returned + bare, so the spawn walked into a possible 409 with nothing in the log + explaining why nothing waited. + """ + mock_k8s_client.list_jobs.return_value = [self._terminating_job()] + # ``getattr(..., None)`` only yields None if the attribute is really + # absent — a bare MagicMock would hand back an auto-attribute. + del mock_k8s_client.wait_for_job_gone + + with patch("kubernetes_spawner._events.logger") as mock_logger: + spawner.spawn_event_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + action="propose", + dedupe_key=self._KEY, + slice_id="slice-2", + phase="implement", + repos=["owner/repo"], + ) + + assert mock_k8s_client.create_container.call_count == 1 + calls = mock_logger.warning.call_args_list + assert [c.kwargs.get("reason") for c in calls] == ["no_wait_helper"] + assert not any("still present" in c.args[0] for c in calls if c.args) + + def test_unnamed_terminating_job_is_reported_not_silently_skipped( + self, spawner, mock_k8s_client + ): + """A Job the listing did not name is a third no-observation path (#3597). + + It skipped the wait with no log at all, so the taxonomy the route + applies held on two of the three skip-paths. It also must not be + folded into the counts the other two report — nothing could ever have + waited on it, whatever the backend's capabilities. + """ + named = self._terminating_job() + unnamed = ContainerInfo( + container_id="uid-unnamed", + container_name="", + job_name=None, + namespace="test-ns", + status=ContainerStatus.RUNNING, + deletion_timestamp=datetime(2026, 7, 25, 1, 49, 8, tzinfo=UTC), + ) + mock_k8s_client.list_jobs.return_value = [named, unnamed] + mock_k8s_client.wait_for_job_gone.return_value = True + + with patch("kubernetes_spawner._events.logger") as mock_logger: + spawner.spawn_event_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + action="propose", + dedupe_key=self._KEY, + slice_id="slice-2", + phase="implement", + repos=["owner/repo"], + ) + + assert mock_k8s_client.create_container.call_count == 1 + warnings = mock_logger.warning.call_args_list + assert [c.kwargs.get("reason") for c in warnings] == ["unaddressable"] + assert warnings[0].kwargs["terminating"] == 1 + # A bare count is not actionable: carry the id through so an operator + # grepping ``reason=unaddressable`` gets the same handle the restart + # route's counterpart line reports. + assert warnings[0].kwargs["container_ids"] == ["uid-unnamed"] + # Only the nameable Job was ever waitable, so only it is counted. + infos = [ + c for c in mock_logger.info.call_args_list if c.args and "waiting for" in c.args[0] + ] + assert len(infos) == 1 + assert infos[0].kwargs["terminating"] == 1 + mock_k8s_client.wait_for_job_gone.assert_called_once() + assert mock_k8s_client.wait_for_job_gone.call_args.args[0] == named.job_name + + def test_live_job_is_adopted_without_any_wait(self, spawner, mock_k8s_client): + """The unchanged common path: a genuinely live Job is still adopted. + + Guards the #3597 fix against over-reach — only a Job carrying a + deletion stamp loses adoptability; a healthy RUNNING one must still + suppress the duplicate spawn (and must not pay for a wait). + """ + live = ContainerInfo( + container_id="uid-live", + container_name="egg-agent-pipe-1-slice-2-coder-ev", + job_name="egg-agent-pipe-1-slice-2-coder-ev", + namespace="test-ns", + status=ContainerStatus.RUNNING, + ) + mock_k8s_client.list_jobs.return_value = [live] + + result = spawner.spawn_event_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + action="propose", + dedupe_key=self._KEY, + slice_id="slice-2", + phase="implement", + repos=["owner/repo"], + ) + + assert result is None, "a live Job must still be adopted" + mock_k8s_client.create_container.assert_not_called() + mock_k8s_client.wait_for_job_gone.assert_not_called() + def test_terminated_job_does_not_block_respawn(self, spawner, mock_k8s_client): """A label-matching but TERMINATED Job (EXITED/FAILED) lingering under the finished-TTL must NOT be adopted — a re-derived identical event @@ -2675,10 +2924,29 @@ def delete_job(self, name, namespace=None, **kwargs): # Idempotent pre-spawn cleanup; our generated names never collide. self.jobs = [j for j in self.jobs if j.job_name != name] + def wait_for_job_gone(self, name, namespace=None, timeout_s=0.0): + """Model the reap completing: the Terminating Job finally disappears.""" + before = len(self.jobs) + self.jobs = [j for j in self.jobs if j.job_name != name] + return len(self.jobs) < before or before == 0 + # --- test helpers ----------------------------------------------------- def crash_all(self): self.jobs = [j.model_copy(update={"status": ContainerStatus.FAILED}) for j in self.jobs] + def begin_delete_all(self): + """Model an ACCEPTED but not-yet-complete k8s delete (#3597). + + This is what ``restart_agent``'s teardown looks like from the event + loop's side for the seconds that follow: the Job is stamped with a + ``deletionTimestamp`` and keeps reporting its pre-delete status + until its pods finish terminating. + """ + self.jobs = [ + j.model_copy(update={"deletion_timestamp": datetime(2026, 7, 25, 1, 49, 8, tzinfo=UTC)}) + for j in self.jobs + ] + @property def names(self): return [j.job_name for j in self.jobs] @@ -2719,6 +2987,7 @@ def _wire(self, store, mock_k8s_client): mock_k8s_client.create_container.side_effect = store.create_container mock_k8s_client.remove_container.side_effect = store.remove_container mock_k8s_client.delete_job.side_effect = store.delete_job + mock_k8s_client.wait_for_job_gone.side_effect = store.wait_for_job_gone def test_crash_then_respawn_creates_a_fresh_job(self, spawner, mock_k8s_client, mock_gateway): import event_loop @@ -2797,6 +3066,40 @@ def test_terminal_job_alone_does_not_block_respawn(self, spawner, mock_k8s_clien assert corpse_name not in store.names assert store.statuses == [ContainerStatus.RUNNING] + def test_restart_deleted_job_mid_termination_respawns(self, spawner, mock_k8s_client): + """The restart race, end to end against the real spawner (#3597). + + ``restart_agent`` deletes the role's live Job and delegates the + respawn to the event loop. k8s deletion is asynchronous, so for the + next few seconds the Job is Terminating but still reports RUNNING. + The loop's very next poll re-derives the same event — and used to + adopt that corpse, create nothing, and leave the role with no pod + and no Job while the pipeline still reported ``running``. + + Drives the real ``spawn_event_job`` against the stateful Job store so + the adoption filter, the terminating-Job wait, and Job creation are + exercised together rather than through a fake that always spawns. + """ + store = _StatefulEventJobs() + self._wire(store, mock_k8s_client) + + # 1. The role has a live one-shot Job. + assert self._spawn(spawner) is not None + assert store.statuses == [ContainerStatus.RUNNING] + + # 2. restart_agent deletes it; the delete is accepted but not complete. + store.begin_delete_all() + assert store.statuses == [ContainerStatus.RUNNING], "still reports RUNNING" + + # 3. The event loop's next poll lands inside the deletion window. + assert self._spawn(spawner) is not None, "the respawn must not adopt a corpse" + + # A real replacement exists, and the corpse was waited out first so the + # create could not 409 on the recycled Job name. + assert mock_k8s_client.create_container.call_count == 2 + assert store.statuses == [ContainerStatus.RUNNING] + assert all(j.deletion_timestamp is None for j in store.jobs) + def test_live_job_still_blocks_respawn(self, spawner, mock_k8s_client): """Regression guard: a genuinely RUNNING Job for the key is still adopted (no duplicate pod) — fix #1 narrows adoption to live Jobs, it diff --git a/orchestrator/tests/test_restart_agent.py b/orchestrator/tests/test_restart_agent.py index 863b6148b..d973d8364 100644 --- a/orchestrator/tests/test_restart_agent.py +++ b/orchestrator/tests/test_restart_agent.py @@ -746,6 +746,541 @@ def test_restart_agent_failed_pipeline( # Dead event loop is restarted by relaunching the driver thread (#3244). mock_spawn_thread.assert_called_once() + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_waits_for_deleted_job_to_actually_be_gone( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, client + ): + """Teardown is observed, not merely requested (#3597). + + Job deletion is asynchronous: ``remove_agent_job`` returns as soon + as the API server accepts it, and the Job then lingers in + ``Terminating`` — still reporting RUNNING — until its pods are gone. + The event loop polls every ~5s, so returning immediately let the + next poll match the corpse on its dedupe-key label, adopt it, and + decline to respawn: the role silently vanished. The route must wait + for the delete it issued to be observed before handing the respawn + to the loop, and say so in its response. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + live_job = MagicMock() + live_job.job_name = "egg-agent-issue-100-coder-15de0e94" + live_job.container_id = "uid-1" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [live_job] + mock_spawner.k8s.namespace = "egg-agents" + mock_spawner.k8s.wait_for_job_gone.return_value = True + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + live_loop = MagicMock() + live_loop.slice_id = None + live_loop.invalidate_role_arms.return_value = [] + + with patch("event_loop.get_live_event_loops", return_value=[live_loop]): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + mock_spawner.remove_agent_job.assert_called_once_with( + "egg-agent-issue-100-coder-15de0e94", force=True + ) + # The deletion we issued was waited out before returning. + wait_call = mock_spawner.k8s.wait_for_job_gone.call_args + assert wait_call.args[0] == "egg-agent-issue-100-coder-15de0e94" + assert wait_call.kwargs["timeout_s"] > 0 + data = response.get_json()["data"] + assert data["jobs_torn_down"] == 1 + assert data["teardown_confirmed"] is True + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_reports_unconfirmed_teardown_instead_of_claiming_success( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """A teardown that outlives the wait budget is reported, not hidden (#3597). + + The operator-facing complaint in the incident was that the working + and the vanished case were indistinguishable. A Job that is still + terminating when the route gives up is exactly the case where the + respawn may be delayed, so it must not be reported as a clean + restart. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + live_job = MagicMock() + live_job.job_name = "egg-agent-issue-100-coder-15de0e94" + live_job.container_id = "uid-1" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [live_job] + mock_spawner.k8s.namespace = "egg-agents" + mock_spawner.k8s.wait_for_job_gone.return_value = False + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with patch("event_loop.get_live_event_loops", return_value=[]): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + data = response.get_json()["data"] + assert data["jobs_torn_down"] == 1 + assert data["teardown_confirmed"] is False + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_with_nothing_to_tear_down_says_so( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """ "Nothing to kill" is distinguishable from "killed it" (#3597).""" + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [] + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with patch("event_loop.get_live_event_loops", return_value=[]): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + data = response.get_json()["data"] + assert data["jobs_torn_down"] == 0 + # Nothing was deleted, so there is nothing outstanding to wait for. + assert data["teardown_confirmed"] is True + mock_spawner.k8s.wait_for_job_gone.assert_not_called() + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_does_not_report_a_failed_delete_as_a_clean_teardown( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """A delete that RAISED is not "there was nothing to tear down" (#3597). + + The delete is best-effort and its failure is swallowed, so + ``jobs_torn_down`` stays 0 — byte-identical to the genuine clean-exit + case above. The Job is in fact still live and NOT terminating, so the + next event-loop poll adopts it and no respawn follows: the operator's + restart was a complete no-op. ``teardown_confirmed`` is the only field + that can carry that, so it must not stay ``True``. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + live_job = MagicMock() + live_job.job_name = "egg-agent-issue-100-coder-aaaaaaaa" + live_job.container_id = "uid-1" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [live_job] + mock_spawner.k8s.namespace = "egg-agents" + mock_spawner.remove_agent_job.side_effect = JobOperationError("apiserver rejected delete") + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with patch("event_loop.get_live_event_loops", return_value=[]): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + mock_spawner.remove_agent_job.assert_called_once_with( + "egg-agent-issue-100-coder-aaaaaaaa", force=True + ) + data = response.get_json()["data"] + # The failure is swallowed, so the count cannot distinguish this case. + assert data["jobs_torn_down"] == 0 + # ... which is exactly why the flag has to. + assert data["teardown_confirmed"] is False + # Nothing was deleted, so there is nothing to wait on either. + mock_spawner.k8s.wait_for_job_gone.assert_not_called() + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_does_not_claim_teardown_it_could_not_observe( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """A Job with no ``job_name`` is unobservable, not confirmed (#3597). + + The fallback handle is the container id (a uid). Waiting on it reads + a Job name that never existed, so the read 404s on the first poll and + ``wait_for_job_gone`` returns True immediately — "gone" without any + observation of the real teardown. The route must not launder that + into ``teardown_confirmed: true``. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + live_job = MagicMock() + live_job.job_name = None + live_job.container_id = "uid-1" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [live_job] + mock_spawner.k8s.namespace = "egg-agents" + # The uid-as-name read 404s, so a real waiter reports "gone". + mock_spawner.k8s.wait_for_job_gone.return_value = True + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with patch("event_loop.get_live_event_loops", return_value=[]): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + mock_spawner.remove_agent_job.assert_called_once_with("uid-1", force=True) + data = response.get_json()["data"] + assert data["jobs_torn_down"] == 1 + assert data["teardown_confirmed"] is False + # No point burning the budget on a name that cannot 404 meaningfully. + mock_spawner.k8s.wait_for_job_gone.assert_not_called() + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_handles_a_raising_teardown_waiter_locally( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """A waiter that raises is an unconfirmed teardown, not a list failure (#3597). + + Mirrors ``_await_terminating_event_jobs`` on the event-loop path: the + raise is caught at the wait, so the rest of the deleted Jobs are still + waited on and the outer handler's "Failed to list live one-shot Jobs" + never fires for what is really a wait failure. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + first = MagicMock() + first.job_name = "egg-agent-issue-100-coder-aaaaaaaa" + first.container_id = "uid-1" + second = MagicMock() + second.job_name = "egg-agent-issue-100-coder-bbbbbbbb" + second.container_id = "uid-2" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [first, second] + mock_spawner.k8s.namespace = "egg-agents" + mock_spawner.k8s.wait_for_job_gone.side_effect = [ + RuntimeError("apiserver blew up"), + True, + ] + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with patch("event_loop.get_live_event_loops", return_value=[]): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + # The raise did not abort the loop: the second Job was still waited on. + assert mock_spawner.k8s.wait_for_job_gone.call_count == 2 + data = response.get_json()["data"] + # Both deletes landed — the raise is a wait failure, not a list failure. + assert data["jobs_torn_down"] == 2 + assert data["teardown_confirmed"] is False + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_does_not_claim_still_terminating_without_a_wait( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """A no-observation path must not log "still terminating" (#3597). + + "Job still terminating after teardown wait" asserts the wait ran and + found the Job present. When the waiter raises, no observation was made + at all, so that message is unsupportable — the raise branch must clear + ``teardown_confirmed`` and move on rather than falling through to it. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + live_job = MagicMock() + live_job.job_name = "egg-agent-issue-100-coder-aaaaaaaa" + live_job.container_id = "uid-1" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [live_job] + mock_spawner.k8s.namespace = "egg-agents" + mock_spawner.k8s.wait_for_job_gone.side_effect = RuntimeError("apiserver blew up") + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with ( + patch("event_loop.get_live_event_loops", return_value=[]), + patch("routes.pipelines.logger") as mock_logger, + ): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + assert response.get_json()["data"]["teardown_confirmed"] is False + warnings = [call.args[0] for call in mock_logger.warning.call_args_list if call.args] + assert any("teardown wait raised" in msg for msg in warnings) + # The wait never observed the Job, so the route cannot claim it did. + assert not any("still terminating after teardown wait" in msg for msg in warnings) + # And it is a wait failure, not a listing failure. + assert not any("Failed to list live one-shot Jobs" in msg for msg in warnings) + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_reports_a_missing_wait_helper_once( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """A backend without the wait helper is one fact, logged once (#3597). + + The capability is loop-invariant, so checking it per deleted Job would + emit N identical warnings for a single backend limitation. And it is a + no-observation path: the teardown must be reported unconfirmed, not + borrowed from the "still terminating" message that implies a wait ran. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + first = MagicMock() + first.job_name = "egg-agent-issue-100-coder-aaaaaaaa" + first.container_id = "uid-1" + second = MagicMock() + second.job_name = "egg-agent-issue-100-coder-bbbbbbbb" + second.container_id = "uid-2" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [first, second] + mock_spawner.k8s.namespace = "egg-agents" + # ``getattr(..., None)`` only yields None when the attribute is really + # absent — a bare MagicMock would hand back an auto-attribute. + del mock_spawner.k8s.wait_for_job_gone + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with ( + patch("event_loop.get_live_event_loops", return_value=[]), + patch("routes.pipelines.logger") as mock_logger, + ): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + data = response.get_json()["data"] + # Both deletes landed; only the observation is missing. + assert data["jobs_torn_down"] == 2 + assert data["teardown_confirmed"] is False + calls = mock_logger.warning.call_args_list + not_performed = [c for c in calls if c.args and "teardown wait not performed" in c.args[0]] + assert len(not_performed) == 1, "one backend-capability fact, one log line" + assert not_performed[0].kwargs["reason"] == "no_wait_helper" + assert not_performed[0].kwargs["jobs"] == 2 + messages = [c.args[0] for c in calls if c.args] + assert not any("still terminating after teardown wait" in m for m in messages) + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_does_not_fold_an_unaddressable_job_into_the_helper_count( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """Each no-observation report covers only the Jobs it is about (#3597). + + A Job the listing gave no ``job_name`` for could never have been + waited on, whatever the backend's capabilities. Checking the wait + helper first swallowed it: the ``jobs=`` count claimed it as one the + helper's absence cost us, and its own line never fired at all. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + named = MagicMock() + named.job_name = "egg-agent-issue-100-coder-aaaaaaaa" + named.container_id = "uid-1" + unnamed = MagicMock() + unnamed.job_name = None + unnamed.container_id = "uid-2" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [named, unnamed] + mock_spawner.k8s.namespace = "egg-agents" + del mock_spawner.k8s.wait_for_job_gone + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with ( + patch("event_loop.get_live_event_loops", return_value=[]), + patch("routes.pipelines.logger") as mock_logger, + ): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + data = response.get_json()["data"] + assert data["jobs_torn_down"] == 2 + assert data["teardown_confirmed"] is False + calls = mock_logger.warning.call_args_list + unaddressable = [c for c in calls if c.kwargs.get("reason") == "unaddressable"] + assert len(unaddressable) == 1, "the unnameable Job reports for itself" + assert unaddressable[0].kwargs["container_id"] == "uid-2" + not_performed = [c for c in calls if c.kwargs.get("reason") == "no_wait_helper"] + assert len(not_performed) == 1 + assert not_performed[0].kwargs["jobs"] == 1, "only the Job the helper would have waited on" + + @patch("routes.pipelines._spawn_pipeline_run_thread") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_restart_reports_a_spent_wait_budget_as_unobserved( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, mock_spawn_thread, client + ): + """A Job the wait never ran for is unobserved, not "still terminating" (#3597). + + The budget is shared across every deleted Job, so a slow first wait can + leave later Jobs with nothing left. No wait ran for them, so the route + must not claim it looked and found them present. + """ + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + mock_resolve.return_value = (mock_store, pipeline) + + live_job = MagicMock() + live_job.job_name = "egg-agent-issue-100-coder-aaaaaaaa" + live_job.container_id = "uid-1" + + mock_spawner = MagicMock() + mock_spawner.k8s.list_containers.return_value = [live_job] + mock_spawner.k8s.namespace = "egg-agents" + # A waiter that WOULD confirm, so a green result could only come from + # the budget check failing to short-circuit. + mock_spawner.k8s.wait_for_job_gone.return_value = True + mock_spawner.check_and_increment_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + with ( + patch("event_loop.get_live_event_loops", return_value=[]), + patch("routes.pipelines._routes_restart._JOB_TEARDOWN_WAIT_SECONDS", 0.0), + patch("routes.pipelines.logger") as mock_logger, + ): + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={}, + ) + + assert response.status_code == 200 + assert response.get_json()["data"]["teardown_confirmed"] is False + mock_spawner.k8s.wait_for_job_gone.assert_not_called() + calls = mock_logger.warning.call_args_list + not_performed = [c for c in calls if c.args and "teardown wait not performed" in c.args[0]] + assert len(not_performed) == 1 + assert not_performed[0].kwargs["reason"] == "budget_exhausted" + assert not_performed[0].kwargs["job_name"] == "egg-agent-issue-100-coder-aaaaaaaa" + messages = [c.args[0] for c in calls if c.args] + assert not any("still terminating after teardown wait" in m for m in messages) + @patch("routes.pipelines.get_pipeline_state_lock") @patch("routes.pipelines.get_container_spawner") @patch("routes.pipelines._resolve_pipeline") diff --git a/orchestrator/tests/test_restart_mcp_tools.py b/orchestrator/tests/test_restart_mcp_tools.py index 40437fb1d..924642213 100644 --- a/orchestrator/tests/test_restart_mcp_tools.py +++ b/orchestrator/tests/test_restart_mcp_tools.py @@ -145,6 +145,37 @@ def test_returns_structured_success(self, handler): assert result["restarted"] is True assert result["agent_role"] == "coder" + def test_surfaces_teardown_and_delegation_instead_of_empty_container_id(self, handler): + """The result says what happened, not ``container_id: ""`` (#3597). + + ``restart_agent`` deliberately spawns no resident pod (#3164), so the + old ``container_id`` was always empty — which reads as a failure and, + during the #3597 incident, looked identical whether the role was + respawned or had silently vanished. Pass the route's own reporting + through instead. + """ + with patch.object(handler, "_make_request") as mock_req: + mock_req.return_value = { + "success": True, + "data": { + "respawn": "delegated to orchestrator event loop", + "live_event_loop": True, + "jobs_torn_down": 1, + "teardown_confirmed": True, + "restart_count": 2, + }, + } + result = handler.handle_tool_call( + "restart_agent", + {"task_id": "issue-42", "agent_role": "coder"}, + ) + + assert result["respawn"] == "delegated to orchestrator event loop" + assert result["live_event_loop"] is True + assert result["jobs_torn_down"] == 1 + assert result["teardown_confirmed"] is True + assert result["restart_count"] == 2 + def test_returns_error_on_failure(self, handler): """Failed restart returns error dict.""" with patch.object(handler, "_make_request") as mock_req: