@@ -0,0 +1,175 @@
+import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
+import type { AddressInfo } from "node:net";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { type KubeConfig, type KubernetesObject, ListWatch, Watch } from "@kubernetes/client-node";
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import { ReconnectingInformer } from "./reconnectingInformer.js";
+
+describe("ReconnectingInformer watch stalls", () => {
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  type WatchCall = { path: string; done: (err: unknown) => void; aborted: boolean };
+
+  function makeInformer(
+    watch: Pick<Watch, "watch">,
+    opts: { watchTimeoutSeconds?: number; onStall?: (quietMs: number) => void } = {}
+  ) {
+    return new ReconnectingInformer<KubernetesObject>({
+      name: "test-informer",
+      logger: new SimpleStructuredLogger("test-informer"),
+      reconnectIntervalMs: 1,
+      path: "/api/v1/namespaces/v4-runs/pods",
+      list: async () => ({ items: [], metadata: { resourceVersion: "1" } }),
+      makeInformer: (path, list) => new ListWatch(path, watch as Watch, list, false),
+      ...opts,
+    });
+  }
+
+  /** A fake Watch that stays silent until told otherwise. */
+  function setup(opts: Parameters<typeof makeInformer>[1] = { watchTimeoutSeconds: 300 }) {
+    const calls: WatchCall[] = [];
+    const watch = {
+      watch: vi.fn(
+        async (path: string, _query: unknown, _callback: unknown, done: WatchCall["done"]) => {
+          const call: WatchCall = { path, done, aborted: false };
+          calls.push(call);
+          return { abort: () => (call.aborted = true) } as unknown as AbortController;
+        }
+      ),
+    };
+    return { calls, informer: makeInformer(watch, opts) };
+  }
+
+  it("asks the server to close each watch at the timeout", async () => {
+    const { calls, informer } = setup();
+    await informer.start();
+
+    expect(calls[0]!.path).toBe("/api/v1/namespaces/v4-runs/pods?timeoutSeconds=300");
+    await informer.stop();
+  });
+
+  // A connection that died without closing raises nothing on its own.
+  it("reconnects a watch that neither delivers nor closes past the timeout", async () => {
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval", "Date"] });
+    const onStall = vi.fn();
+    const { calls, informer } = setup({ watchTimeoutSeconds: 300, onStall });
+    await informer.start();
+
+    await vi.advanceTimersByTimeAsync(300_000);
+    expect(calls).toHaveLength(1);
+
+    // The check runs every 30 s, so the first past 330 s quiet is at 360 s.
+    await vi.advanceTimersByTimeAsync(60_000);
+    await vi.waitFor(() => expect(calls).toHaveLength(2));
+    expect(calls[0]!.aborted).toBe(true);
+    expect(onStall).toHaveBeenCalledOnce();
+    await informer.stop();
+  });
+
+  it("leaves alone a watch the server closes and the client renews", async () => {
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval", "Date"] });
+    const onStall = vi.fn();
+    const { calls, informer } = setup({ watchTimeoutSeconds: 300, onStall });
+    await informer.start();
+
+    for (let i = 1; i <= 3; i++) {
+      await vi.advanceTimersByTimeAsync(300_000);
+      calls.at(-1)!.done(null);
+      await vi.waitFor(() => expect(calls).toHaveLength(i + 1));
+    }
+    await vi.advanceTimersByTimeAsync(60_000);
+
+    expect(calls).toHaveLength(4);
+    expect(onStall).not.toHaveBeenCalled();
+    await informer.stop();
+  });
+
+  it("sets no timeout and checks nothing without a watch timeout", async () => {
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval", "Date"] });
+    const { calls, informer } = setup({});
+    await informer.start();
+
+    await vi.advanceTimersByTimeAsync(3_600_000);
+
+    expect(calls).toHaveLength(1);
+    expect(calls[0]!.path).toBe("/api/v1/namespaces/v4-runs/pods");
+    await informer.stop();
+  });
+
+  /** Polls on real timers, which the fake ones in these tests leave alone. */
+  async function realTimeUntil(condition: () => boolean) {
+    for (let i = 0; i < 400 && !condition(); i++) {
+      await new Promise((resolve) => setTimeout(resolve, 5));
+    }
+    expect(condition()).toBe(true);
+  }
+
+  /** A real client Watch against a local API server that answers each watch as told. */
+  async function realWatch(answer: (n: number, res: ServerResponse) => void) {
+    const watches: IncomingMessage[] = [];
+    const server = createServer((req, res) => {
+      watches.push(req);
+      answer(watches.length, res);
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+    // A KubeConfig always fetches through an https agent; this is plain http.
+    const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+    const kc = {
+      getCurrentCluster: () => ({ server: url }),
+      applyToFetchOptions: async () => ({}),
+    } as unknown as KubeConfig;
+    const close = () =>
+      new Promise<void>((resolve) => {
+        server.closeAllConnections();
+        server.close(() => resolve());
+      });
+    return { watch: new Watch(kc), watches, close };
+  }
+
+  // The client only returns the request's controller once the headers arrive.
+  it("aborts a watch whose headers never arrive", async () => {
+    const before = Watch.HEADERS_TIMEOUT_MS;
+    Watch.HEADERS_TIMEOUT_MS = 50;
+    const { watch, close } = await realWatch(() => {});
+    try {
+      const done = vi.fn();
+      await watch.watch("/api/v1/pods", {}, () => {}, done);
+      expect(done).toHaveBeenCalledWith(expect.objectContaining({ name: "AbortError" }));
+    } finally {
+      Watch.HEADERS_TIMEOUT_MS = before;
+      await close();
+    }
+  });
+
+  // The stall's reconnect meets a blackholed connection too: it must give up on
+  // it rather than hold the reconnect guard, and watch again.
+  it("recovers when the reconnect after a stall never gets its headers", async () => {
+    const before = Watch.HEADERS_TIMEOUT_MS;
+    Watch.HEADERS_TIMEOUT_MS = 50;
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval", "Date"] });
+    // The first watch connects and goes silent, the second never answers, the third connects.
+    const { watch, watches, close } = await realWatch((n, res) => {
+      if (n !== 2) {
+        res.writeHead(200, { "content-type": "application/json" });
+        res.flushHeaders();
+      }
+    });
+    const informer = makeInformer(watch, { watchTimeoutSeconds: 300 });
+    const reconnecting = () => (informer as unknown as { reconnecting: boolean }).reconnecting;
+    try {
+      await informer.start();
+      await vi.waitFor(() => expect(watches).toHaveLength(1));
+
+      await vi.advanceTimersByTimeAsync(360_000);
+      // The second watch's header timeout runs on real time; only once it fires
+      // does the reconnect watch a third time and settle.
+      await realTimeUntil(() => watches.length === 3 && !reconnecting());
+    } finally {
+      await informer.stop();
+      Watch.HEADERS_TIMEOUT_MS = before;
+      await close();
+    }
+  });
+});
@@ -4,21 +4,40 @@ import type { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structur
 
 const MAX_RECONNECT_BACKOFF_MS = 30_000;
 
+// Past the server closing the watch at its timeout, how long to wait for the
+// client to watch again before taking the watch for stalled.
+const WATCH_STALL_GRACE_MS = 30_000;
+
 type ReconnectingInformerOptions<T extends KubernetesObject> = {
   /** Used in log fields, to tell informers apart. */
   name: string;
   logger: SimpleStructuredLogger;
   reconnectIntervalMs: number;
+  /** The watch path, which gets the timeout as a query parameter. */
+  path: string;
   list: ListPromise<T>;
-  /** Builds the informer around the list it is given, which wraps `list`. */
-  makeInformer: (list: ListPromise<T>) => ListWatch<T>;
+  /** Builds the informer around the path and list it is given, which wraps `list`. */
+  makeInformer: (path: string, list: ListPromise<T>) => ListWatch<T>;
   /** Called for every error the informer raises, before any reconnect. */
   onError?: (err: unknown) => void;
+  /**
+   * How long the server keeps each watch open before closing it. A watch that has
+   * neither connected nor delivered an event for this plus 30 s is taken as
+   * stalled. Without it, a watch has no timeout and is never checked.
+   */
+  watchTimeoutSeconds?: number;
+  /** Called when the stall check takes the watch for stalled, before it reconnects. */
+  onStall?: (quietMs: number) => void;
 };
 
 /**
  * An informer that reconnects with a capped backoff until a start succeeds, for
  * as long as it runs. The client's own error handling gives up after one failure.
+ *
+ * A connection that died without closing (a dropped NAT entry, a blackholed load
+ * balancer) raises nothing. With a watch timeout, the server closes each watch
+ * and the client watches again, so a healthy watch connects at least that often,
+ * and a stall check reconnects one that goes quiet for longer.
  */
 export class ReconnectingInformer<T extends KubernetesObject> {
   readonly informer: ListWatch<T>;
@@ -27,6 +46,10 @@ export class ReconnectingInformer<T extends KubernetesObject> {
   private readonly reconnectIntervalMs: number;
   private readonly listFn: ListPromise<T>;
   private readonly onErrorHook?: (err: unknown) => void;
+  private readonly onStallHook?: (quietMs: number) => void;
+  private readonly watchTimeoutMs?: number;
+  private lastWatchActivityAt = 0;
+  private stallCheck?: NodeJS.Timeout;
   private running = false;
   private reconnecting = false;
   private erroredDuringReconnect = false;
@@ -39,8 +62,19 @@ export class ReconnectingInformer<T extends KubernetesObject> {
     this.reconnectIntervalMs = opts.reconnectIntervalMs;
     this.listFn = opts.list;
     this.onErrorHook = opts.onError;
-    this.informer = opts.makeInformer(() => this.list());
+    this.onStallHook = opts.onStall;
+    this.watchTimeoutMs =
+      opts.watchTimeoutSeconds === undefined ? undefined : opts.watchTimeoutSeconds * 1000;
+    const path =
+      opts.watchTimeoutSeconds === undefined
+        ? opts.path
+        : `${opts.path}?timeoutSeconds=${opts.watchTimeoutSeconds}`;
+    this.informer = opts.makeInformer(path, () => this.list());
     this.informer.on("error", (err?: unknown) => void this.onError(err));
+    this.informer.on("connect", () => this.markWatchActivity());
+    this.informer.on("add", () => this.markWatchActivity());
+    this.informer.on("update", () => this.markWatchActivity());
+    this.informer.on("delete", () => this.markWatchActivity());
   }
 
   get isRunning(): boolean {
@@ -53,17 +87,57 @@ export class ReconnectingInformer<T extends KubernetesObject> {
       return;
     }
     this.running = true;
-    await this.startInformer();
+    if (this.watchTimeoutMs !== undefined) {
+      this.markWatchActivity();
+      // Installed first, so a first watch that never comes up is covered too.
+      this.stallCheck = setInterval(
+        () => void this.checkWatchStalled(),
+        Math.min(WATCH_STALL_GRACE_MS, this.watchTimeoutMs)
+      );
+      this.stallCheck.unref();
+    }
+    try {
+      await this.startInformer();
+    } catch (err: unknown) {
+      clearInterval(this.stallCheck);
+      throw err;
+    }
   }
 
   async stop() {
     if (!this.running) {
       return;
     }
     this.running = false;
+    clearInterval(this.stallCheck);
     await this.informer.stop();
   }
 
+  private markWatchActivity() {
+    this.lastWatchActivityAt = Date.now();
+  }
+
+  /**
+   * Skips while a reconnect runs: a reconnect into a blackholed connection ends
+   * once the client's header timeout aborts it, and the next check retries.
+   */
+  private async checkWatchStalled() {
+    const quietMs = Date.now() - this.lastWatchActivityAt;
+    if (
+      !this.running ||
+      this.reconnecting ||
+      this.watchTimeoutMs === undefined ||
+      quietMs <= this.watchTimeoutMs + WATCH_STALL_GRACE_MS
+    ) {
+      return;
+    }
+    this.onStallHook?.(quietMs);
+    this.markWatchActivity();
+    await this.informer.stop();
+    // The stopped request's own error arrives while this reconnect waits, and is skipped.
+    await this.reconnect(new Error("watch stalled"));
+  }
+
   /**
    * The client relists on its own after a 410 and leaves that list's rejection
    * unhandled, so a failure there becomes a reconnect and the abandoned relist
@@ -100,15 +174,22 @@ export class ReconnectingInformer<T extends KubernetesObject> {
     }
   }
 
+  private async onError(err: unknown) {
+    if (!this.running) {
+      return;
+    }
+    this.onErrorHook?.(err);
+    await this.reconnect(err);
+  }
+
   /**
    * Retries until a start ends with no error raised during it. A watch that
    * fails to connect raises its error inside the start, which still resolves.
    */
-  private async onError(err: unknown) {
+  private async reconnect(err: unknown) {
     if (!this.running) {
       return;
     }
-    this.onErrorHook?.(err);
     if (this.reconnecting) {
       this.erroredDuringReconnect = true;
       return;
 
@@ -352,6 +352,7 @@ export const Env = z
     // Failed pod handler
     FAILED_POD_HANDLER_ENABLED: BoolEnv.default(true),
     FAILED_POD_HANDLER_RECONNECT_INTERVAL_MS: z.coerce.number().int().default(1000),
+    FAILED_POD_HANDLER_WATCH_TIMEOUT_SECONDS: z.coerce.number().int().positive().default(300),
 
     // Debug
     DEBUG: BoolEnv.default(false),
 
@@ -292,6 +292,7 @@ class ManagedSupervisor {
           register,
           namespace: env.KUBERNETES_NAMESPACE,
           reconnectIntervalMs: env.FAILED_POD_HANDLER_RECONNECT_INTERVAL_MS,
+          watchTimeoutSeconds: env.FAILED_POD_HANDLER_WATCH_TIMEOUT_SECONDS,
         });
       } else {
         this.logger.warn("Failed pod handler disabled");