executor HTTP: keep the answer alive, so a long apply is not given up on after 5 minutes

Nextcloud 34 took 6 minutes to apply (a large pull). Node's fetch drops a response that sends no headers for 5 minutes, so
the driver gave up ("executor unreachable ... UND_ERR_HEADERS_TIMEOUT") on an apply that then finished; the owner's retry and
the repair met "another operation is running". The executor now answers 200 at once and sends a space every 30 s, on the
driver's routes and on the model's /agent/* ones (pull and up there are just as long). The operation is never stopped when the
caller hangs up: an apply half done is worse than one that finishes. A failure after the headers went out is read from the body
by the driver's client. The keep-alive is shared with the model sidecar (http/keep-alive.ts). No new dependency.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

src/http/executor-server.test.ts

@@ -1,3 +1,4 @@
+import type { AddressInfo } from 'node:net';
 import { afterEach, describe, expect, it } from 'vitest';
 import type { AgentSessions } from '../app/executor/agent.ts';
 import type { ComposeExecutor } from '../app/executor/executor.ts';
@@ -178,3 +179,54 @@ describe('repair sessions over HTTP', () => {
     expect(calls).toEqual([{ op: 'open', request: { project: 'nextcloud', service: 'app' } }]);
   });
 });
+
+// Nextcloud 34 took 6 minutes to apply (a 1 GB pull); Node's fetch drops a response that sends no headers for 5 minutes, so the
+// driver gave up on an apply that then finished (2026-09-20). The answer now starts at once and is kept alive.
+describe('a long operation over HTTP', () => {
+  async function slowApply(work: () => Promise<ExecutorResponse<unknown>>) {
+    const executor: ComposeExecutor = { ...recordingExecutor().executor, apply: work as ComposeExecutor['apply'] };
+    const app = await buildExecutorServer({ executor, heartbeatMs: 10 });
+    await app.listen({ host: '127.0.0.1', port: 0 });
+    closers.push(() => app.close());
+    return `http://127.0.0.1:${String((app.server.address() as AddressInfo).port)}`;
+  }
+
+  it('sends the headers at once and keeps the connection alive while the executor works, and the answer still parses', async () => {
+    let release = (): void => undefined;
+    let finished = false;
+    const url = await slowApply(() => new Promise((resolve) => (release = () => ((finished = true), resolve({ outcome: 'refused', reason: 'not_allowed', detail: 'slow' })))));
+    const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(APPLY) });
+    expect(finished).toBe(false); // the apply is still going, yet the response is here
+    expect(response.status).toBe(200);
+    await new Promise((resolve) => setTimeout(resolve, 60)); // several heartbeats
+    release();
+    expect(await response.json()).toEqual({ outcome: 'refused', reason: 'not_allowed', detail: 'slow' });
+  });
+
+  it('does not stop the operation when the driver hangs up: an apply half done is worse than an apply that finishes', async () => {
+    let release = (): void => undefined;
+    let done = false;
+    const url = await slowApply(() => new Promise((resolve) => (release = () => ((done = true), resolve({ outcome: 'refused', reason: 'busy', detail: 'x' })))));
+    const hangUp = new AbortController();
+    const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(APPLY), signal: hangUp.signal });
+    hangUp.abort();
+    await response.text().catch(() => undefined);
+    await new Promise((resolve) => setTimeout(resolve, 40));
+    expect(done).toBe(false);
+    release();
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    expect(done).toBe(true);
+  });
+
+  it('an unexpected failure after the headers went out is still an answer the client can read, as a failure of the executor', async () => {
+    const url = await slowApply(() => Promise.reject(new Error('disk failure at /secret/path')));
+    const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(APPLY) });
+    expect(await response.json()).toEqual({ error: 'internal', message: 'unexpected executor error' });
+  });
+
+  it('a request that is not valid is still refused with 400 before anything starts', async () => {
+    const url = await slowApply(() => Promise.resolve({ outcome: 'refused', reason: 'busy', detail: 'x' }));
+    const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ project: 'x' }) });
+    expect(response.status).toBe(400);
+  });
+});

src/http/executor-server.ts

@@ -17,6 +17,7 @@ import {
   SessionOpenRequest,
 } from '../domain/executor-api.ts';
 import type { ExecutorResponse } from '../domain/executor-api.ts';
+import { answerWhenDone } from './keep-alive.ts';
 
 // The executor's HTTP surface on the internal network of the driver, the executor and the model. The driver's routes (one
 // operation on one service each, and opening/closing a repair session) answer only the driver's address; the model's `hl`
@@ -29,6 +30,8 @@ export interface ExecutorServerDeps {
   /** Whether a remote address is the driver's; without it every address may call the driver's routes (tests, local runs). */
   readonly isDriver?: (address: string) => Promise<boolean>;
   readonly logger?: boolean;
+  /** How often a running operation sends a space to keep the connection alive (http/keep-alive.ts). */
+  readonly heartbeatMs?: number;
 }
 
 /** Requests are a few hundred bytes; anything much larger is not from the driver. The model writes whole compose files. */
@@ -54,11 +57,19 @@ export async function buildExecutorServer(deps: ExecutorServerDeps): Promise<Fas
     return reply.code(500).send({ error: 'internal', message: 'unexpected executor error' });
   });
 
+  const heartbeat = { ...(deps.heartbeatMs === undefined ? {} : { heartbeatMs: deps.heartbeatMs }) };
+  /** The answer to a request whose work threw: the status line is already sent, so it is the body the caller reads as a failure of the executor. */
+  const internalFailure = (log: { error: (fields: object, message: string) => void }, cause: unknown) => {
+    log.error({ err: cause }, 'request failed');
+    return { error: 'internal', message: 'unexpected executor error' };
+  };
+
   const route = <T>(url: string, schema: z.ZodType<T>, run: (request: T) => Promise<ExecutorResponse<unknown>>) => {
     app.post(url, async (request, reply) => {
       const parsed = schema.safeParse(request.body);
       if (!parsed.success) return badRequest(reply, issues(parsed.error));
-      return reply.send(await run(parsed.data));
+      // Never stopped when the driver hangs up: an apply half done is worse than an apply that finishes (its lock says so).
+      await answerWhenDone(reply, heartbeat, () => run(parsed.data), (cause) => internalFailure(request.log, cause));
     });
   };
 
@@ -88,7 +99,7 @@ export async function buildExecutorServer(deps: ExecutorServerDeps): Promise<Fas
         if (token === null) return reply.code(401).send({ error: 'unauthorized', message: 'a repair session token is required' });
         const parsed = schema.safeParse(request.body ?? {});
         if (!parsed.success) return badRequest(reply, issues(parsed.error));
-        return reply.send(await run(token, parsed.data));
+        await answerWhenDone(reply, heartbeat, () => run(token, parsed.data), (cause) => internalFailure(request.log, cause));
       });
     };
     const Nothing = z.strictObject({});

src/http/keep-alive.ts

@@ -0,0 +1,33 @@
+import type { FastifyReply } from 'fastify';
+
+// Node's fetch drops a response that sends no headers, or no new body bytes, for 5 minutes. Some answers here take longer
+// (a model session, an apply that pulls a large image), and the caller then gives up on work that goes on and finishes.
+// So the response starts at once (200, JSON) and a space goes out every `heartbeatMs`: JSON allows leading whitespace, so
+// the caller's `response.json()` reads the answer as it always did. The status is 200 whatever happens, so a failure travels
+// in the body: the caller decides what an answer means, as it already does.
+
+export const HEARTBEAT_MS = 30_000;
+
+/**
+ * Answers `reply` with what `work` returns, as JSON, kept alive meanwhile. `hangUp` fires when the caller closes the
+ * connection before the answer: `work` decides whether that stops it (a model run: yes; an apply: never, half done is worse).
+ * A `work` that throws is answered with `failure(cause)`.
+ */
+export async function answerWhenDone(reply: FastifyReply, options: { readonly heartbeatMs?: number }, work: (hangUp: AbortSignal) => Promise<unknown>, failure: (cause: unknown) => unknown): Promise<void> {
+  const { raw } = reply;
+  reply.hijack();
+  raw.writeHead(200, { 'content-type': 'application/json; charset=utf-8' });
+  raw.flushHeaders();
+  const hangUp = new AbortController();
+  raw.once('close', () => {
+    if (!raw.writableFinished) hangUp.abort(new Error('the caller closed the connection'));
+  });
+  const heartbeat = setInterval(() => raw.write(' '), options.heartbeatMs ?? HEARTBEAT_MS);
+  try {
+    raw.end(JSON.stringify(await work(hangUp.signal)));
+  } catch (cause) {
+    raw.end(JSON.stringify(failure(cause)));
+  } finally {
+    clearInterval(heartbeat);
+  }
+}

src/http/model-server.ts

@@ -2,6 +2,7 @@ import Fastify from 'fastify';
 import type { FastifyInstance } from 'fastify';
 import type { ClaudeRunRequest, ClaudeRunResult } from '../app/model/claude.ts';
 import { ModelRunRequest } from '../domain/model-api.ts';
+import { answerWhenDone } from './keep-alive.ts';
 import type { ModelRunResponse } from '../domain/model-api.ts';
 
 // The model sidecar's HTTP surface, for the driver on the internal network: run one prompt, one at a time.
@@ -37,28 +38,19 @@ export async function buildModelServer(deps: ModelServerDeps): Promise<FastifyIn
     if (!parsed.success) return reply.code(400).send({ error: 'bad_request', message: parsed.error.issues.map((issue) => `${issue.path.join('.')}: ${issue.message}`).join('; ') });
     if (busy) return reply.send({ outcome: 'failed', kind: 'busy', detail: 'another model run is in progress' } satisfies ModelRunResponse);
     busy = true;
-    const { raw } = reply;
-    reply.hijack();
-    raw.writeHead(200, { 'content-type': 'application/json; charset=utf-8' });
-    raw.flushHeaders();
-    const hangUp = new AbortController();
-    raw.once('close', () => {
-      if (!raw.writableFinished) hangUp.abort(new Error('the driver closed the connection'));
-    });
-    const heartbeat = setInterval(() => raw.write(' '), deps.heartbeatMs ?? 30_000);
-    try {
-      const { executor, profile, prompt, output_schema: outputSchema, session_token: sessionToken } = parsed.data;
-      const result = await deps.run({ model: executor, profile, prompt, outputSchema, timeoutMs: deps.timeoutMs, ...(sessionToken === undefined ? {} : { sessionToken }) }, hangUp.signal);
-      const answer: ModelRunResponse = result.ok ? { outcome: 'done', output: result.output, cost_usd: result.costUsd, turns: result.turns } : { outcome: 'failed', kind: result.kind, detail: result.detail, ...(result.costUsd === undefined ? {} : { cost_usd: result.costUsd }) };
-      raw.end(JSON.stringify(answer));
-    } catch (cause) {
+    await answerWhenDone(
+      reply,
+      { ...(deps.heartbeatMs === undefined ? {} : { heartbeatMs: deps.heartbeatMs }) },
+      async (hangUp) => {
+        const { executor, profile, prompt, output_schema: outputSchema, session_token: sessionToken } = parsed.data;
+        const result = await deps.run({ model: executor, profile, prompt, outputSchema, timeoutMs: deps.timeoutMs, ...(sessionToken === undefined ? {} : { sessionToken }) }, hangUp);
+        return result.ok ? ({ outcome: 'done', output: result.output, cost_usd: result.costUsd, turns: result.turns } satisfies ModelRunResponse) : ({ outcome: 'failed', kind: result.kind, detail: result.detail, ...(result.costUsd === undefined ? {} : { cost_usd: result.costUsd }) } satisfies ModelRunResponse);
+      },
       // The status line is already sent, so a failure travels in the body like every other one.
-      const detail = cause instanceof Error ? cause.message : String(cause);
-      raw.end(JSON.stringify({ outcome: 'failed', kind: 'unavailable', detail } satisfies ModelRunResponse));
-    } finally {
-      clearInterval(heartbeat);
+      (cause) => ({ outcome: 'failed', kind: 'unavailable', detail: cause instanceof Error ? cause.message : String(cause) }) satisfies ModelRunResponse,
+    ).finally(() => {
       busy = false;
-    }
+    });
   });
   app.get('/health', () => ({ ok: true }));
 

src/infra/http-executor-client.test.ts

@@ -85,6 +85,11 @@ describe('HTTP executor client failures', () => {
     expect(await client.backup({ project: 'jackett' })).toEqual({ outcome: 'unavailable', detail: 'executor failed (500): unexpected executor error' });
   });
 
+  it('a failure that comes after the executor already answered 200 (it keeps the connection alive) is read from the body', async () => {
+    const client = createHttpExecutorClient({ baseUrl: 'http://executor:8081', fetch: answering({ error: 'internal', message: 'unexpected executor error' }, 200) });
+    expect(await client.backup({ project: 'jackett' })).toEqual({ outcome: 'unavailable', detail: 'executor failed: unexpected executor error' });
+  });
+
   it('an abort is not swallowed: the driver’s time limit stays in charge', async () => {
     const controller = new AbortController();
     const client = createHttpExecutorClient({

src/infra/http-executor-client.ts

@@ -53,6 +53,8 @@ export function createHttpExecutorClient(options: HttpExecutorClientOptions): Ex
     }
     if (response.status >= 400 && response.status < 500) return { outcome: 'unavailable', detail: `executor refused the request (${String(response.status)}): ${messageOf(payload)}` };
     if (!response.ok) return { outcome: 'unavailable', detail: `executor failed (${String(response.status)}): ${messageOf(payload)}` };
+    // The executor answers 200 at once and keeps the connection alive, so a failure that comes later is in the body.
+    if (typeof payload === 'object' && payload !== null && 'error' in payload) return { outcome: 'unavailable', detail: `executor failed: ${messageOf(payload)}` };
     const parsed = schema.safeParse(payload);
     if (!parsed.success) return { outcome: 'unavailable', detail: `executor answered ${path} in an unexpected shape: ${parsed.error.issues[0]?.message ?? 'invalid'}` };
     return parsed.data;