[codex] fix passthrough stream idle disconnects

src/server.ts

@@ -199,9 +199,14 @@ async function handleResponses(
         : `Provider unreachable: ${err instanceof Error ? err.message : String(err)}`;
       return formatErrorResponse(502, "upstream_error", msg);
     }
-    return new Response(relayWithAbort(upstreamResponse.body, upstream), {
+    const headers = sanitizePassthroughHeaders(upstreamResponse.headers);
+    const isEventStream = headers.get("content-type")?.toLowerCase().includes("text/event-stream") ?? false;
+    const body = isEventStream
+      ? relaySseWithHeartbeat(upstreamResponse.body, upstream)
+      : relayWithAbort(upstreamResponse.body, upstream);
+    return new Response(body, {
       status: upstreamResponse.status,
-      headers: sanitizePassthroughHeaders(upstreamResponse.headers),
+      headers,
     });
   }
 
@@ -343,6 +348,56 @@ export function relayWithAbort(
   });
 }
 
+export function relaySseWithHeartbeat(
+  body: ReadableStream<Uint8Array> | null,
+  upstream: AbortController,
+  heartbeatMs = 15_000,
+): ReadableStream<Uint8Array> | null {
+  if (!body) return null;
+  const reader = body.getReader();
+  const heartbeat = new TextEncoder().encode(": opencodex keepalive\n\n");
+  let timer: ReturnType<typeof setInterval> | undefined;
+  let closed = false;
+
+  const cleanup = () => {
+    closed = true;
+    if (timer) clearInterval(timer);
+    timer = undefined;
+  };
+
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      timer = setInterval(() => {
+        if (closed) return;
+        try {
+          controller.enqueue(heartbeat);
+        } catch {
+          cleanup();
+        }
+      }, heartbeatMs);
+    },
+    async pull(controller) {
+      try {
+        const { done, value } = await reader.read();
+        if (done) {
+          cleanup();
+          controller.close();
+          return;
+        }
+        controller.enqueue(value);
+      } catch (err) {
+        cleanup();
+        try { controller.error(err); } catch { /* already torn down */ }
+      }
+    },
+    cancel(reason) {
+      cleanup();
+      upstream.abort(reason);
+      reader.cancel(reason).catch(() => {});
+    },
+  });
+}
+
 /**
  * Bun's fetch auto-decompresses the response body but leaves the upstream `content-encoding`
  * (and a now-stale `content-length`) on `response.headers`. Relaying those with the already-decoded

src/service.ts

@@ -85,6 +85,12 @@ function systemdEnvironmentAssignment(name: string, value: string | undefined):
   return `Environment=${systemdQuote(`${name}=${value}`)}`;
 }
 
+function systemdOutputTarget(value: string): string {
+  // StandardOutput/StandardError use output specifiers such as append:/path.
+  // Quoting the full specifier makes systemd reject it as an invalid output target.
+  return value.replace(/%/g, "%%").replace(/\n/g, "\\n");
+}
+
 function sh(cmd: string): string {
   return execSync(cmd, { encoding: "utf8", stdio: ["pipe", "pipe", "pipe"] }).trim();
 }
@@ -183,8 +189,8 @@ ExecStart=${systemdQuote(bun)} ${systemdQuote(cli)} start
 Restart=on-failure
 RestartSec=5
 ${envLines}
-StandardOutput=${systemdQuote(`append:${log}`)}
-StandardError=${systemdQuote(`append:${log}`)}
+StandardOutput=${systemdOutputTarget(`append:${log}`)}
+StandardError=${systemdOutputTarget(`append:${log}`)}
 
 [Install]
 WantedBy=default.target

tests/passthrough-abort.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, test } from "bun:test";
-import { linkAbortSignal, relayWithAbort } from "../src/server";
+import { linkAbortSignal, relaySseWithHeartbeat, relayWithAbort } from "../src/server";
 
 function streamFromChunks(chunks: Uint8Array[]): ReadableStream<Uint8Array> {
   let i = 0;
@@ -46,6 +46,24 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => {
     expect(ac.signal.aborted).toBe(false);
   });
 
+  test("SSE passthrough emits heartbeat comments while upstream is silent", async () => {
+    const ac = new AbortController();
+    const body = new ReadableStream<Uint8Array>({ pull() { return new Promise<void>(() => {}); } });
+    const relayed = relaySseWithHeartbeat(body, ac, 5)!;
+    const reader = relayed.getReader();
+    const first = await Promise.race([
+      reader.read(),
+      new Promise<never>((_, reject) => setTimeout(() => reject(new Error("heartbeat timeout")), 200)),
+    ]);
+
+    expect(first.done).toBe(false);
+    expect(new TextDecoder().decode(first.value)).toBe(": opencodex keepalive\n\n");
+
+    await reader.cancel("client gone");
+    expect(ac.signal.aborted).toBe(true);
+    expect(ac.signal.reason).toBe("client gone");
+  });
+
   test("turn-level abort signal aborts the upstream fetch before headers arrive", () => {
     const upstream = new AbortController();
     const turn = new AbortController();

tests/service.test.ts

@@ -0,0 +1,13 @@
+import { describe, expect, test } from "bun:test";
+import { buildUnit } from "../src/service";
+
+describe("systemd service unit", () => {
+  test("uses unquoted append targets for service logs", () => {
+    const unit = buildUnit();
+
+    expect(unit).toContain("StandardOutput=append:");
+    expect(unit).toContain("StandardError=append:");
+    expect(unit).not.toContain('StandardOutput="append:');
+    expect(unit).not.toContain('StandardError="append:');
+  });
+});