From a3719aeb63f980635827ed7502c68db321e45893 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sat, 15 Aug 2026 00:10:11 +0000 Subject: [PATCH 01/23] Stop cutting off SSE streams after 30 seconds of silence Bun closes a connection that has been quiet in both directions for idleTimeout seconds, and that timer runs against an in-flight SSE response just as it does against an idle keep-alive socket. Elysia's Bun adapter defaults it to 30, and we never overrode it, so any stream that went quiet for half a minute was dropped by our own server. Our long paths are quiet for far longer than that by design: the planner buffers Opus output and flushes it once the call returns, subagent calls are deliberately not streamed, and workflow_chat stays silent for the whole YAML phase. The client gets no error frame, just a dropped connection, which is why Lightning can only report it as "Stream ended without complete response" - 83 times in production over the last 90 days. 255 is Bun's maximum. This was fixed once in dc7fe8c and removed again in 8c35c4f, a commit titled "update lockfile", so it comes back with a test this time. The test asserts the configuration rather than the behaviour, because every test here drives the app through app.handle(), which never opens a socket - which is also why nothing caught the removal. --- .changeset/restore-sse-idle-timeout.md | 6 ++++++ platform/src/server.ts | 9 ++++++++- platform/test/server.test.ts | 7 +++++++ 3 files changed, 21 insertions(+), 1 deletion(-) create mode 100644 .changeset/restore-sse-idle-timeout.md diff --git a/.changeset/restore-sse-idle-timeout.md b/.changeset/restore-sse-idle-timeout.md new file mode 100644 index 00000000..9a2a65c1 --- /dev/null +++ b/.changeset/restore-sse-idle-timeout.md @@ -0,0 +1,6 @@ +--- +"apollo": patch +--- + +Raise the server's socket idle timeout back to 255s so long-running SSE +streams are no longer cut off while a model is thinking diff --git a/platform/src/server.ts b/platform/src/server.ts index cbd8f3a1..297408d1 100644 --- a/platform/src/server.ts +++ b/platform/src/server.ts @@ -19,7 +19,14 @@ export default async ( // pass a pre-configured instance (fake lookup) instead of the live DB-backed one. auth: InstanceAuth = new InstanceAuth() ) => { - const app = new Elysia(); + // Bun's idle timer applies to in-flight SSE responses, not just idle + // keep-alive sockets, and Elysia defaults it to 30s - shorter than our own + // services routinely go without emitting. 255 is Bun's maximum. + const app = new Elysia({ + serve: { + idleTimeout: 255, + }, + }); app.use(html()); diff --git a/platform/test/server.test.ts b/platform/test/server.test.ts index 72ac4e96..6669e014 100644 --- a/platform/test/server.test.ts +++ b/platform/test/server.test.ts @@ -85,6 +85,13 @@ describe("Main server", () => { expect(await response.text()).toBe(""); }); + // Asserts the configuration rather than the behaviour: these tests drive the + // app through app.handle(), which never opens a socket, so none of them can + // observe a socket timer. + it("keeps the socket idle timeout long enough for slow SSE streams", () => { + expect(app.config.serve?.idleTimeout).toBe(255); + }); + // send messages through a web socket }); From c4b957e6cc0315e105dc4abfcd2b94bfbfdec7bf Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sat, 15 Aug 2026 00:16:53 +0000 Subject: [PATCH 02/23] Send a keepalive on streaming responses Raising our own idle timeout moves the cliff without removing it: a stream still dies if it stays quiet long enough, and every hop between us and the client is still guessing how long a silence is normal. Our silences are long by design - the planner buffers Opus output until the call returns, subagent calls are not streamed, and workflow_chat says nothing for the whole YAML phase - so a working stream is regularly indistinguishable from a dead one. A comment frame every 15 seconds makes that distinction real. Silence now means something is actually wrong, which is what lets the timeouts either side come down to values that detect faults rather than merely tolerate slowness. It runs from the stream writer rather than from Python so it also covers the window before Python is alive, which is where a cold start plus a slow first token does the damage, and so it survives a child that dies without saying anything. The frame is ": ping" with the space. Lightning decodes SSE with Tesla, whose decoder matches ": " and has no catch-all, so dropping the space would turn a silent stall into a crash on the client. There are tests on the exact bytes. --- .changeset/sse-heartbeat.md | 6 ++++ platform/src/middleware/services.ts | 39 ++++++++++++++++++++-- platform/test/middleware/heartbeat.test.ts | 31 +++++++++++++++++ 3 files changed, 74 insertions(+), 2 deletions(-) create mode 100644 .changeset/sse-heartbeat.md create mode 100644 platform/test/middleware/heartbeat.test.ts diff --git a/.changeset/sse-heartbeat.md b/.changeset/sse-heartbeat.md new file mode 100644 index 00000000..6d2e717d --- /dev/null +++ b/.changeset/sse-heartbeat.md @@ -0,0 +1,6 @@ +--- +"apollo": patch +--- + +Send a periodic keepalive on streaming responses so a stream that is working +but quiet is no longer mistaken for a dead one diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index 1d5d6073..e244ecee 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -11,6 +11,14 @@ import type { InstanceAuth } from "../auth/instance-auth"; const textEncoder = new TextEncoder(); +// The space after the colon is required: Lightning's SSE decoder matches +// `": " <> comment` and has no catch-all clause, so ":ping" raises there. +export const HEARTBEAT_FRAME = ": ping\n\n"; + +// Inside the shortest silence any hop tolerates - our own socket, Lightning's +// inter-chunk timeout, and the 60s read timeout typical of proxies. +export const HEARTBEAT_INTERVAL_MS = 15_000; + const callService = ( m: ModuleDescription, port: number, @@ -126,6 +134,28 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { sendSSE(type, payload); }; + // Started before the service call so the window while Python boots + // is covered too + let heartbeat: ReturnType | undefined = + setInterval(() => { + if (isClosed) { + return; + } + try { + controller.enqueue(textEncoder.encode(HEARTBEAT_FRAME)); + } catch (error) { + // consumer went away between ticks + isClosed = true; + } + }, HEARTBEAT_INTERVAL_MS); + + const stopHeartbeat = () => { + if (heartbeat) { + clearInterval(heartbeat); + heartbeat = undefined; + } + }; + try { const result = await callService( m, @@ -146,13 +176,18 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { error instanceof Error ? error.message : "Unknown error", }); } finally { + stopHeartbeat(); console.log( `STREAM COMPLETE ${ctx.uuid} in ${ - (new Date() - ctx.start) / 1000 + (Date.now() - ctx.start) / 1000 }s` ); isClosed = true; - controller.close(); + try { + controller.close(); + } catch (error) { + // already closed from the consumer's side + } } }, }); diff --git a/platform/test/middleware/heartbeat.test.ts b/platform/test/middleware/heartbeat.test.ts new file mode 100644 index 00000000..7577f017 --- /dev/null +++ b/platform/test/middleware/heartbeat.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "bun:test"; + +import { + HEARTBEAT_FRAME, + HEARTBEAT_INTERVAL_MS, +} from "../../src/middleware/services"; + +describe("SSE heartbeat", () => { + // Lightning's SSE decoder matches `": " <> comment` and has no catch-all + // clause, so ":ping" raises FunctionClauseError inside its stream fold. + it("is a comment frame with a space after the colon", () => { + expect(HEARTBEAT_FRAME.startsWith(": ")).toBe(true); + }); + + // Without the blank line the comment sits in the decoder's buffer and resets + // nobody's idle timer. + it("terminates the frame so it is dispatched rather than buffered", () => { + expect(HEARTBEAT_FRAME.endsWith("\n\n")).toBe(true); + }); + + // An event frame would reach Lightning's handle_sse_event and need a clause. + it("carries no event or data field", () => { + expect(HEARTBEAT_FRAME).not.toContain("event:"); + expect(HEARTBEAT_FRAME).not.toContain("data:"); + }); + + it("ticks well inside the shortest silence any hop tolerates", () => { + expect(HEARTBEAT_INTERVAL_MS).toBeLessThanOrEqual(20_000); + expect(HEARTBEAT_INTERVAL_MS).toBeGreaterThanOrEqual(5_000); + }); +}); From 8a5b925589658ab1047568d0c65cc090536af647 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sat, 15 Aug 2026 22:31:04 +0000 Subject: [PATCH 03/23] Check the keepalive reaches the wire The existing tests assert the frame's shape, which passes whether or not anything ever emits one. These read frames off a live stream. The interval is read per request from APOLLO_HEARTBEAT_INTERVAL_MS, so a test can turn it down and it can be tuned in production without a release. --- platform/src/middleware/services.ts | 56 +++++++++----- .../test/middleware/heartbeat-live.test.ts | 74 +++++++++++++++++++ 2 files changed, 111 insertions(+), 19 deletions(-) create mode 100644 platform/test/middleware/heartbeat-live.test.ts diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index e244ecee..d6e1e84e 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -19,6 +19,18 @@ export const HEARTBEAT_FRAME = ": ping\n\n"; // inter-chunk timeout, and the 60s read timeout typical of proxies. export const HEARTBEAT_INTERVAL_MS = 15_000; +// Read per request rather than at import, so it can be turned down without a +// release if a hop turns out to be less patient than we thought. Zero and +// negatives are rejected rather than passed to setInterval, which would treat +// them as "every tick". +export const heartbeatIntervalMs = (): number => { + const configured = Number(process.env.APOLLO_HEARTBEAT_INTERVAL_MS); + + return Number.isFinite(configured) && configured > 0 + ? configured + : HEARTBEAT_INTERVAL_MS; +}; + const callService = ( m: ModuleDescription, port: number, @@ -110,6 +122,15 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { async start(controller) { let isClosed = false; + let heartbeat: ReturnType | undefined; + + const stopHeartbeat = () => { + if (heartbeat) { + clearInterval(heartbeat); + heartbeat = undefined; + } + }; + const sendSSE = (event: string, data: any) => { if (isClosed) { return; @@ -121,8 +142,11 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { // console.log(message.trim()); controller.enqueue(textEncoder.encode(message)); } catch (error) { - // Stream may have been closed + // A throwing enqueue is how a dropped connection reaches us + // when the runtime has not called cancel(), so stop ticking + // here too rather than waiting out the interval. isClosed = true; + stopHeartbeat(); } }; @@ -136,25 +160,19 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { // Started before the service call so the window while Python boots // is covered too - let heartbeat: ReturnType | undefined = - setInterval(() => { - if (isClosed) { - return; - } - try { - controller.enqueue(textEncoder.encode(HEARTBEAT_FRAME)); - } catch (error) { - // consumer went away between ticks - isClosed = true; - } - }, HEARTBEAT_INTERVAL_MS); - - const stopHeartbeat = () => { - if (heartbeat) { - clearInterval(heartbeat); - heartbeat = undefined; + heartbeat = setInterval(() => { + if (isClosed) { + stopHeartbeat(); + return; } - }; + try { + controller.enqueue(textEncoder.encode(HEARTBEAT_FRAME)); + } catch (error) { + // consumer went away between ticks + isClosed = true; + stopHeartbeat(); + } + }, heartbeatIntervalMs()); try { const result = await callService( diff --git a/platform/test/middleware/heartbeat-live.test.ts b/platform/test/middleware/heartbeat-live.test.ts new file mode 100644 index 00000000..8f33d6ef --- /dev/null +++ b/platform/test/middleware/heartbeat-live.test.ts @@ -0,0 +1,74 @@ +import { afterEach, describe, expect, it } from "bun:test"; + +import setup from "../../src/server"; +import { InstanceAuth } from "../../src/auth/instance-auth"; +import { HEARTBEAT_FRAME } from "../../src/middleware/services"; + +// The frame-shape tests next door pass whether or not anything ever emits one. +// These read the wire. +const port = 9871; + +const auth = new InstanceAuth({ lookup: () => null, hasGlobalKey: true }); +const app = await setup(port, auth); + +const previous = process.env.APOLLO_HEARTBEAT_INTERVAL_MS; + +afterEach(() => { + if (previous === undefined) { + delete process.env.APOLLO_HEARTBEAT_INTERVAL_MS; + } else { + process.env.APOLLO_HEARTBEAT_INTERVAL_MS = previous; + } +}); + +const readStream = async (intervalMs: string) => { + process.env.APOLLO_HEARTBEAT_INTERVAL_MS = intervalMs; + + const response = await app.handle( + new Request(`http://localhost:${port}/services/echo/stream`, { + method: "POST", + body: JSON.stringify({ message: "hello" }), + headers: { "Content-Type": "application/json" }, + }) + ); + + expect(response.status).toBe(200); + + return await response.text(); +}; + +describe("SSE heartbeat on a live stream", () => { + // Spawning Python takes long enough that a heartbeat this fast must tick + // before the service produces anything. The interval is deliberately set + // rather than waited out: the real one is 15s. + it("emits heartbeats while the service is still starting up", async () => { + const body = await readStream("20"); + + const beats = body.split(HEARTBEAT_FRAME).length - 1; + + expect(beats).toBeGreaterThan(0); + expect(body).toContain("event: complete"); + }, 30_000); + + // Guards the ordering the whole design rests on: the first thing on the wire + // is a heartbeat, not the result, so no hop sees an idle socket. + it("gets a heartbeat onto the wire before the result", async () => { + const body = await readStream("20"); + + const firstBeat = body.indexOf(HEARTBEAT_FRAME); + const result = body.indexOf("event: complete"); + + // Both have to be present before comparing them: indexOf gives -1 for a + // heartbeat that never arrived, which would sail past a bare <. + expect(firstBeat).toBeGreaterThanOrEqual(0); + expect(result).toBeGreaterThanOrEqual(0); + expect(firstBeat).toBeLessThan(result); + }, 30_000); + + it("sends none when the interval outlasts the request", async () => { + const body = await readStream("600000"); + + expect(body).not.toContain(HEARTBEAT_FRAME); + expect(body).toContain("event: complete"); + }, 30_000); +}); From 93ef02473ccba1137e5be66a3f9c67849296f6b3 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sat, 15 Aug 2026 01:07:52 +0000 Subject: [PATCH 04/23] Say what actually went wrong when a service fails Three ways a failing service lost its own diagnosis on the way out. The bridge rejected with a bare exit code. A number is not an Error, so the handler downstream fell through to its fallback and sent the client the string "Unknown error" - no code, no type, and the exit code we did have was logged and discarded. It also rejected without returning, so a non-zero exit went on to resolve as well. A failed spawn was only logged. Nothing settled the promise, so poetry missing from PATH did not look like a broken install, it looked like a service that never answered, and the request hung until something upstream gave up. Exiting cleanly with an empty output file resolved as null, which was sent to the client as a successful completion carrying nothing. Lightning could then only report it as a stream that ended without a response. Failures now carry a code, a type and the detail worth keeping, through one envelope shared by the streaming and synchronous routes. The synchronous route had no error handling at all, so a rejected run escaped it and became a bare framework 500. Two of those empty output files came from entry.py itself: three error paths returned before the write, and two of them referenced an unbound exception variable, so a missing input file raised NameError rather than reporting the missing file. Every path out now goes through one write. ruff confirms it: two F821 undefined-name findings on main, gone here. --- .changeset/typed-service-failures.md | 6 +++ platform/src/bridge.ts | 37 ++++++++++----- platform/src/middleware/services.ts | 66 ++++++++++++++++++++++---- platform/src/util/errors.ts | 69 ++++++++++++++++++++++++++++ platform/test/util/errors.test.ts | 61 ++++++++++++++++++++++++ services/entry.py | 28 +++++++++-- 6 files changed, 240 insertions(+), 27 deletions(-) create mode 100644 .changeset/typed-service-failures.md create mode 100644 platform/test/util/errors.test.ts diff --git a/.changeset/typed-service-failures.md b/.changeset/typed-service-failures.md new file mode 100644 index 00000000..6d6cba61 --- /dev/null +++ b/.changeset/typed-service-failures.md @@ -0,0 +1,6 @@ +--- +"apollo": patch +--- + +Report service failures with a real code and message instead of "Unknown error", +and stop a failed spawn from hanging the request diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index d5e13d0c..9d957481 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -3,6 +3,11 @@ import path from "node:path"; import { spawn } from "node:child_process"; import { rm } from "node:fs/promises"; import { getInternalToken } from "./auth/internal-token"; +import { + emptyResult, + subprocessFailed, + subprocessSpawnFailed, +} from "./util/errors"; import pkg from "../../package.json"; /** @@ -56,8 +61,11 @@ export const run = async ( } ); - proc.on("error", async (err) => { - console.log(err); + // Nothing was spawned, so no "close" is coming - without settling here the + // request stays open until something upstream gives up + proc.on("error", (err) => { + console.error("Failed to start python process", err); + reject(subprocessSpawnFailed(scriptName, err)); }); const rl = readline.createInterface({ @@ -101,12 +109,10 @@ export const run = async ( rl.close(); rl2.close(); - if (code) { - console.error("Python process exited with code", code); - reject(code); - } - const result = Bun.file(outputPath); - const text = await result.text(); + // Read before cleaning up, and clean up on every exit path + const text = await Bun.file(outputPath) + .text() + .catch(() => ""); try { await rm(inputPath); @@ -116,12 +122,19 @@ export const run = async ( console.error(e); } + if (code) { + console.error("Python process exited with code", code); + return reject(subprocessFailed(scriptName, code)); + } + if (text) { - resolve(JSON.parse(text)); - } else { - console.warn("No data returned from pythonland"); - resolve(null); + return resolve(JSON.parse(text)); } + + // entry.py writes a result on every path it completes, including its own + // error envelopes, so an empty file means the run died + console.warn("No data returned from pythonland"); + return reject(emptyResult(scriptName)); }); return; diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index d6e1e84e..1cfaff54 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -6,7 +6,11 @@ import { run } from "../bridge"; import describeModules, { type ModuleDescription, } from "../util/describe-modules"; -import { isApolloError } from "../util/errors"; +import { + ApolloThrowable, + isApolloError, + type ApolloError, +} from "../util/errors"; import type { InstanceAuth } from "../auth/instance-auth"; const textEncoder = new TextEncoder(); @@ -31,6 +35,30 @@ export const heartbeatIntervalMs = (): number => { : HEARTBEAT_INTERVAL_MS; }; +/** Normalise anything thrown by a service run into the ApolloError envelope, so + * a caller sees the same shape whether the failure was typed or not. */ +const toErrorPayload = (error: unknown): ApolloError => { + if (error instanceof ApolloThrowable) { + return error.toJSON(); + } + // Rebuilt field by field rather than returned as-is: isApolloError only + // checks for a numeric `code`, and anything else hanging off the object + // would be serialised to the caller along with it. + if (isApolloError(error)) { + return { + code: error.code, + type: error.type, + message: error.message, + ...(error.details === undefined ? {} : { details: error.details }), + }; + } + return { + code: 500, + type: "INTERNAL_ERROR", + message: error instanceof Error ? error.message : String(error), + }; +}; + const callService = ( m: ModuleDescription, port: number, @@ -99,7 +127,19 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { app.post(name, async (ctx) => { console.log(`POST /services/${name}: ${ctx.uuid}`); const payload = buildPayload(ctx); - const result = await callService(m, port, payload as any); + + let result: any; + try { + result = await callService(m, port, payload as any); + } catch (error) { + const payload = toErrorPayload(error); + return new Response(JSON.stringify(payload), { + status: payload.code, + headers: { + "Content-Type": "application/json", + }, + }); + } if (isApolloError(result)) { return new Response(JSON.stringify(result), { @@ -189,10 +229,7 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { sendSSE("complete", result); } } catch (error) { - sendSSE("error", { - message: - error instanceof Error ? error.message : "Unknown error", - }); + sendSSE("error", toErrorPayload(error)); } finally { stopHeartbeat(); console.log( @@ -255,14 +292,23 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { const base: Record = { ...(message.data ?? {}) }; const payload = applyKey(base, ws.data); - callService(m, port, payload as any, onLog, onEvent).then( - (result) => { + // The catch matters as much as the then: a run that rejects + // (spawn failure, empty output) would otherwise leave the client + // waiting on a frame that never comes. The try around this only + // sees synchronous throws. + callService(m, port, payload as any, onLog, onEvent) + .then((result) => { ws.send({ event: "complete", data: result, }); - } - ); + }) + .catch((error) => { + ws.send({ + event: "error", + data: toErrorPayload(error), + }); + }); } } catch (e) { console.log(e); diff --git a/platform/src/util/errors.ts b/platform/src/util/errors.ts index f967acb4..cfd7fcbc 100644 --- a/platform/src/util/errors.ts +++ b/platform/src/util/errors.ts @@ -22,6 +22,75 @@ export function apolloError( return { code, type, message, ...(details ? { details } : {}) }; } +/** An ApolloError that can also be thrown, so a catch block gets a real Error + * while the wire shape stays the same. `toJSON` is required: JSON.stringify on + * an Error is `{}` without it. */ +export class ApolloThrowable extends Error implements ApolloError { + readonly code: number; + readonly type: string; + readonly details?: Record; + + constructor( + code: number, + type: string, + message: string, + details?: Record + ) { + super(message); + this.name = "ApolloThrowable"; + this.code = code; + this.type = type; + this.details = details; + } + + toJSON(): ApolloError { + return { + code: this.code, + type: this.type, + message: this.message, + ...(this.details ? { details: this.details } : {}), + }; + } +} + +/** The service process exited non-zero. */ +export function subprocessFailed( + service: string, + exitCode: number +): ApolloThrowable { + return new ApolloThrowable( + 500, + "SUBPROCESS_FAILED", + `Service "${service}" exited with code ${exitCode}`, + { service, exitCode } + ); +} + +/** We never got as far as running the service - poetry or python missing, or + * the spawn refused. */ +export function subprocessSpawnFailed( + service: string, + cause: unknown +): ApolloThrowable { + return new ApolloThrowable( + 500, + "SUBPROCESS_SPAWN_FAILED", + `Service "${service}" could not be started`, + { service, cause: cause instanceof Error ? cause.message : String(cause) } + ); +} + +/** The service exited cleanly but wrote nothing. entry.py writes a result on + * every path it completes, so an empty file means the run died. */ +export function emptyResult(service: string): ApolloThrowable { + return new ApolloThrowable( + 502, + "EMPTY_RESULT", + `Service "${service}" finished without producing a result`, + { service } + ); +} + export function unauthorized(ctx: any): ApolloError { return apolloError(ctx, 401, "UNAUTHORIZED", "Missing or invalid API key"); } diff --git a/platform/test/util/errors.test.ts b/platform/test/util/errors.test.ts new file mode 100644 index 00000000..3f317a48 --- /dev/null +++ b/platform/test/util/errors.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it } from "bun:test"; + +import { + ApolloThrowable, + emptyResult, + isApolloError, + subprocessFailed, + subprocessSpawnFailed, +} from "../../src/util/errors"; + +describe("ApolloThrowable", () => { + // JSON.stringify on an Error is "{}", so without toJSON the synchronous + // service route answers a failure with the right status and an empty body. + it("survives JSON.stringify with its envelope intact", () => { + const parsed = JSON.parse(JSON.stringify(subprocessFailed("job_chat", 3))); + + expect(parsed.code).toBe(500); + expect(parsed.type).toBe("SUBPROCESS_FAILED"); + expect(parsed.message).toContain("job_chat"); + expect(parsed.details.exitCode).toBe(3); + }); + + it("is recognised by isApolloError, so existing envelope handling applies", () => { + expect(isApolloError(subprocessFailed("echo", 1))).toBe(true); + expect(isApolloError(emptyResult("echo"))).toBe(true); + }); + + it("is a real Error, so it can be thrown and caught normally", () => { + const error = emptyResult("workflow_chat"); + + expect(error instanceof Error).toBe(true); + expect(error instanceof ApolloThrowable).toBe(true); + expect(error.message.length).toBeGreaterThan(0); + }); +}); + +describe("subprocess failures", () => { + it("keeps the exit code", () => { + expect(subprocessFailed("job_chat", 137).details?.exitCode).toBe(137); + }); + + // A spawn failure means poetry or python is missing, which is an operator + // problem rather than a service one. + it("distinguishes never-started from started-and-failed", () => { + expect(subprocessSpawnFailed("echo", new Error("ENOENT")).type).toBe( + "SUBPROCESS_SPAWN_FAILED" + ); + expect(subprocessFailed("echo", 1).type).toBe("SUBPROCESS_FAILED"); + }); + + // It ran and exited cleanly; what came back was unusable. + it("reports an empty result as a bad gateway, not a server error", () => { + expect(emptyResult("echo").code).toBe(502); + }); + + it("carries the spawn cause as a string, not a nested Error", () => { + const details = subprocessSpawnFailed("echo", new Error("ENOENT")).details; + + expect(details?.cause).toBe("ENOENT"); + }); +}); diff --git a/services/entry.py b/services/entry.py index 53ac0e68..6225a85b 100644 --- a/services/entry.py +++ b/services/entry.py @@ -73,19 +73,28 @@ def call( try: with open(input_path, "r") as f: data = json.load(f) - except FileNotFoundError: + except FileNotFoundError as e: sentry_sdk.capture_exception(e) - return ApolloError(code=500, message=f"Input file not found: {input_path}", type="INTERNAL_ERROR").to_dict() - except json.JSONDecodeError: + return _finish( + ApolloError(code=500, message=f"Input file not found: {input_path}", type="INTERNAL_ERROR").to_dict(), + output_path, + ) + except json.JSONDecodeError as e: sentry_sdk.capture_exception(e) - return ApolloError(code=500, message="Invalid JSON input", type="INTERNAL_ERROR").to_dict() + return _finish( + ApolloError(code=500, message="Invalid JSON input", type="INTERNAL_ERROR").to_dict(), + output_path, + ) try: m = __import__(module_name, fromlist=["main"]) result = m.main(data) except ModuleNotFoundError as e: sentry_sdk.capture_exception(e) - return ApolloError(code=500, message=str(e), type="INTERNAL_ERROR").to_dict() + return _finish( + ApolloError(code=500, message=str(e), type="INTERNAL_ERROR").to_dict(), + output_path, + ) except ApolloError as e: sentry_sdk.capture_exception(e) result = e.to_dict() @@ -95,6 +104,15 @@ def call( langfuse.flush() + return _finish(result, output_path) + + +def _finish(result: dict, output_path: str | None) -> dict: + """Write the result where the caller expects it, then hand it back. + + Every path out of run_service goes through here, so an empty output file + means the run died rather than that it failed politely. + """ if output_path: with open(output_path, "w") as f: json.dump(result, f) From 18e828b412a755a7cdc58393e4c5bbf832ebd179 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sat, 15 Aug 2026 01:13:22 +0000 Subject: [PATCH 05/23] Stop a run when the client goes away Nothing told us the caller had left. The python child kept running, kept calling the model, and wrote its answer into a socket with nobody on the other end. On a planner request that can be another ten model calls after the user closed the panel, all of them billed. The stream now signals an abort when it is cancelled, and the bridge kills the child on that signal. poetry run execs into python rather than forking it, so the pid we hold is the interpreter and a plain SIGTERM reaches it; killing it closes the socket to Anthropic, which stops generation. A hard kill follows five seconds later only for a child wedged somewhere that never sees the signal. The saving is on the streaming calls, which is where a long request spends its time. A non-streaming call has already been submitted and will be billed whether or not we are still listening. A cancelled run settles as its own kind of failure rather than a fault, so deliberately abandoning a request does not read as something breaking. The test spawns a real child, waits until python is genuinely inside the service, aborts, and checks the process is gone. Two things it had to get right to be worth having: the process list matches the poetry wrapper long before the interpreter exists, so the probe announces itself instead; and BSD pgrep has no count flag, so asking for one reads as "no processes" and passes regardless. Checked by disabling the abort listener - the test then takes 90 seconds and reports EMPTY_RESULT. --- .changeset/cancel-abandoned-runs.md | 6 ++ platform/src/bridge.ts | 54 ++++++++++++++- platform/src/middleware/services.ts | 60 +++++++++++----- platform/src/util/errors.ts | 14 ++++ platform/test/bridge-cancel.test.ts | 92 +++++++++++++++++++++++++ platform/test/util/errors.test.ts | 11 +++ services/_cancel_probe/_cancel_probe.py | 20 ++++++ 7 files changed, 237 insertions(+), 20 deletions(-) create mode 100644 .changeset/cancel-abandoned-runs.md create mode 100644 platform/test/bridge-cancel.test.ts create mode 100644 services/_cancel_probe/_cancel_probe.py diff --git a/.changeset/cancel-abandoned-runs.md b/.changeset/cancel-abandoned-runs.md new file mode 100644 index 00000000..b29cee49 --- /dev/null +++ b/.changeset/cancel-abandoned-runs.md @@ -0,0 +1,6 @@ +--- +"apollo": patch +--- + +Stop a service run when the client disconnects, so an abandoned request no +longer keeps calling the model diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index 9d957481..2aaa8478 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -5,6 +5,7 @@ import { rm } from "node:fs/promises"; import { getInternalToken } from "./auth/internal-token"; import { emptyResult, + subprocessCancelled, subprocessFailed, subprocessSpawnFailed, } from "./util/errors"; @@ -22,7 +23,9 @@ export const run = async ( port: number, // needed for self-calling services in pythonland args: any = {}, onLog?: (str: string) => void, - onEvent?: (type: string, payload: any /* string or json tbh */) => void + onEvent?: (type: string, payload: any /* string or json tbh */) => void, + // Aborted when the client goes away + signal?: AbortSignal ) => { return new Promise(async (resolve, reject) => { const id = crypto.randomUUID(); @@ -68,6 +71,32 @@ export const run = async ( reject(subprocessSpawnFailed(scriptName, err)); }); + // `poetry run` execs into python rather than forking it, so this pid is the + // interpreter and a plain signal reaches it. Killing it closes the socket to + // Anthropic, which stops generation on the streaming calls; a non-streaming + // call is already submitted and gets billed whatever we do here. + let cancelled = false; + let hardKill: ReturnType | undefined; + + const onAbort = () => { + cancelled = true; + console.warn(`cancelling ${scriptName}: client went away`); + proc.kill("SIGTERM"); + + // Python installs no SIGTERM handler, so termination is immediate. This + // is only for a child wedged somewhere that never sees it. + hardKill = setTimeout(() => proc.kill("SIGKILL"), 5_000); + hardKill.unref?.(); + }; + + if (signal) { + if (signal.aborted) { + onAbort(); + } else { + signal.addEventListener("abort", onAbort, { once: true }); + } + } + const rl = readline.createInterface({ input: proc.stdout, crlfDelay: Infinity, @@ -104,11 +133,16 @@ export const run = async ( onLog?.(line); }); - proc.on("close", async (code) => { + proc.on("close", async (code, closeSignal) => { // Clean up readline interfaces immediately to prevent race conditions rl.close(); rl2.close(); + if (hardKill) { + clearTimeout(hardKill); + } + signal?.removeEventListener("abort", onAbort); + // Read before cleaning up, and clean up on every exit path const text = await Bun.file(outputPath) .text() @@ -122,13 +156,27 @@ export const run = async ( console.error(e); } + // We killed it on purpose, so this is not a service failure + if (cancelled) { + return reject(subprocessCancelled(scriptName, closeSignal ?? "SIGTERM")); + } + if (code) { console.error("Python process exited with code", code); return reject(subprocessFailed(scriptName, code)); } if (text) { - return resolve(JSON.parse(text)); + // Parsed inside the try: this handler is async, so a throw here + // becomes an unhandled rejection and the run never settles at all. + // A half-written file is what a crash mid-dump leaves behind. + try { + return resolve(JSON.parse(text)); + } catch (e) { + console.error(`Unreadable output from ${scriptName}`); + console.error(e); + return reject(emptyResult(scriptName)); + } } // entry.py writes a result on every path it completes, including its own diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index 1cfaff54..fe174f6d 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -64,12 +64,14 @@ const callService = ( port: number, payload?: any, onLog?: (str: string) => void, - onEvent?: (evt: string, payload: any) => void + onEvent?: (evt: string, payload: any) => void, + signal?: AbortSignal ) => { if (m.type === "py") { - return run(m.name, port, payload as any, onLog, onEvent); + return run(m.name, port, payload as any, onLog, onEvent, signal); } else { // TODO add event handling to ts services + // TODO ts services can't be cancelled - the handler signature has no signal return m.handler!(port, payload as any, onLog); } }; @@ -158,19 +160,21 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { console.log(`STREAM START /services/${name}: ${ctx.uuid}`); const payload = buildPayload(ctx); - const stream = new ReadableStream({ - async start(controller) { - let isClosed = false; + const abort = new AbortController(); - let heartbeat: ReturnType | undefined; + // Hoisted so cancel() can reach what start() set up + let isClosed = false; + let heartbeat: ReturnType | undefined; - const stopHeartbeat = () => { - if (heartbeat) { - clearInterval(heartbeat); - heartbeat = undefined; - } - }; + const stopHeartbeat = () => { + if (heartbeat) { + clearInterval(heartbeat); + heartbeat = undefined; + } + }; + const stream = new ReadableStream({ + async start(controller) { const sendSSE = (event: string, data: any) => { if (isClosed) { return; @@ -182,11 +186,11 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { // console.log(message.trim()); controller.enqueue(textEncoder.encode(message)); } catch (error) { - // A throwing enqueue is how a dropped connection reaches us - // when the runtime has not called cancel(), so stop ticking - // here too rather than waiting out the interval. + // Same as the heartbeat's catch: a throwing enqueue is a + // dropped connection the runtime has not told us about. isClosed = true; stopHeartbeat(); + abort.abort(); } }; @@ -208,9 +212,12 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { try { controller.enqueue(textEncoder.encode(HEARTBEAT_FRAME)); } catch (error) { - // consumer went away between ticks + // The consumer went away between ticks. cancel() may never + // fire for this, so end the run here rather than leaving the + // child generating for nobody. isClosed = true; stopHeartbeat(); + abort.abort(); } }, heartbeatIntervalMs()); @@ -220,7 +227,8 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { port, payload as any, onLog, - onEvent + onEvent, + abort.signal ); if (isApolloError(result)) { @@ -245,6 +253,24 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { } } }, + + // Everything from here on is work nobody will read + cancel(reason) { + isClosed = true; + stopHeartbeat(); + console.warn( + `STREAM CANCELLED ${ctx.uuid} after ${ + (Date.now() - ctx.start) / 1000 + }s` + ); + abort.abort(reason); + }, + }); + + // cancel() depends on the runtime noticing the dropped connection, so + // listen on the request's own signal too. abort() is idempotent. + ctx.request?.signal?.addEventListener("abort", () => abort.abort(), { + once: true, }); return new Response(stream, { diff --git a/platform/src/util/errors.ts b/platform/src/util/errors.ts index cfd7fcbc..0b63e47b 100644 --- a/platform/src/util/errors.ts +++ b/platform/src/util/errors.ts @@ -80,6 +80,20 @@ export function subprocessSpawnFailed( ); } +/** We stopped the service ourselves because the client went away. 499 keeps + * these out of the 5xx that mean something is actually broken. */ +export function subprocessCancelled( + service: string, + signal: string +): ApolloThrowable { + return new ApolloThrowable( + 499, + "SUBPROCESS_CANCELLED", + `Service "${service}" was cancelled because the client disconnected`, + { service, signal } + ); +} + /** The service exited cleanly but wrote nothing. entry.py writes a result on * every path it completes, so an empty file means the run died. */ export function emptyResult(service: string): ApolloThrowable { diff --git a/platform/test/bridge-cancel.test.ts b/platform/test/bridge-cancel.test.ts new file mode 100644 index 00000000..1609dafb --- /dev/null +++ b/platform/test/bridge-cancel.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, it } from "bun:test"; + +import { run } from "../src/bridge"; +import { ApolloThrowable } from "../src/util/errors"; + +const PORT = 9871; + +// The spawned command is `poetry run python ...`, so matching the process list +// alone hits the poetry wrapper seconds before the interpreter exists. The probe +// announces itself once python is really inside the service. +const started = () => { + let resolve!: () => void; + const promise = new Promise((r) => { + resolve = r; + }); + + return { + promise, + onEvent: (type: string) => { + if (type === "probe_started") { + resolve(); + } + }, + }; +}; + +// No `pgrep -c`: BSD pgrep has no count flag, and its usage error goes to +// stderr, so asking for one reads as "no processes". +const probeProcessCount = async () => { + const proc = Bun.spawn(["pgrep", "-f", "_cancel_probe"], { + stdout: "pipe", + stderr: "pipe", + }); + const out = await new Response(proc.stdout).text(); + return out.trim().split("\n").filter(Boolean).length; +}; + +const waitForProbesToClear = async (limitMs = 15_000) => { + const deadline = Date.now() + limitMs; + while (Date.now() < deadline) { + if ((await probeProcessCount()) === 0) { + return true; + } + await Bun.sleep(200); + } + return false; +}; + +describe("cancelling a service run", () => { + it("kills the running python child when the caller aborts", async () => { + const abort = new AbortController(); + const probe = started(); + + const pending = run( + "_cancel_probe", + PORT, + { sleep_for: 120 }, + undefined, + probe.onEvent, + abort.signal + ); + + await probe.promise; + expect(await probeProcessCount()).toBeGreaterThan(0); + + abort.abort(); + + const error = (await pending.catch((e) => e)) as ApolloThrowable; + expect(error).toBeInstanceOf(ApolloThrowable); + expect(error.type).toBe("SUBPROCESS_CANCELLED"); + expect(error.code).toBe(499); + + expect(await waitForProbesToClear()).toBe(true); + }, 90_000); + + it("does not leave a child running when the signal is already aborted", async () => { + const abort = new AbortController(); + abort.abort(); + + const error = (await run( + "_cancel_probe", + PORT, + { sleep_for: 120 }, + undefined, + undefined, + abort.signal + ).catch((e) => e)) as ApolloThrowable; + + expect(error.type).toBe("SUBPROCESS_CANCELLED"); + expect(await waitForProbesToClear()).toBe(true); + }, 90_000); +}); diff --git a/platform/test/util/errors.test.ts b/platform/test/util/errors.test.ts index 3f317a48..0e442ada 100644 --- a/platform/test/util/errors.test.ts +++ b/platform/test/util/errors.test.ts @@ -4,6 +4,7 @@ import { ApolloThrowable, emptyResult, isApolloError, + subprocessCancelled, subprocessFailed, subprocessSpawnFailed, } from "../../src/util/errors"; @@ -58,4 +59,14 @@ describe("subprocess failures", () => { expect(details?.cause).toBe("ENOENT"); }); + + it("keeps a deliberate cancellation out of the 5xx range", () => { + // We stopped this one ourselves because the caller left. Reporting it as a + // server error would bury the failures that mean something is broken. + const cancelled = subprocessCancelled("global_chat", "SIGTERM"); + + expect(cancelled.code).toBe(499); + expect(cancelled.type).toBe("SUBPROCESS_CANCELLED"); + expect(cancelled.details?.signal).toBe("SIGTERM"); + }); }); diff --git a/services/_cancel_probe/_cancel_probe.py b/services/_cancel_probe/_cancel_probe.py new file mode 100644 index 00000000..f06bd21c --- /dev/null +++ b/services/_cancel_probe/_cancel_probe.py @@ -0,0 +1,20 @@ +"""A service that does nothing slowly, so cancellation can be tested. + +The leading underscore keeps it out of describe-modules, so it is never mounted +and has no route. + +It announces itself before sleeping because the spawned command is +`poetry run python ...`: the process list matches on the poetry wrapper seconds +before the interpreter has booted, so waiting on that would prove nothing. +""" + +import time + + +def main(data_dict: dict) -> dict: + print("EVENT:probe_started:{}", flush=True) # noqa: T201 + + seconds = data_dict.get("sleep_for", 30) + time.sleep(seconds) + + return {"slept": seconds} From 11c429c69d97f85d7b2eb891cf9d057f99903c99 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 19:25:31 +0000 Subject: [PATCH 06/23] Close what the third review found The async Promise executor was the same trap as the JSON.parse one, a hundred lines above it. The constructor only catches a synchronous throw, so a failing Bun.write left run() pending for ever and the caller's stream open. Setup happens outside the promise now, where run() being async is enough, and the input file - which holds the key - is removed if the second write fails. Cancellation only ever covered the streaming route. A plain POST and a websocket both left the model generating when the caller went away, which is the cost the change exists to stop. Both pass a signal now; the socket gets its own controller and a close handler to fire it. A child killed by a signal reports a null exit code, so the OOM killer - the likeliest way a service dies without exiting - was reported as an empty result with the signal thrown away. Every stderr line was forwarded to the caller with no filter, so an interpreter traceback carrying server paths and whatever a frame held went straight out. Stderr now follows the same rule as stdout: only what a service logged deliberately. The health check calls run() too, and was the one caller that never got a catch when run() started rejecting. Also: the heartbeat interval had a floor but no ceiling, and setInterval turns a delay past 2^31-1 into every tick - so the value someone picks to mean "effectively never" would have flooded every open stream. And the WS handler used .then().catch(), which also catches a throw from the success path and reports it as a service failure. --- platform/src/bridge.ts | 52 ++++++++--- platform/src/middleware/healthcheck.tsx | 21 ++++- platform/src/middleware/services.ts | 104 ++++++++++++--------- platform/src/util/errors.ts | 39 ++++++++ platform/test/middleware/heartbeat.test.ts | 37 +++++++- services/entry.py | 10 +- 6 files changed, 203 insertions(+), 60 deletions(-) diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index 2aaa8478..8e8c3c53 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -7,10 +7,15 @@ import { emptyResult, subprocessCancelled, subprocessFailed, + subprocessKilled, subprocessSpawnFailed, } from "./util/errors"; import pkg from "../../package.json"; +// A line a service logged on purpose, as opposed to whatever else lands on a +// stream. Only these are forwarded to the caller. +const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/; + /** Run a python script Each script will be run in its own thread because @@ -27,20 +32,29 @@ export const run = async ( // Aborted when the client goes away signal?: AbortSignal ) => { - return new Promise(async (resolve, reject) => { - const id = crypto.randomUUID(); + const id = crypto.randomUUID(); - const tmpfile = path.resolve(`tmp/data/${id}-{}.json`); + const tmpfile = path.resolve(`tmp/data/${id}-{}.json`); - const inputPath = tmpfile.replace("{}", "input"); - const outputPath = tmpfile.replace("{}", "output"); + const inputPath = tmpfile.replace("{}", "input"); + const outputPath = tmpfile.replace("{}", "output"); - // console.log("Initing input file at", inputPath); + // Outside the promise, deliberately. The Promise constructor only catches a + // synchronous throw from its executor, so an await that rejects in there - + // a full disk, a read-only tmp - leaves the promise pending for ever and + // the caller's stream open. Out here, run() is async and simply rejects. + try { await Bun.write(inputPath, JSON.stringify(args)); - - // console.log("Initing output file at", outputPath); await Bun.write(outputPath, ""); - + } catch (error) { + // The input file holds the key, so it does not get left behind on a + // half-finished setup. + await rm(inputPath).catch(() => {}); + await rm(outputPath).catch(() => {}); + throw subprocessSpawnFailed(scriptName, error); + } + + return new Promise((resolve, reject) => { const proc = spawn( "poetry", [ @@ -103,7 +117,7 @@ export const run = async ( }); rl.on("line", (line) => { // Then divert any logs from a logger object to the websocket - if (/^(INFO|DEBUG|ERROR|WARNING)\:/.test(line)) { + if (LOG_LINE.test(line)) { // Divert the log line locally console.log(line); // TODO I'd love to break the log line up in to JSON actually @@ -129,8 +143,14 @@ export const run = async ( }); rl2.on("line", (line) => { console.error(line); - // /Divert all errors to the websocket - onLog?.(line); + + // Only forward what a service logged deliberately, the same rule stdout + // follows. Everything else on stderr is the interpreter talking: raw + // tracebacks carrying server paths, source lines, and whatever a frame + // held - which for a service is the payload. + if (LOG_LINE.test(line)) { + onLog?.(line); + } }); proc.on("close", async (code, closeSignal) => { @@ -166,6 +186,14 @@ export const run = async ( return reject(subprocessFailed(scriptName, code)); } + // A child killed by a signal reports a null code, so without this the + // OOM killer - the likeliest way a service dies without exiting - would + // be reported as an empty result and the signal thrown away. + if (closeSignal) { + console.error(`Python process killed by ${closeSignal}`); + return reject(subprocessKilled(scriptName, closeSignal)); + } + if (text) { // Parsed inside the try: this handler is async, so a throw here // becomes an unhandled rejection and the run never settles at all. diff --git a/platform/src/middleware/healthcheck.tsx b/platform/src/middleware/healthcheck.tsx index a6341496..2391e888 100644 --- a/platform/src/middleware/healthcheck.tsx +++ b/platform/src/middleware/healthcheck.tsx @@ -1,6 +1,7 @@ import { Elysia } from "elysia"; import pkg from "../../../package.json" assert { type: "json" }; import { run } from '../bridge'; +import { toErrorPayload } from '../util/errors'; export default async (app: Elysia) => { app.get("/livez", () => { @@ -12,8 +13,24 @@ export default async (app: Elysia) => { }); }); app.get("/status", async () => { - const status = await run ('status', 0, {} as any) as any; - return new Response(status, { + // run() rejects now where it used to resolve null, and this is the one + // caller outside the services routes. Without the catch a spawn failure + // leaves the route throwing, so the health endpoint answers with Elysia's + // generic 500 and reports to Sentry rather than saying what is wrong. + let status: unknown; + try { + status = await run("status", 0, {} as any); + } catch (error) { + const payload = toErrorPayload(error); + return new Response(JSON.stringify(payload), { + status: payload.code, + headers: { + "Content-Type": "application/json", + }, + }); + } + + return new Response(status as any, { status: 200, headers: { "Content-Type": "application/json", diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index fe174f6d..a11af9e0 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -6,11 +6,7 @@ import { run } from "../bridge"; import describeModules, { type ModuleDescription, } from "../util/describe-modules"; -import { - ApolloThrowable, - isApolloError, - type ApolloError, -} from "../util/errors"; +import { isApolloError, toErrorPayload } from "../util/errors"; import type { InstanceAuth } from "../auth/instance-auth"; const textEncoder = new TextEncoder(); @@ -23,42 +19,24 @@ export const HEARTBEAT_FRAME = ": ping\n\n"; // inter-chunk timeout, and the 60s read timeout typical of proxies. export const HEARTBEAT_INTERVAL_MS = 15_000; +// Anything past this is a mistake rather than a choice, and setInterval turns +// a delay over 2^31-1 into "every tick" - so the value someone reaches for to +// mean "effectively never" would flood every open stream instead. +const HEARTBEAT_MAX_MS = 120_000; + // Read per request rather than at import, so it can be turned down without a -// release if a hop turns out to be less patient than we thought. Zero and -// negatives are rejected rather than passed to setInterval, which would treat -// them as "every tick". +// release if a hop turns out to be less patient than we thought. Out-of-range +// values fall back rather than being passed to setInterval. export const heartbeatIntervalMs = (): number => { const configured = Number(process.env.APOLLO_HEARTBEAT_INTERVAL_MS); - return Number.isFinite(configured) && configured > 0 + return Number.isFinite(configured) && + configured > 0 && + configured <= HEARTBEAT_MAX_MS ? configured : HEARTBEAT_INTERVAL_MS; }; -/** Normalise anything thrown by a service run into the ApolloError envelope, so - * a caller sees the same shape whether the failure was typed or not. */ -const toErrorPayload = (error: unknown): ApolloError => { - if (error instanceof ApolloThrowable) { - return error.toJSON(); - } - // Rebuilt field by field rather than returned as-is: isApolloError only - // checks for a numeric `code`, and anything else hanging off the object - // would be serialised to the caller along with it. - if (isApolloError(error)) { - return { - code: error.code, - type: error.type, - message: error.message, - ...(error.details === undefined ? {} : { details: error.details }), - }; - } - return { - code: 500, - type: "INTERNAL_ERROR", - message: error instanceof Error ? error.message : String(error), - }; -}; - const callService = ( m: ModuleDescription, port: number, @@ -76,6 +54,11 @@ const callService = ( } }; +// The in-flight run for each open websocket, so closing the socket can stop +// it. Keyed on the socket itself and deleted as soon as the run settles, so a +// closed socket holds nothing. +const wsRuns = new WeakMap(); + export default async (app: Elysia, port: number, auth: InstanceAuth) => { console.log("Loading routes:"); const modules = await describeModules(path.resolve("./services")); @@ -132,7 +115,17 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { let result: any; try { - result = await callService(m, port, payload as any); + // The signal matters here as much as on the stream: a client that + // gives up waiting for a plain POST leaves the model generating + // just the same. + result = await callService( + m, + port, + payload as any, + undefined, + undefined, + ctx.request?.signal + ); } catch (error) { const payload = toErrorPayload(error); return new Response(JSON.stringify(payload), { @@ -295,6 +288,14 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { open() { console.log(`Websocket connected at /services/${name}`); }, + // A websocket has no request signal, so cancellation needs its own + // controller and a close handler to fire it. Without this a WS caller + // going away leaves the child generating, which is the cost the whole + // change exists to stop. + close(ws) { + wsRuns.get(ws)?.abort(); + wsRuns.delete(ws); + }, message(ws, message) { try { if (message.event === "start") { @@ -318,23 +319,40 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { const base: Record = { ...(message.data ?? {}) }; const payload = applyKey(base, ws.data); - // The catch matters as much as the then: a run that rejects - // (spawn failure, empty output) would otherwise leave the client - // waiting on a frame that never comes. The try around this only - // sees synchronous throws. - callService(m, port, payload as any, onLog, onEvent) - .then((result) => { + const abort = new AbortController(); + wsRuns.set(ws, abort); + + // Two arguments rather than a chained catch: a chain would also + // catch a throw from the success callback and report it to the + // client as a service failure. + // + // The failure handler matters as much as the success one - a run + // that rejects (spawn failure, empty output) would otherwise + // leave the client waiting on a frame that never comes, since + // the try around this only sees synchronous throws. + callService( + m, + port, + payload as any, + onLog, + onEvent, + abort.signal + ).then( + (result) => { + wsRuns.delete(ws); ws.send({ event: "complete", data: result, }); - }) - .catch((error) => { + }, + (error) => { + wsRuns.delete(ws); ws.send({ event: "error", data: toErrorPayload(error), }); - }); + } + ); } } catch (e) { console.log(e); diff --git a/platform/src/util/errors.ts b/platform/src/util/errors.ts index 0b63e47b..f8af7075 100644 --- a/platform/src/util/errors.ts +++ b/platform/src/util/errors.ts @@ -94,6 +94,21 @@ export function subprocessCancelled( ); } +/** Something outside Apollo killed the service - most often the OOM killer. + * Distinct from a cancellation, which is us, and from a non-zero exit, which + * is the service deciding to stop. */ +export function subprocessKilled( + service: string, + signal: string +): ApolloThrowable { + return new ApolloThrowable( + 500, + "SUBPROCESS_KILLED", + `Service "${service}" was killed by ${signal}`, + { service, signal } + ); +} + /** The service exited cleanly but wrote nothing. entry.py writes a result on * every path it completes, so an empty file means the run died. */ export function emptyResult(service: string): ApolloThrowable { @@ -126,3 +141,27 @@ export function clientMisconfigured(ctx: any): ApolloError { "Client has no API key configured" ); } + +/** Normalise anything thrown by a service run into the ApolloError envelope, so + * a caller sees the same shape whether the failure was typed or not. */ +export function toErrorPayload(error: unknown): ApolloError { + if (error instanceof ApolloThrowable) { + return error.toJSON(); + } + // Rebuilt field by field rather than returned as-is: isApolloError only + // checks for a numeric `code`, and anything else hanging off the object + // would be serialised to the caller along with it. + if (isApolloError(error)) { + return { + code: error.code, + type: error.type, + message: error.message, + ...(error.details === undefined ? {} : { details: error.details }), + }; + } + return { + code: 500, + type: "INTERNAL_ERROR", + message: error instanceof Error ? error.message : String(error), + }; +} diff --git a/platform/test/middleware/heartbeat.test.ts b/platform/test/middleware/heartbeat.test.ts index 7577f017..8822c170 100644 --- a/platform/test/middleware/heartbeat.test.ts +++ b/platform/test/middleware/heartbeat.test.ts @@ -1,8 +1,9 @@ -import { describe, expect, it } from "bun:test"; +import { afterEach, describe, expect, it } from "bun:test"; import { HEARTBEAT_FRAME, HEARTBEAT_INTERVAL_MS, + heartbeatIntervalMs, } from "../../src/middleware/services"; describe("SSE heartbeat", () => { @@ -29,3 +30,37 @@ describe("SSE heartbeat", () => { expect(HEARTBEAT_INTERVAL_MS).toBeGreaterThanOrEqual(5_000); }); }); + +describe("heartbeat interval override", () => { + const previous = process.env.APOLLO_HEARTBEAT_INTERVAL_MS; + + afterEach(() => { + if (previous === undefined) { + delete process.env.APOLLO_HEARTBEAT_INTERVAL_MS; + } else { + process.env.APOLLO_HEARTBEAT_INTERVAL_MS = previous; + } + }); + + const resolves = (value: string) => { + process.env.APOLLO_HEARTBEAT_INTERVAL_MS = value; + return heartbeatIntervalMs(); + }; + + it("takes a sensible override", () => { + expect(resolves("5000")).toBe(5000); + }); + + // setInterval treats a delay past 2^31-1 as "every tick", so the value + // someone picks to mean "effectively never" is the one that would flood + // every open stream. + it("falls back rather than overflowing setInterval", () => { + expect(resolves("3000000000")).toBe(HEARTBEAT_INTERVAL_MS); + }); + + it("falls back on zero, negatives and nonsense", () => { + expect(resolves("0")).toBe(HEARTBEAT_INTERVAL_MS); + expect(resolves("-1")).toBe(HEARTBEAT_INTERVAL_MS); + expect(resolves("soon")).toBe(HEARTBEAT_INTERVAL_MS); + }); +}); diff --git a/services/entry.py b/services/entry.py index 6225a85b..60689581 100644 --- a/services/entry.py +++ b/services/entry.py @@ -74,15 +74,21 @@ def call( with open(input_path, "r") as f: data = json.load(f) except FileNotFoundError as e: + # The path is the server's own, so it is for the log, not the + # caller. sentry_sdk.capture_exception(e) return _finish( - ApolloError(code=500, message=f"Input file not found: {input_path}", type="INTERNAL_ERROR").to_dict(), + ApolloError( + code=500, message="Input file not found", type="INTERNAL_ERROR" + ).to_dict(), output_path, ) except json.JSONDecodeError as e: sentry_sdk.capture_exception(e) return _finish( - ApolloError(code=500, message="Invalid JSON input", type="INTERNAL_ERROR").to_dict(), + ApolloError( + code=500, message="Invalid JSON input", type="INTERNAL_ERROR" + ).to_dict(), output_path, ) From faef7dfc8c80eb19f0d3268a662fda827bf7cb1e Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 19:45:45 +0000 Subject: [PATCH 07/23] Give websockets the same patience, and stop hiding a real warning serve.idleTimeout does not reach websockets - Bun keeps a separate timer for them, defaulting to two minutes. So the route most likely to be waiting on a slow answer was still being dropped, and the heartbeat bought it nothing. The comment next to it claimed app.listen sets no reusePort. Elysia's Bun adapter sets it unconditionally, so the guard that warns about a per-process internal token meeting a shared port was being silenced on a false premise. It is accurate now, and goes quiet once APOLLO_INTERNAL_TOKEN is set. --- platform/src/server.ts | 14 +++++++++++--- platform/test/server.test.ts | 9 +++++++++ 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/platform/src/server.ts b/platform/src/server.ts index 297408d1..a67451da 100644 --- a/platform/src/server.ts +++ b/platform/src/server.ts @@ -26,6 +26,13 @@ export default async ( serve: { idleTimeout: 255, }, + // Websockets have their own idle timer, which serve.idleTimeout does not + // reach - it defaults to 120s, so a WS caller waiting on a slow answer + // would be dropped well before the SSE route's heartbeat had earned it + // anything. + websocket: { + idleTimeout: 255, + }, }); app.use(html()); @@ -58,9 +65,10 @@ export default async ( } } - // app.listen below sets no reusePort, so the multi-process internal-token warn - // is dormant; pass the flag here if clustering is ever enabled. - logInternalTokenProvenance(false); + // Elysia's Bun adapter sets reusePort unconditionally, so the guard that + // warns about a per-process token meeting a shared port is live, not + // hypothetical. It stays quiet once APOLLO_INTERNAL_TOKEN is set. + logInternalTokenProvenance(true); await auth.init(); // No stop path exists otherwise; close the DB pool so a graceful pod termination diff --git a/platform/test/server.test.ts b/platform/test/server.test.ts index 6669e014..b395dfce 100644 --- a/platform/test/server.test.ts +++ b/platform/test/server.test.ts @@ -1056,3 +1056,12 @@ describe("Instance auth key encryption", () => { } }); }); + +describe("Websocket timeouts", () => { + // serve.idleTimeout does not reach websockets: Bun keeps a separate timer + // for them, defaulting to 120s. Without this a WS caller waiting on a slow + // answer is dropped long before the heartbeat has bought anything. + it("gives websockets the same patience as SSE streams", () => { + expect(app.config.websocket?.idleTimeout).toBe(255); + }); +}); From 8e25fe68d7fbaebf71e7ad59da81484fad9b41dd Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 19:53:06 +0000 Subject: [PATCH 08/23] Tighten and sweep the temp payload files A run's input file holds whatever the server put on the payload, and the only thing that removes it is the bridge's close handler - so anything that stopped the process mid-run left one behind, readable by anyone on the box, in a directory nothing sweeps. The shutdown handler says as much: in-flight children are not drained. It is 0600 now, and startup clears whatever a previous process left. This worktree had four sitting in it from earlier test runs. --- platform/src/bridge.ts | 8 +++++++- platform/src/server.ts | 29 +++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index 8e8c3c53..05c95b72 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -1,7 +1,7 @@ import readline from "node:readline"; import path from "node:path"; import { spawn } from "node:child_process"; -import { rm } from "node:fs/promises"; +import { chmod, rm } from "node:fs/promises"; import { getInternalToken } from "./auth/internal-token"; import { emptyResult, @@ -45,6 +45,12 @@ export const run = async ( // the caller's stream open. Out here, run() is async and simply rejects. try { await Bun.write(inputPath, JSON.stringify(args)); + + // This file carries the provider key the server swapped in, and the only + // thing that removes it is the close handler - so a process that dies + // first leaves it behind. Readable by its owner alone, at least. + await chmod(inputPath, 0o600); + await Bun.write(outputPath, ""); } catch (error) { // The input file holds the key, so it does not get left behind on a diff --git a/platform/src/server.ts b/platform/src/server.ts index a67451da..1948ffde 100644 --- a/platform/src/server.ts +++ b/platform/src/server.ts @@ -11,6 +11,34 @@ import { captureException } from "./util/sentry"; import { clientsDbUrl, closeDb } from "./db"; import { runMigrations } from "./db/migrate"; import { randomUUID } from "node:crypto"; +import { readdir, rm } from "node:fs/promises"; +import path from "node:path"; + +// A run's input file carries the provider key the server swapped in, and only +// the bridge's close handler removes it - so anything that stopped the process +// mid-run left one behind. Nothing else ever sweeps them, so a pod that +// restarts often accumulates keys on disk indefinitely. Startup is the one +// moment we know no run of ours is using them. +const sweepTempPayloads = async () => { + const dir = path.resolve("tmp/data"); + + try { + const stale = await readdir(dir); + + await Promise.all( + stale.map((name) => rm(path.join(dir, name), { force: true })) + ); + + if (stale.length) { + console.log(`Removed ${stale.length} temp payload(s) left by a previous run`); + } + } catch (error) { + // No directory yet on a first boot, which is not worth reporting. + if ((error as { code?: string }).code !== "ENOENT") { + console.error("Could not sweep tmp/data", error); + } + } +}; import pkg from "../../package.json"; export default async ( @@ -69,6 +97,7 @@ export default async ( // warns about a per-process token meeting a shared port is live, not // hypothetical. It stays quiet once APOLLO_INTERNAL_TOKEN is set. logInternalTokenProvenance(true); + await sweepTempPayloads(); await auth.init(); // No stop path exists otherwise; close the DB pool so a graceful pod termination From f36a48007317da9069da126224fb821de12b50b1 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 20:44:48 +0000 Subject: [PATCH 09/23] Say what the temp file holds without naming it The commit messages and the issue were written carefully and the code comments were not, which is the same mistake one layer down - this repo is public and a comment is as readable as anything else. --- platform/src/bridge.ts | 10 +++++----- platform/src/server.ts | 10 +++++----- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index 05c95b72..aec67138 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -46,15 +46,15 @@ export const run = async ( try { await Bun.write(inputPath, JSON.stringify(args)); - // This file carries the provider key the server swapped in, and the only - // thing that removes it is the close handler - so a process that dies - // first leaves it behind. Readable by its owner alone, at least. + // The payload can hold values that belong to the deployment rather than + // the caller, and only the close handler removes this file - so a process + // that dies first leaves one behind. await chmod(inputPath, 0o600); await Bun.write(outputPath, ""); } catch (error) { - // The input file holds the key, so it does not get left behind on a - // half-finished setup. + // Removed rather than left behind by a half-finished setup, for the same + // reason it is 0600 above. await rm(inputPath).catch(() => {}); await rm(outputPath).catch(() => {}); throw subprocessSpawnFailed(scriptName, error); diff --git a/platform/src/server.ts b/platform/src/server.ts index 1948ffde..0fbe6f4a 100644 --- a/platform/src/server.ts +++ b/platform/src/server.ts @@ -14,11 +14,11 @@ import { randomUUID } from "node:crypto"; import { readdir, rm } from "node:fs/promises"; import path from "node:path"; -// A run's input file carries the provider key the server swapped in, and only -// the bridge's close handler removes it - so anything that stopped the process -// mid-run left one behind. Nothing else ever sweeps them, so a pod that -// restarts often accumulates keys on disk indefinitely. Startup is the one -// moment we know no run of ours is using them. +// A run's input file can hold values that belong to the deployment rather +// than the caller, and only the bridge's close handler removes it - so +// anything that stopped the process mid-run left one behind, and nothing else +// ever sweeps them. Startup is the one moment we know no run of ours is +// reading them. const sweepTempPayloads = async () => { const dir = path.resolve("tmp/data"); From f19706b433b8af44211cfbf312068fed6f575983 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 03:55:31 +0000 Subject: [PATCH 10/23] Stop echo reflecting the whole payload echo returned its input verbatim and logged it. A payload reaching a service can carry values the server set rather than the caller, and logger output is forwarded to the client as SSE log events, so anything in there leaves twice. It now runs the payload through mask_secrets before returning or logging it. That helper already existed for Langfuse traces and does the same job here, so there is no second one to keep in step. The mask itself was narrower than what the server can fill in: it listed three field names and recognised one provider's key format. It now covers the fields for the other providers too, and matches both key shapes, so a value under a name nobody listed is still caught. Follow-ups in #634. --- .changeset/echo-mask-payload.md | 7 ++++++ services/echo/echo.py | 10 ++++++-- services/echo/tests/test_echo.py | 39 ++++++++++++++++++++++++++++++++ services/langfuse_util.py | 20 ++++++++++++++-- 4 files changed, 72 insertions(+), 4 deletions(-) create mode 100644 .changeset/echo-mask-payload.md create mode 100644 services/echo/tests/test_echo.py diff --git a/.changeset/echo-mask-payload.md b/.changeset/echo-mask-payload.md new file mode 100644 index 00000000..177e7984 --- /dev/null +++ b/.changeset/echo-mask-payload.md @@ -0,0 +1,7 @@ +--- +"apollo": patch +--- + +echo: mask sensitive values rather than reflecting the whole payload back to +the caller and into the logs. The shared mask now covers every field the +server may fill in, and recognises both provider key formats by shape diff --git a/services/echo/echo.py b/services/echo/echo.py index 90ecaf32..eaba29dc 100644 --- a/services/echo/echo.py +++ b/services/echo/echo.py @@ -1,11 +1,17 @@ +from langfuse_util import mask_secrets from util import ApolloError + from .log import log + # Sample python service to echo requests back to the caller def main(x): # raise a 400 if the payload is empty (ignoring the session id which is system-set) ## useful for diagnosing errors if not x or set(x.keys()) == {"session_id"}: raise ApolloError(code=400, message="payload is required", type="BAD_REQUEST") - log(x) - return x + # Not the raw payload: the server sets fields on it that belong to the + # deployment rather than to the caller. + safe = mask_secrets(x) + log(safe) + return safe diff --git a/services/echo/tests/test_echo.py b/services/echo/tests/test_echo.py new file mode 100644 index 00000000..7cb276bf --- /dev/null +++ b/services/echo/tests/test_echo.py @@ -0,0 +1,39 @@ +import pytest +from echo.echo import main +from util import ApolloError + + +def test_echoes_the_payload_back() -> None: + assert main({"message": "hello"}) == {"message": "hello"} + + +def test_rejects_an_empty_payload() -> None: + with pytest.raises(ApolloError): + main({}) + + with pytest.raises(ApolloError): + main({"session_id": "abc"}) + + +def test_does_not_return_server_set_fields() -> None: + # The server sets these itself, so their values must not come back out. + echoed = main({"message": "hello", "api_key": "test-value"}) + + assert "test-value" not in str(echoed) + assert echoed["message"] == "hello" + + +def test_masks_a_key_shaped_value_anywhere_in_the_payload() -> None: + # Caught by shape, so a value under a field name nobody listed is still + # masked. Both provider prefixes, since the server may fill in either. + assert "sk-ant-abc123" not in str(main({"note": "uses sk-ant-abc123"})) + assert "sk-proj-0123456789abcdef" not in str( + main({"note": "uses sk-proj-0123456789abcdef"}), + ) + + +def test_masks_a_provider_field_the_server_may_fill_in() -> None: + echoed = main({"message": "hello", "openai_api_key": "test-openai"}) + + assert "test-openai" not in str(echoed) + assert echoed["message"] == "hello" diff --git a/services/langfuse_util.py b/services/langfuse_util.py index 31913882..aecebdf2 100644 --- a/services/langfuse_util.py +++ b/services/langfuse_util.py @@ -5,8 +5,24 @@ import yaml -_SECRET_KEY_NAMES = {"api_key", "anthropic_api_key", "authorization"} -_SECRET_VALUE_PATTERN = re.compile(r"sk-ant-[\w\-]+") +# Every field the server may fill in on a payload, not just the one it fills +# in today: a value here belongs to the deployment rather than to the caller. +_SECRET_KEY_NAMES = { + "api_key", + "anthropic_api_key", + "openai_api_key", + "pinecone_api_key", + "langfuse_secret_key", + "langfuse_public_key", + "authorization", + "x-api-key", +} + +# Catches a key by its shape, so one under a field name we did not list is +# still masked. Anthropic's prefix is distinctive enough on its own; the +# second branch covers the other sk- forms, with a length floor so it stays +# off ordinary prose. +_SECRET_VALUE_PATTERN = re.compile(r"sk-(?:ant-[\w\-]+|[A-Za-z0-9_\-]{16,})") def mask_secrets(data: Any) -> Any: # noqa: ANN401 From 68c0998b30a733f2e945b327dc937590a21708ee Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 10:24:36 +0000 Subject: [PATCH 11/23] Mask sensitive values on their way out of a service A payload reaching a service carries values the server put there rather than the caller. Three routes led back out: echo returned its input verbatim, any service logging its payload reached the caller too (logger output is forwarded as SSE log events), and the error envelope returns the exception text. The logger is the one that matters, because it needs nothing of the service: vocab_mapper logs its whole payload on the first line of main. Masking in create_logger covers that and anything written later without each service having to remember. Also masks the request context search_adaptor_docs sends to Sentry, which the three sibling services already stripped. The mask itself was narrower than what the server can fill in, so it now lists the other providers' fields. Widening its value pattern to a general sk- shape first made it match ordinary words - task-, risk-, disk- and friends - which would have quietly corrupted every Langfuse trace, since this is the same function used as the export mask. It is anchored now, with tests pinning both directions. The instance-auth tests read a masked field off echo's response, which they can no longer do. They assert that the request is accepted and that what the caller sent does not come back; what the server substitutes is covered directly against InstanceAuth.authenticate. --- .changeset/echo-mask-payload.md | 8 ++- CLAUDE.md | 3 +- platform/test/server.test.ts | 23 ++++--- services/echo/README.md | 5 +- services/echo/__init__.py | 0 services/echo/tests/__init__.py | 0 services/echo/tests/unit/__init__.py | 0 services/echo/tests/{ => unit}/test_echo.py | 0 services/entry.py | 10 ++- .../tests/unit/test_mask_secrets.py | 61 +++++++++++++++++++ services/langfuse_util.py | 20 +++--- .../search_adaptor_docs.py | 3 +- services/util.py | 22 +++++++ 13 files changed, 132 insertions(+), 23 deletions(-) create mode 100644 services/echo/__init__.py create mode 100644 services/echo/tests/__init__.py create mode 100644 services/echo/tests/unit/__init__.py rename services/echo/tests/{ => unit}/test_echo.py (100%) create mode 100644 services/global_chat/tests/unit/test_mask_secrets.py diff --git a/.changeset/echo-mask-payload.md b/.changeset/echo-mask-payload.md index 177e7984..29f5e4e2 100644 --- a/.changeset/echo-mask-payload.md +++ b/.changeset/echo-mask-payload.md @@ -2,6 +2,8 @@ "apollo": patch --- -echo: mask sensitive values rather than reflecting the whole payload back to -the caller and into the logs. The shared mask now covers every field the -server may fill in, and recognises both provider key formats by shape +Mask sensitive values on their way out of a service, rather than relying on +each one to remember: service loggers mask what they emit, echo masks what it +returns, and the error envelope masks the exception text. The shared mask now +covers every field the server may fill in, and no longer matches ordinary +hyphenated words diff --git a/CLAUDE.md b/CLAUDE.md index 25b2e406..d6161135 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -178,7 +178,8 @@ SSE to clients. against the `apollo-mappings` Pinecone index (collections `loinc-mappings-v2`, `snomed-mappings`). `embed_loinc_dataset` / `embed_snomed_dataset` populate it. - `status/` - Health check: validates Anthropic, OpenAI and Pinecone keys. -- `echo/` - Test service that returns its input; useful for verifying the server +- `echo/` - Test service that returns its input (with server-set values such as + `api_key` masked); useful for verifying the server pipeline. Note: there are **three distinct Pinecone indexes** — `docsite` (OpenFn docs), diff --git a/platform/test/server.test.ts b/platform/test/server.test.ts index b395dfce..db449950 100644 --- a/platform/test/server.test.ts +++ b/platform/test/server.test.ts @@ -254,13 +254,20 @@ describe("Instance authentication", () => { }); // Row 1 + // + // echo used to hand its payload back verbatim, which is what let these rows + // read the swapped key off the response. It masks now, so what is left to + // observe here is that the request was accepted and the field was set. That + // the *correct* key is selected is covered directly against + // InstanceAuth.authenticate further down, where lightningClient.anthropicKey + // is asserted without a round trip through a service. it("accepts a known credential and swaps in the client's stored key", async () => { const res = await app.handle(postKey("services/echo", { x: 1 }, ALPHA)); expect(res.status).toBe(200); const body = await res.json(); expect(body.x).toBe(1); - expect(body.api_key).toBe("sk-ant-stored-alpha"); - expect(body.api_key).not.toBe(ALPHA); + expect(body.api_key).toBe("[REDACTED]"); + expect(JSON.stringify(body)).not.toContain(ALPHA); }); // Row 2 — a recognised client whose stored key is NULL is a server-side @@ -341,7 +348,7 @@ describe("Instance authentication", () => { postKey("services/echo", { x: 1 }, "sk-ant-unknown") ); expect(swap.status).toBe(200); - expect((await swap.json()).api_key).toBe("sk-ant-stored-alpha"); + expect((await swap.json()).api_key).toBe("[REDACTED]"); expect(unknown.status).toBe(401); }); @@ -483,7 +490,9 @@ describe("Instance authentication", () => { const res = await app.handle(req); expect(res.status).toBe(200); const body = await res.json(); - expect(body.api_key).toBe("sk-ant-internal-hop"); + // Masked on the way out like anything else. What this row pins is that an + // internal call is accepted and its body forwarded rather than rejected. + expect(body.api_key).toBe("[REDACTED]"); }); // WS upgrade auth decision via app.handle(): a bare upgrade carries no api_key, so @@ -544,8 +553,8 @@ describe("Instance authentication", () => { // that ws.data carries the lightningClient set during beforeHandle. it("swaps a known client's stored key on a WS upgrade via ?api_key=", async () => { const body = await wsRoundTrip(`?api_key=${encodeURIComponent(ALPHA)}`); - expect(body.api_key).toBe("sk-ant-stored-alpha"); - expect(body.api_key).not.toBe(ALPHA); + expect(body.api_key).toBe("[REDACTED]"); + expect(JSON.stringify(body)).not.toContain(ALPHA); expect(body.ws).toBe(1); }); @@ -589,7 +598,7 @@ describe("Instance authentication", () => { ); }); }); - expect(body.api_key).toBe("sk-ant-internal-fwd"); + expect(body.api_key).toBe("[REDACTED]"); }); }); diff --git a/services/echo/README.md b/services/echo/README.md index 26efc3d2..95b88547 100644 --- a/services/echo/README.md +++ b/services/echo/README.md @@ -10,4 +10,7 @@ Call the endpoint at `services/echo` curl -X POST localhost:3000/services/echo --json @tmp/data.json ``` -Whatever you include in the body will be be returned straight back. +Whatever you include in the body will be returned straight back, with one +exception: values the server fills in itself, such as `api_key`, come back as +`[REDACTED]`, as does anything else shaped like a key. Those belong to the +deployment rather than to the caller, so echo does not hand them out. diff --git a/services/echo/__init__.py b/services/echo/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/services/echo/tests/__init__.py b/services/echo/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/services/echo/tests/unit/__init__.py b/services/echo/tests/unit/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/services/echo/tests/test_echo.py b/services/echo/tests/unit/test_echo.py similarity index 100% rename from services/echo/tests/test_echo.py rename to services/echo/tests/unit/test_echo.py diff --git a/services/entry.py b/services/entry.py index 60689581..b12dbd81 100644 --- a/services/entry.py +++ b/services/entry.py @@ -95,10 +95,14 @@ def call( try: m = __import__(module_name, fromlist=["main"]) result = m.main(data) + # An exception message goes back to the caller verbatim, and the payload is + # in scope wherever one is raised, so mask before it leaves. except ModuleNotFoundError as e: sentry_sdk.capture_exception(e) return _finish( - ApolloError(code=500, message=str(e), type="INTERNAL_ERROR").to_dict(), + ApolloError( + code=500, message=mask_secrets(str(e)), type="INTERNAL_ERROR" + ).to_dict(), output_path, ) except ApolloError as e: @@ -106,7 +110,9 @@ def call( result = e.to_dict() except Exception as e: sentry_sdk.capture_exception(e) - result = ApolloError(code=500, message=str(e), type="INTERNAL_ERROR").to_dict() + result = ApolloError( + code=500, message=mask_secrets(str(e)), type="INTERNAL_ERROR" + ).to_dict() langfuse.flush() diff --git a/services/global_chat/tests/unit/test_mask_secrets.py b/services/global_chat/tests/unit/test_mask_secrets.py new file mode 100644 index 00000000..0c8eebb9 --- /dev/null +++ b/services/global_chat/tests/unit/test_mask_secrets.py @@ -0,0 +1,61 @@ +"""Boundaries of the shared mask. + +`mask_secrets` is both the Langfuse export mask and what stops a service +handing a key back to its caller, so it has to catch keys without eating the +workflow YAML, job code and prose it also passes over. +""" + +import pytest + +from langfuse_util import mask_secrets + +KEY_SHAPED = [ + "sk-ant-abc123", + "sk-ant-api03-Zm9vYmFy_baz", + "sk-proj-0123456789abcdef", + "use sk-ant-abc123 to authenticate", + "(sk-ant-abc123)", +] + +# Every one of these ends in a token the pattern used to bite through: task-, +# risk-, disk-, mask-, kiosk-. +ORDINARY_TEXT = [ + "task-scheduler-configuration", + "risk-assessment-framework", + "disk-usage-monitoring-job", + "mask-generation-pipeline", + "kiosk-registration-workflow-id", + "https://docs.openfn.org/build/task-automation-guide", + "const task = 'task-runner-config-value';", +] + + +@pytest.mark.parametrize("text", KEY_SHAPED) +def test_masks_a_key_shaped_value(text: str) -> None: + assert "[REDACTED]" in mask_secrets(text) + + +@pytest.mark.parametrize("text", ORDINARY_TEXT) +def test_leaves_ordinary_text_alone(text: str) -> None: + assert mask_secrets(text) == text + + +def test_masks_by_field_name_whatever_the_value_looks_like() -> None: + masked = mask_secrets({"api_key": "not-key-shaped", "message": "hello"}) + + assert masked["api_key"] == "[REDACTED]" + assert masked["message"] == "hello" + + +def test_keeps_the_langfuse_public_key() -> None: + # Public by design, and useful in a trace for telling which project a span + # was bound for. + masked = mask_secrets({"langfuse_public_key": "pk-lf-123"}) + + assert masked["langfuse_public_key"] == "pk-lf-123" + + +def test_masks_nested_and_listed_values() -> None: + masked = mask_secrets({"outer": [{"api_key": "sk-ant-abc123"}]}) + + assert masked["outer"][0]["api_key"] == "[REDACTED]" diff --git a/services/langfuse_util.py b/services/langfuse_util.py index aecebdf2..d6d29427 100644 --- a/services/langfuse_util.py +++ b/services/langfuse_util.py @@ -5,24 +5,28 @@ import yaml -# Every field the server may fill in on a payload, not just the one it fills -# in today: a value here belongs to the deployment rather than to the caller. +# The control. Every field the server may fill in on a payload, not just the +# one it fills in today: a value here belongs to the deployment rather than to +# the caller. langfuse_public_key is deliberately absent - it is meant to be +# visible, and masking it would cost a useful identifier for nothing. _SECRET_KEY_NAMES = { "api_key", "anthropic_api_key", "openai_api_key", "pinecone_api_key", "langfuse_secret_key", - "langfuse_public_key", "authorization", "x-api-key", } -# Catches a key by its shape, so one under a field name we did not list is -# still masked. Anthropic's prefix is distinctive enough on its own; the -# second branch covers the other sk- forms, with a length floor so it stays -# off ordinary prose. -_SECRET_VALUE_PATTERN = re.compile(r"sk-(?:ant-[\w\-]+|[A-Za-z0-9_\-]{16,})") +# A backstop for a key that turns up somewhere the name list cannot see, such +# as inside a string. The lookbehind is load-bearing: without it the sk- also +# matches the tail of ta[sk-], ri[sk-], di[sk-] and friends, which are ordinary +# words in the workflow YAML and job code this same function masks on the way +# to Langfuse. +_SECRET_VALUE_PATTERN = re.compile( + r"(? Any: # noqa: ANN401 diff --git a/services/search_adaptor_docs/search_adaptor_docs.py b/services/search_adaptor_docs/search_adaptor_docs.py index 31ede7b9..bf13ee6b 100644 --- a/services/search_adaptor_docs/search_adaptor_docs.py +++ b/services/search_adaptor_docs/search_adaptor_docs.py @@ -1,6 +1,7 @@ import time from typing import Dict, List, Any import sentry_sdk +from langfuse_util import mask_secrets from util import create_logger, ApolloError, AdaptorSpecifier, get_db_connection from load_adaptor_docs.load_adaptor_docs import load_adaptor_docs @@ -211,7 +212,7 @@ def main(data: dict) -> dict: """ logger.info("Starting search_adaptor_docs...") - sentry_sdk.set_context("request_data", data) + sentry_sdk.set_context("request_data", mask_secrets(data)) # Validate required fields if "adaptor" not in data: diff --git a/services/util.py b/services/util.py index 45d6d0f7..94bebc55 100644 --- a/services/util.py +++ b/services/util.py @@ -6,6 +6,7 @@ import psycopg2 import requests +from langfuse_util import mask_secrets APOLLO_VERSION = os.getenv("APOLLO_VERSION", "unknown") @@ -72,6 +73,26 @@ def set_log_output(f: str | None) -> None: filename = f +class _MaskingFilter(logging.Filter): + """Masks key-shaped values on their way out of a service logger. + + This output does not stay on the server: the bridge matches the log prefix + on stdout and forwards the line to the caller as an SSE event. A service + that logs its own payload would therefore hand the caller the key the + server put there. Doing it here means no service has to remember. + """ + + def filter(self, record: logging.LogRecord) -> bool: + # Resolve %-args first, so the mask sees the text that will be emitted + # rather than a format string. + if record.args: + record.msg = record.getMessage() + record.args = () + + record.msg = mask_secrets(record.msg) + return True + + def create_logger(name: str) -> logging.Logger: """ Create or retrieve a logger with the given name. @@ -80,6 +101,7 @@ def create_logger(name: str) -> logging.Logger: logging.basicConfig(level=logging.INFO, stream=sys.stdout) if name not in loggers: logger = logging.getLogger(name) + logger.addFilter(_MaskingFilter()) loggers[name] = logger return loggers[name] From 5b21ba8c2daf542f2ee61395867362eadf9f64ef Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 17:19:56 +0000 Subject: [PATCH 12/23] Close the routes the first mask left open Review of the masking itself found four ways round it. The error envelope masked two branches and not the one services take. Nearly all of them catch broadly and rewrap as ApolloError(500, str(e)), which entry.py returned untouched - the same disclosure, by the sibling branch of the same case statement. Masking now happens once at the exit, so it covers the message, the details, and whatever gets added later. The log filter was attached to each logger built by create_logger, and a filter on a logger only runs for records emitted through it. A plain getLogger, or a third-party logger like httpx, writes to the same stdout handler and sailed past - and vocab_mapper already silences httpx precisely because it reaches that stream. It sits on the handler now, so what a service uses to log makes no difference. Rendering the message inside the filter moved %-formatting out of the handler, where a bad format string is caught and reported, and into the caller's own logging call, where it is not. A cosmetic typo in a log line would have failed the request. And a message that was neither string nor container passed through unmasked, because the formatter calls str() on it after the filter has run. Sentry saw everything unmasked: it scrubs frame locals by name, but not exception text and not a set_context payload. A before_send hook covers both, and replaces the per-service discipline three services each hand-rolled differently. Also: the lookbehind excluded a leading hyphen or underscore, which are legitimately in front of a key; api-key and apiKey were not recognised alongside api_key, and x-api-key stopped being recognised when the names were normalised; the recursion had no depth bound, which a deep payload could turn into a failed request; and two except clauses captured an exception variable they never bound, raising NameError instead of the 500 they meant to return. --- services/entry.py | 60 ++++++++++++------- .../tests/unit/test_mask_secrets.py | 25 +++++++- services/langfuse_util.py | 39 +++++++++--- .../search_adaptor_docs.py | 22 +++---- services/util.py | 48 +++++++++++---- 5 files changed, 142 insertions(+), 52 deletions(-) diff --git a/services/entry.py b/services/entry.py index b12dbd81..5ab84d1d 100644 --- a/services/entry.py +++ b/services/entry.py @@ -1,17 +1,18 @@ -import sys -import os +import argparse import json +import os import uuid -import argparse -from dotenv import load_dotenv + import sentry_sdk -from util import set_apollo_port, ApolloError +from dotenv import load_dotenv +from util import ApolloError, set_apollo_port load_dotenv() # Langfuse: init after load_dotenv so env vars are available, before any Anthropic client is created from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor from opentelemetry.instrumentation.threading import ThreadingInstrumentor + AnthropicInstrumentor().instrument() ThreadingInstrumentor().instrument() @@ -42,17 +43,33 @@ def _should_export_span(span): 'unknown': 0.0, } +def _scrub_event(event: dict, _hint: dict) -> dict: + """Mask keys in what Sentry is about to send. + + Sentry scrubs frame locals by name, but not the exception message or a + set_context payload - and services raise ApolloError(500, str(e)) with the + request in scope. One hook here rather than each service remembering to + strip its own context before calling set_context. + """ + for section in ("exception", "contexts", "extra", "logentry"): + if section in event: + event[section] = mask_secrets(event[section]) + + return event + + sentry_sdk.init( dsn=os.getenv('SENTRY_DSN'), environment=env, sample_rate=1.0, traces_sample_rate=trace_rates.get(env, 0.0), enable_tracing=True, - auto_enabling_integrations=False + auto_enabling_integrations=False, + before_send=_scrub_event, ) def call( - service: str, *, input_path: str | None = None, output_path: str | None = None, apollo_port: int | None = None + service: str, *, input_path: str | None = None, output_path: str | None = None, apollo_port: int | None = None, ) -> dict: """ Dynamically imports a module and invokes its main function with input data. @@ -71,7 +88,7 @@ def call( data = {} if input_path: try: - with open(input_path, "r") as f: + with open(input_path) as f: data = json.load(f) except FileNotFoundError as e: # The path is the server's own, so it is for the log, not the @@ -95,23 +112,18 @@ def call( try: m = __import__(module_name, fromlist=["main"]) result = m.main(data) - # An exception message goes back to the caller verbatim, and the payload is - # in scope wherever one is raised, so mask before it leaves. except ModuleNotFoundError as e: sentry_sdk.capture_exception(e) - return _finish( - ApolloError( - code=500, message=mask_secrets(str(e)), type="INTERNAL_ERROR" - ).to_dict(), - output_path, - ) + result = ApolloError( + code=500, message=str(e), type="INTERNAL_ERROR", + ).to_dict() except ApolloError as e: sentry_sdk.capture_exception(e) result = e.to_dict() except Exception as e: sentry_sdk.capture_exception(e) result = ApolloError( - code=500, message=mask_secrets(str(e)), type="INTERNAL_ERROR" + code=500, message=str(e), type="INTERNAL_ERROR", ).to_dict() langfuse.flush() @@ -120,11 +132,17 @@ def call( def _finish(result: dict, output_path: str | None) -> dict: - """Write the result where the caller expects it, then hand it back. - - Every path out of run_service goes through here, so an empty output file - means the run died rather than that it failed politely. + """Mask, write the result where the caller expects it, then hand it back. + + Every path out of `call` already came through here so that an empty output + file means the run died rather than that it failed politely. Masking here + too means a value the server put in the payload cannot leave down a branch + someone forgot: most services catch broadly and rewrap as + `ApolloError(500, str(e))`, so masking per-branch would miss the one nearly + all of them take. """ + result = mask_secrets(result) + if output_path: with open(output_path, "w") as f: json.dump(result, f) diff --git a/services/global_chat/tests/unit/test_mask_secrets.py b/services/global_chat/tests/unit/test_mask_secrets.py index 0c8eebb9..38425b81 100644 --- a/services/global_chat/tests/unit/test_mask_secrets.py +++ b/services/global_chat/tests/unit/test_mask_secrets.py @@ -6,7 +6,6 @@ """ import pytest - from langfuse_util import mask_secrets KEY_SHAPED = [ @@ -15,6 +14,10 @@ "sk-proj-0123456789abcdef", "use sk-ant-abc123 to authenticate", "(sk-ant-abc123)", + # A key can legitimately follow a hyphen or underscore, so only an + # alphanumeric before the sk- means it is the tail of a word. + "-sk-ant-zzzzzzzzzzzzzzzzzz", + "_sk-ant-zzzzzzzzzzzzzzzzzz", ] # Every one of these ends in a token the pattern used to bite through: task-, @@ -27,9 +30,29 @@ "kiosk-registration-workflow-id", "https://docs.openfn.org/build/task-automation-guide", "const task = 'task-runner-config-value';", + "obelisk-carving-notes", + "whisk-attachment-guide", ] +@pytest.mark.parametrize( + "name", + ["api_key", "api-key", "apiKey", "X-Api-Key", "Authorization"], +) +def test_masks_a_secret_field_however_it_is_spelled(name: str) -> None: + assert mask_secrets({name: "anything"})[name] == "[REDACTED]" + + +def test_does_not_recurse_without_bound() -> None: + # json.loads accepts around a thousand levels, and this runs over whatever + # a caller sends, so a deep payload must not take the process down. + nested = {"leaf": "value"} + for _ in range(600): + nested = {"next": nested} + + assert isinstance(mask_secrets(nested), dict) + + @pytest.mark.parametrize("text", KEY_SHAPED) def test_masks_a_key_shaped_value(text: str) -> None: assert "[REDACTED]" in mask_secrets(text) diff --git a/services/langfuse_util.py b/services/langfuse_util.py index d6d29427..5a3fc837 100644 --- a/services/langfuse_util.py +++ b/services/langfuse_util.py @@ -16,33 +16,58 @@ "pinecone_api_key", "langfuse_secret_key", "authorization", - "x-api-key", + "x_api_key", } +# Depth beyond which a payload is not being logged in good faith. Guards +# against a self-inflicted RecursionError: json.loads accepts around a +# thousand levels, and this runs over anything a caller sends. +_MAX_DEPTH = 50 + # A backstop for a key that turns up somewhere the name list cannot see, such # as inside a string. The lookbehind is load-bearing: without it the sk- also # matches the tail of ta[sk-], ri[sk-], di[sk-] and friends, which are ordinary # words in the workflow YAML and job code this same function masks on the way -# to Langfuse. +# to Langfuse. Only alphanumerics need excluding - a key can legitimately +# follow a hyphen or an underscore. _SECRET_VALUE_PATTERN = re.compile( - r"(? Any: # noqa: ANN401 +def _normalise_name(key: object) -> str: + return str(key).lower().replace("-", "").replace("_", "") + + +# Compared in normalised form, so api-key, apiKey and X-Api-Key are all caught +# by the one readable entry above. +_NORMALISED_SECRET_NAMES = {_normalise_name(name) for name in _SECRET_KEY_NAMES} + + +def _is_secret_name(key: object) -> bool: + return _normalise_name(key) in _NORMALISED_SECRET_NAMES + + +def mask_secrets(data: Any, _depth: int = 0) -> Any: # noqa: ANN401 """Langfuse mask callback: redact API keys from all traced data. Applied by the SDK to every span's input, output and metadata before export, so keys passed as function arguments to @observe-decorated - functions never reach Langfuse. + functions never reach Langfuse. Also used anywhere a service's own output + could carry a value the server put in the payload. """ + if _depth > _MAX_DEPTH: + return "[TRUNCATED]" + if isinstance(data, dict): return { - k: "[REDACTED]" if str(k).lower() in _SECRET_KEY_NAMES and v else mask_secrets(v) + k: "[REDACTED]" + if _is_secret_name(k) and v + else mask_secrets(v, _depth + 1) for k, v in data.items() } if isinstance(data, (list, tuple)): - return [mask_secrets(v) for v in data] + return [mask_secrets(v, _depth + 1) for v in data] if isinstance(data, str): return _SECRET_VALUE_PATTERN.sub("[REDACTED]", data) return data diff --git a/services/search_adaptor_docs/search_adaptor_docs.py b/services/search_adaptor_docs/search_adaptor_docs.py index bf13ee6b..d089b1bd 100644 --- a/services/search_adaptor_docs/search_adaptor_docs.py +++ b/services/search_adaptor_docs/search_adaptor_docs.py @@ -1,9 +1,9 @@ import time -from typing import Dict, List, Any + import sentry_sdk from langfuse_util import mask_secrets -from util import create_logger, ApolloError, AdaptorSpecifier, get_db_connection from load_adaptor_docs.load_adaptor_docs import load_adaptor_docs +from util import AdaptorSpecifier, ApolloError, create_logger, get_db_connection logger = create_logger("search_adaptor_docs") @@ -23,7 +23,7 @@ def ensure_docs_loaded(adaptor: AdaptorSpecifier, conn, skip_if_exists: bool = T load_result = load_adaptor_docs( adaptor=adaptor.specifier, skip_if_exists=skip_if_exists, - conn=conn + conn=conn, ) duration = time.time() - start_time @@ -35,7 +35,7 @@ def ensure_docs_loaded(adaptor: AdaptorSpecifier, conn, skip_if_exists: bool = T logger.warning(f"Failed to load adaptor docs for {adaptor.specifier} after {duration:.3f}s") except Exception as e: duration = time.time() - start_time if 'start_time' in locals() else 0 - logger.warning(f"Failed to load adaptor docs after {duration:.3f}s: {str(e)}") + logger.warning(f"Failed to load adaptor docs after {duration:.3f}s: {e!s}") sentry_sdk.capture_exception(e) @@ -190,7 +190,7 @@ def fetch_all_functions(adaptor: AdaptorSpecifier, conn, format: str = "json", a return [ { "function_name": row[0], - "text": json_to_natural_language(row[1], adaptor) + "text": json_to_natural_language(row[1], adaptor), } for row in rows ] @@ -255,7 +255,7 @@ def main(data: dict) -> dict: "adaptor": adaptor.name, "version": adaptor.version, "query_type": "list", - "functions": functions + "functions": functions, } elif query_type == "signatures": @@ -266,7 +266,7 @@ def main(data: dict) -> dict: "adaptor": adaptor.name, "version": adaptor.version, "query_type": "signatures", - "signatures": signatures + "signatures": signatures, } elif query_type == "function": @@ -282,7 +282,7 @@ def main(data: dict) -> dict: "query_type": "function", "function_name": function_name, "format": format, - "data": func_data + "data": func_data, } elif query_type == "all": @@ -295,14 +295,14 @@ def main(data: dict) -> dict: "query_type": "all", "format": format, "count": len(functions), - "functions": functions + "functions": functions, } except ApolloError: raise except Exception as e: - logger.error(f"Error querying database: {str(e)}") - raise ApolloError(500, f"Query failed: {str(e)}", type="DATABASE_ERROR") + logger.error(f"Error querying database: {e!s}") + raise ApolloError(500, f"Query failed: {e!s}", type="DATABASE_ERROR") finally: conn.close() diff --git a/services/util.py b/services/util.py index 94bebc55..db773814 100644 --- a/services/util.py +++ b/services/util.py @@ -74,35 +74,59 @@ def set_log_output(f: str | None) -> None: class _MaskingFilter(logging.Filter): - """Masks key-shaped values on their way out of a service logger. + """Masks key-shaped values on their way out to the log stream. This output does not stay on the server: the bridge matches the log prefix on stdout and forwards the line to the caller as an SSE event. A service that logs its own payload would therefore hand the caller the key the - server put there. Doing it here means no service has to remember. + server put there. + + Attached to the stdout handler rather than to each logger, because a + filter on a logger only runs for records emitted through it - a plain + `logging.getLogger(__name__)`, or a third-party logger like httpx, writes + to the same handler and would sail past. + + Note the two levels are not equally strong. A dict is masked by field name + and by value shape; anything already rendered to text, including a payload + interpolated into an f-string, has only the shape to go on. """ def filter(self, record: logging.LogRecord) -> bool: - # Resolve %-args first, so the mask sees the text that will be emitted - # rather than a format string. - if record.args: - record.msg = record.getMessage() - record.args = () - - record.msg = mask_secrets(record.msg) + # A container is worth masking structurally, so field names apply. + if isinstance(record.msg, (dict, list, tuple)) and not record.args: + record.msg = mask_secrets(record.msg) + return True + + # Otherwise mask the text that will actually be emitted. Rendering + # normally happens inside the handler, where a bad format string is + # caught and reported; here it would escape into the caller's own + # logging call, so a cosmetic typo must not fail their request. + try: + rendered = record.getMessage() + except Exception: + return True + + record.msg = mask_secrets(rendered) + record.args = () return True +_masking_filter = _MaskingFilter() + + def create_logger(name: str) -> logging.Logger: """ Create or retrieve a logger with the given name. Logs to stdout by default. """ logging.basicConfig(level=logging.INFO, stream=sys.stdout) + + for handler in logging.getLogger().handlers: + if _masking_filter not in handler.filters: + handler.addFilter(_masking_filter) + if name not in loggers: - logger = logging.getLogger(name) - logger.addFilter(_MaskingFilter()) - loggers[name] = logger + loggers[name] = logging.getLogger(name) return loggers[name] From 281c4cadae033c66d12d0657e6ace86bb66eda2e Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 17:23:58 +0000 Subject: [PATCH 13/23] Test that the exit mask holds, not just that echo behaves The follow-up from #634, done at the boundary rather than per service. Driven through an unmounted probe that reflects its payload and masks nothing itself, plus one that raises with the payload in scope, since pointing these at echo would only prove echo masks. All three fail if the mask at entry.call's exit is removed. --- services/_masking_probe/__init__.py | 0 services/_masking_probe/_masking_probe.py | 24 ++++++++ services/_masking_probe_raiser/__init__.py | 0 .../_masking_probe_raiser.py | 7 +++ .../tests/unit/test_entry_masks_results.py | 58 +++++++++++++++++++ 5 files changed, 89 insertions(+) create mode 100644 services/_masking_probe/__init__.py create mode 100644 services/_masking_probe/_masking_probe.py create mode 100644 services/_masking_probe_raiser/__init__.py create mode 100644 services/_masking_probe_raiser/_masking_probe_raiser.py create mode 100644 services/echo/tests/unit/test_entry_masks_results.py diff --git a/services/_masking_probe/__init__.py b/services/_masking_probe/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/services/_masking_probe/_masking_probe.py b/services/_masking_probe/_masking_probe.py new file mode 100644 index 00000000..1cb17fb1 --- /dev/null +++ b/services/_masking_probe/_masking_probe.py @@ -0,0 +1,24 @@ +"""Services that do the wrong thing on purpose, so the exit mask can be tested. + +The leading underscore keeps the directory out of describe-modules, so none of +this is mounted and none of it has a route. +""" + +from util import ApolloError + + +def main(data_dict: dict) -> dict: + """Reflects the payload without masking anything itself. + + Stands in for a service written by someone who did not think about it, + which is the case the exit mask exists for. + """ + return data_dict + + +def raise_with_payload(data_dict: dict) -> dict: + """The shape nearly every service uses: catch broadly, rewrap the text.""" + try: + raise ValueError(f"upstream rejected {data_dict.get('api_key')}") + except ValueError as e: + raise ApolloError(500, str(e), type="INTERNAL_ERROR") from e diff --git a/services/_masking_probe_raiser/__init__.py b/services/_masking_probe_raiser/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/services/_masking_probe_raiser/_masking_probe_raiser.py b/services/_masking_probe_raiser/_masking_probe_raiser.py new file mode 100644 index 00000000..8a54019d --- /dev/null +++ b/services/_masking_probe_raiser/_masking_probe_raiser.py @@ -0,0 +1,7 @@ +"""Raises with the payload in scope. Unmounted; see _masking_probe.""" + +from _masking_probe._masking_probe import raise_with_payload + + +def main(data_dict: dict) -> dict: + return raise_with_payload(data_dict) diff --git a/services/echo/tests/unit/test_entry_masks_results.py b/services/echo/tests/unit/test_entry_masks_results.py new file mode 100644 index 00000000..c3b82930 --- /dev/null +++ b/services/echo/tests/unit/test_entry_masks_results.py @@ -0,0 +1,58 @@ +"""The boundary every service result passes through. + +Services are handed a payload with values the server put there, and whatever +they return goes back to the caller. Rather than trust each one not to reflect +those values, `entry.call` masks on the way out. + +Driven through `_masking_probe`, which reflects its payload and masks nothing +itself, so these fail if the boundary stops working. Pointing them at `echo` +would prove only that echo masks. +""" + +import json + +import pytest +from entry import call + +HTTP_INTERNAL_ERROR = 500 + + +@pytest.fixture +def input_file(tmp_path: object) -> object: + def write(payload: dict) -> str: + path = tmp_path / "input.json" + path.write_text(json.dumps(payload)) + return str(path) + + return write + + +def test_masks_a_server_set_field_in_the_result(input_file: object) -> None: + result = call( + "_masking_probe", + input_path=input_file({"api_key": "secret", "x": 1}), + ) + + assert result["api_key"] == "[REDACTED]" + assert result["x"] == 1 + + +def test_masks_a_key_shaped_value_anywhere_in_the_result(input_file: object) -> None: + result = call( + "_masking_probe", + input_path=input_file({"note": "configured with sk-ant-abc123"}), + ) + + assert "sk-ant-abc123" not in json.dumps(result) + + +def test_masks_a_key_that_reached_an_error_message(input_file: object) -> None: + # A service that raises with the payload in scope is the common shape: + # most catch broadly and rewrap as ApolloError(500, str(e)). + result = call( + "_masking_probe_raiser", + input_path=input_file({"api_key": "sk-ant-abc123"}), + ) + + assert result["code"] == HTTP_INTERNAL_ERROR + assert "sk-ant-abc123" not in json.dumps(result) From 621ac48d752ad21bfa3f487401c3668336e9bc00 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 18:34:56 +0000 Subject: [PATCH 14/23] Close what the third review found The exit mask is now a function every path calls, including the two that returned early. That also settles a conflict with the cancellation stack, which restructured the same branches the other way: both now write the output file on every path and mask on every path, so the two agree instead of one silently winning. Sentry got an allowlist of sections and breadcrumbs were not on it. Every INFO record becomes a breadcrumb, and they ride along on the next error event - so a key logged through a handler the filter had not reached left that way. Masking the whole event covers the sections nobody listed, and transactions get the same hook, since before_send is for errors only. The log filter went on at the first create_logger call, over the handlers that existed by then. Langfuse attaches one to the httpx logger during import, before any service module runs, and it writes to stderr, which the bridge forwards to the caller line for line. The sweep now runs at import and again once the service module is loaded, and covers handlers on other loggers rather than only root. The value pattern matched sk- followed by any hyphenated words, so sk-antelope-migration-plan came back redacted. Now that the same function masks what a service returns, that silently corrupts a caller's own data. It anchors on the prefixes providers use, or wants an unbroken run long enough not to be a name. Also drops the server's absolute path from the input-not-found message. --- services/entry.py | 24 ++++++++------ .../tests/unit/test_mask_secrets.py | 9 +++++ services/langfuse_util.py | 20 +++++++---- services/util.py | 33 ++++++++++++++++--- 4 files changed, 66 insertions(+), 20 deletions(-) diff --git a/services/entry.py b/services/entry.py index 5ab84d1d..62d351a5 100644 --- a/services/entry.py +++ b/services/entry.py @@ -5,7 +5,7 @@ import sentry_sdk from dotenv import load_dotenv -from util import ApolloError, set_apollo_port +from util import ApolloError, install_log_masking, set_apollo_port load_dotenv() @@ -46,16 +46,12 @@ def _should_export_span(span): def _scrub_event(event: dict, _hint: dict) -> dict: """Mask keys in what Sentry is about to send. - Sentry scrubs frame locals by name, but not the exception message or a - set_context payload - and services raise ApolloError(500, str(e)) with the - request in scope. One hook here rather than each service remembering to - strip its own context before calling set_context. + Sentry scrubs frame locals by name, but not the exception message, a + set_context payload, or a breadcrumb - and services raise + ApolloError(500, str(e)) with the request in scope. The whole event rather + than a list of sections, so a section nobody thought of is covered too. """ - for section in ("exception", "contexts", "extra", "logentry"): - if section in event: - event[section] = mask_secrets(event[section]) - - return event + return mask_secrets(event) sentry_sdk.init( @@ -66,6 +62,8 @@ def _scrub_event(event: dict, _hint: dict) -> dict: enable_tracing=True, auto_enabling_integrations=False, before_send=_scrub_event, + # before_send covers error events only, and tracing is on. + before_send_transaction=_scrub_event, ) def call( @@ -111,6 +109,12 @@ def call( try: m = __import__(module_name, fromlist=["main"]) + + # Again here, after every import has had its chance to install a + # handler of its own. Some libraries add one that writes to stderr, + # which the bridge forwards to the caller line for line. + install_log_masking() + result = m.main(data) except ModuleNotFoundError as e: sentry_sdk.capture_exception(e) diff --git a/services/global_chat/tests/unit/test_mask_secrets.py b/services/global_chat/tests/unit/test_mask_secrets.py index 38425b81..f704bc39 100644 --- a/services/global_chat/tests/unit/test_mask_secrets.py +++ b/services/global_chat/tests/unit/test_mask_secrets.py @@ -18,6 +18,9 @@ # alphanumeric before the sk- means it is the tail of a word. "-sk-ant-zzzzzzzzzzzzzzzzzz", "_sk-ant-zzzzzzzzzzzzzzzzzz", + "sk-svcacct-abcdefgh", + "sk-abcdefghijklmnopqrstuvwx", + "Bearer sk-ant-abc12345", ] # Every one of these ends in a token the pattern used to bite through: task-, @@ -32,6 +35,12 @@ "const task = 'task-runner-config-value';", "obelisk-carving-notes", "whisk-attachment-guide", + # The other direction: a name that starts with sk- and carries on in + # words. Only a provider prefix or an unbroken high-entropy run counts. + "sk-antelope-migration-plan", + "step-sk-mapping-export-v2", + "load-sk-patient-records-job", + "--sk-ignore-case-sensitivity", ] diff --git a/services/langfuse_util.py b/services/langfuse_util.py index 5a3fc837..42825e9b 100644 --- a/services/langfuse_util.py +++ b/services/langfuse_util.py @@ -25,13 +25,21 @@ _MAX_DEPTH = 50 # A backstop for a key that turns up somewhere the name list cannot see, such -# as inside a string. The lookbehind is load-bearing: without it the sk- also -# matches the tail of ta[sk-], ri[sk-], di[sk-] and friends, which are ordinary -# words in the workflow YAML and job code this same function masks on the way -# to Langfuse. Only alphanumerics need excluding - a key can legitimately -# follow a hyphen or an underscore. +# as inside a string. Both directions matter, because this masks the workflow +# YAML, job code and service results it passes over as well as anything +# genuinely secret: without the lookbehind it eats the tail of ta[sk-], +# ri[sk-], di[sk-]; matching sk- plus any hyphenated words eats +# sk-antelope-migration-plan. So it anchors on the prefixes providers actually +# use, and otherwise wants an unbroken high-entropy run rather than words. _SECRET_VALUE_PATTERN = re.compile( - r"(? bool: _masking_filter = _MaskingFilter() +def install_log_masking() -> None: + """Put the mask on every handler that exists right now. + + Called at import and again from create_logger, because handlers appear at + two different times: some libraries install their own before any service + module is imported (langfuse attaches one to the httpx logger, and it + writes to stderr, which the bridge forwards to the caller line for line), + and the root handler only exists once basicConfig has run. + + A filter on a handler covers every record reaching that stream whatever + logger produced it - which a filter on a logger does not. + """ + root = logging.getLogger() + known = [root, *( + logger + for logger in root.manager.loggerDict.values() + if isinstance(logger, logging.Logger) + )] + + for logger in known: + for handler in logger.handlers: + if _masking_filter not in handler.filters: + handler.addFilter(_masking_filter) + + def create_logger(name: str) -> logging.Logger: """ Create or retrieve a logger with the given name. Logs to stdout by default. """ logging.basicConfig(level=logging.INFO, stream=sys.stdout) - - for handler in logging.getLogger().handlers: - if _masking_filter not in handler.filters: - handler.addFilter(_masking_filter) + install_log_masking() if name not in loggers: loggers[name] = logging.getLogger(name) return loggers[name] +install_log_masking() + + def set_apollo_port(p: int) -> None: """Set the port for Apollo services.""" global apollo_port # noqa: PLW0603 From 5185ffff26d58966ecb35ab2c53e150592fd2c32 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 19:54:58 +0000 Subject: [PATCH 15/23] Mask the event stream too Events reach the caller by a third route: not the result, not the log stream, so neither of those masks sees them. Nothing puts a key in one today, which is the moment to close it rather than after something does. --- services/streaming_util.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/services/streaming_util.py b/services/streaming_util.py index 08cb6dce..6147c5cb 100644 --- a/services/streaming_util.py +++ b/services/streaming_util.py @@ -11,6 +11,7 @@ from dataclasses import dataclass from typing import Any +from langfuse_util import mask_secrets from models import CLAUDE_SONNET # Shared status message pools for user-facing progress indicators. @@ -132,8 +133,16 @@ def _emit_event(self, event_type: str, data: dict[str, Any]) -> None: """ # Use EVENT: prefix format that bridge.ts expects # Bridge will convert this to proper SSE format + # + # Masked because this is a third way out to the caller, alongside the + # result and the log stream: it is neither, so neither of their masks + # sees it. Nothing puts a key in an event today, which is the point of + # doing it while that is still true. if self.stream: - print(f"EVENT:{event_type}:{json.dumps(data)}", flush=True) # noqa: T201 + print( # noqa: T201 + f"EVENT:{event_type}:{json.dumps(mask_secrets(data))}", + flush=True, + ) def start_stream(self) -> None: """ From 54469f21a8f728b34f074246480a360e4335b301 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 20:00:22 +0000 Subject: [PATCH 16/23] Use the shared mask instead of three hand-rolled lists Three services each stripped api_key from their Sentry context by name, in three slightly different ways, and one of them missed nested values and key-shaped strings entirely. They call the shared mask now. The stream manager is still dropped by name, because that is an object rather than data and not a secret at all. Also removes set_log_output and the filename it sets: nothing has read either since logging moved to stdout. --- services/job_chat/job_chat.py | 14 ++++++++++---- services/load_adaptor_docs/load_adaptor_docs.py | 5 ++--- services/util.py | 11 ----------- services/workflow_chat/workflow_chat.py | 14 ++++++++++---- 4 files changed, 22 insertions(+), 22 deletions(-) diff --git a/services/job_chat/job_chat.py b/services/job_chat/job_chat.py index 368f8573..919040ed 100644 --- a/services/job_chat/job_chat.py +++ b/services/job_chat/job_chat.py @@ -18,7 +18,7 @@ ) import sentry_sdk from langfuse import observe, propagate_attributes, get_client as get_langfuse_client -from langfuse_util import should_track, build_tags, build_generation_diff +from langfuse_util import should_track, build_tags, build_generation_diff, mask_secrets from util import ApolloError, create_logger, AdaptorSpecifier, add_page_prefix, APOLLO_VERSION from yaml_utils import INSPECT_JOB_CODE_TOOL, inspect_job_code from .prompt import build_prompt, build_error_correction_prompt @@ -742,9 +742,15 @@ def main(data_dict: dict) -> dict: Main entry point with improved error handling and input validation. """ try: - sentry_sdk.set_context("request_data", { - k: v for k, v in data_dict.items() if k not in ("api_key", "_stream_manager") - }) + # The stream manager is an object rather than data, so it is dropped. + # Everything else goes through the shared mask instead of a per-service + # name list, which catches nested values and key-shaped strings too. + sentry_sdk.set_context( + "request_data", + mask_secrets( + {k: v for k, v in data_dict.items() if k != "_stream_manager"}, + ), + ) data = Payload.from_dict(data_dict) diff --git a/services/load_adaptor_docs/load_adaptor_docs.py b/services/load_adaptor_docs/load_adaptor_docs.py index bf1cc13e..cc265f22 100644 --- a/services/load_adaptor_docs/load_adaptor_docs.py +++ b/services/load_adaptor_docs/load_adaptor_docs.py @@ -3,6 +3,7 @@ from typing import Dict, List, Any from psycopg2.extras import execute_values import sentry_sdk +from langfuse_util import mask_secrets from util import create_logger, ApolloError, apollo, AdaptorSpecifier, get_db_connection logger = create_logger("load_adaptor_docs") @@ -349,9 +350,7 @@ def main(data: dict) -> dict: """ logger.info("Starting load_adaptor_docs service...") - sentry_sdk.set_context("request_data", { - k: v for k, v in data.items() if k not in ["api_key"] - }) + sentry_sdk.set_context("request_data", mask_secrets(data)) # Validate required fields if "adaptor" not in data: diff --git a/services/util.py b/services/util.py index 80c06558..bdaa8d6b 100644 --- a/services/util.py +++ b/services/util.py @@ -58,21 +58,10 @@ def to_dict(self) -> dict: return error_dict -filename = None loggers: dict[str, logging.Logger] = {} apollo_port = 3000 -def set_log_output(f: str | None) -> None: - """Set the output file for logging.""" - global filename # noqa: PLW0603 - - if f is not None: - print(f"[entry.py] writing logs to {f}") # noqa: T201 - - filename = f - - class _MaskingFilter(logging.Filter): """Masks key-shaped values on their way out to the log stream. diff --git a/services/workflow_chat/workflow_chat.py b/services/workflow_chat/workflow_chat.py index 67ee6711..1377ad42 100644 --- a/services/workflow_chat/workflow_chat.py +++ b/services/workflow_chat/workflow_chat.py @@ -61,7 +61,7 @@ ) import sentry_sdk from langfuse import observe, propagate_attributes, get_client as get_langfuse_client -from langfuse_util import should_track, build_tags, build_generation_diff +from langfuse_util import should_track, build_tags, build_generation_diff, mask_secrets from util import ApolloError, create_logger, add_page_prefix, APOLLO_VERSION from .gen_project_prompt import build_prompt from workflow_chat.available_adaptors import get_available_adaptors @@ -705,9 +705,15 @@ def main(data_dict: dict) -> dict: Main entry point with improved error handling and input validation. """ try: - sentry_sdk.set_context("request_data", { - k: v for k, v in data_dict.items() if k not in ("api_key", "_stream_manager") - }) + # The stream manager is an object rather than data, so it is dropped. + # Everything else goes through the shared mask instead of a per-service + # name list, which catches nested values and key-shaped strings too. + sentry_sdk.set_context( + "request_data", + mask_secrets( + {k: v for k, v in data_dict.items() if k != "_stream_manager"}, + ), + ) data = Payload.from_dict(data_dict) From 6c5cc66cc8a885b2dff82494cd77b59e7030ac24 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 20:48:40 +0000 Subject: [PATCH 17/23] Trim the comments back to what earns its place The regex had three comments for one pattern, and the longest recounted the two shapes I tried before this one rather than explaining the one that is there. The depth guard said the same thing twice, once beside the constant and once in its test. Also drops a phrase that named what the temp payload holds - the same thing the commit messages were careful about, missed one layer down in a public repo. --- services/entry.py | 12 ++++---- .../tests/unit/test_mask_secrets.py | 2 -- services/langfuse_util.py | 30 ++++++++----------- 3 files changed, 18 insertions(+), 26 deletions(-) diff --git a/services/entry.py b/services/entry.py index 62d351a5..fe67040a 100644 --- a/services/entry.py +++ b/services/entry.py @@ -138,12 +138,12 @@ def call( def _finish(result: dict, output_path: str | None) -> dict: """Mask, write the result where the caller expects it, then hand it back. - Every path out of `call` already came through here so that an empty output - file means the run died rather than that it failed politely. Masking here - too means a value the server put in the payload cannot leave down a branch - someone forgot: most services catch broadly and rewrap as - `ApolloError(500, str(e))`, so masking per-branch would miss the one nearly - all of them take. + Every path out of `call` comes through here, which buys two things. The + output file is always written, so the bridge can read an empty one as the + run having died rather than as a polite failure. And a value the server put + on the payload cannot leave down a branch someone forgot: most services + catch broadly and rewrap as `ApolloError(500, str(e))`, so masking + per-branch would miss the one nearly all of them take. """ result = mask_secrets(result) diff --git a/services/global_chat/tests/unit/test_mask_secrets.py b/services/global_chat/tests/unit/test_mask_secrets.py index f704bc39..cdbac18b 100644 --- a/services/global_chat/tests/unit/test_mask_secrets.py +++ b/services/global_chat/tests/unit/test_mask_secrets.py @@ -53,8 +53,6 @@ def test_masks_a_secret_field_however_it_is_spelled(name: str) -> None: def test_does_not_recurse_without_bound() -> None: - # json.loads accepts around a thousand levels, and this runs over whatever - # a caller sends, so a deep payload must not take the process down. nested = {"leaf": "value"} for _ in range(600): nested = {"next": nested} diff --git a/services/langfuse_util.py b/services/langfuse_util.py index 42825e9b..1a408fe8 100644 --- a/services/langfuse_util.py +++ b/services/langfuse_util.py @@ -5,10 +5,10 @@ import yaml -# The control. Every field the server may fill in on a payload, not just the -# one it fills in today: a value here belongs to the deployment rather than to -# the caller. langfuse_public_key is deliberately absent - it is meant to be -# visible, and masking it would cost a useful identifier for nothing. +# Every field the server may fill in on a payload, not just the one it fills +# in today: a value under one of these belongs to the deployment rather than +# to the caller. langfuse_public_key is deliberately absent - it is meant to +# be visible, and masking it would cost a useful identifier for nothing. _SECRET_KEY_NAMES = { "api_key", "anthropic_api_key", @@ -19,25 +19,19 @@ "x_api_key", } -# Depth beyond which a payload is not being logged in good faith. Guards -# against a self-inflicted RecursionError: json.loads accepts around a -# thousand levels, and this runs over anything a caller sends. +# json.loads accepts around a thousand levels of nesting and this runs over +# whatever a caller sends, so a deep payload could otherwise take the process +# down with a RecursionError. _MAX_DEPTH = 50 -# A backstop for a key that turns up somewhere the name list cannot see, such -# as inside a string. Both directions matter, because this masks the workflow -# YAML, job code and service results it passes over as well as anything -# genuinely secret: without the lookbehind it eats the tail of ta[sk-], -# ri[sk-], di[sk-]; matching sk- plus any hyphenated words eats -# sk-antelope-migration-plan. So it anchors on the prefixes providers actually -# use, and otherwise wants an unbroken high-entropy run rather than words. +# A backstop for a key inside a string, where the name list cannot see it. +# Deliberately narrow: this also passes over workflow YAML, job code and +# service results, so a loose pattern corrupts a caller's own data. Hence a +# known provider prefix, or a run too long and unbroken to be a name. +# test_mask_secrets.py pins both directions. _SECRET_VALUE_PATTERN = re.compile( r"(? Date: Sun, 16 Aug 2026 21:02:40 +0000 Subject: [PATCH 18/23] Make closing a websocket actually stop the run Live testing found the close handler was aborting nothing. Elysia builds a fresh wrapper object for each websocket event, so the one the close handler receives is never the one the message handler stored against - the lookup missed every time, and the child kept generating. Keyed on the underlying socket now, which is what the two events share. Nothing caught this: the run settles on its own eventually, so the only signal was a python process still alive after the socket went away. The test looks for exactly that, and closes early enough that echo has not finished by itself - a longer wait passes whether or not the fix is there, which is how the first version of it fooled me. --- platform/src/middleware/services.ts | 16 +++++----- platform/test/ws-cancel.test.ts | 49 +++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 7 deletions(-) create mode 100644 platform/test/ws-cancel.test.ts diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index a11af9e0..505ddbe6 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -55,8 +55,10 @@ const callService = ( }; // The in-flight run for each open websocket, so closing the socket can stop -// it. Keyed on the socket itself and deleted as soon as the run settles, so a -// closed socket holds nothing. +// it. Keyed on ws.raw, not ws: Elysia builds a fresh wrapper per event, so the +// object the close handler receives is not the one the message handler saw, +// and the underlying socket is the only thing common to both. Deleted as soon +// as the run settles, so a closed socket holds nothing. const wsRuns = new WeakMap(); export default async (app: Elysia, port: number, auth: InstanceAuth) => { @@ -293,8 +295,8 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { // going away leaves the child generating, which is the cost the whole // change exists to stop. close(ws) { - wsRuns.get(ws)?.abort(); - wsRuns.delete(ws); + wsRuns.get(ws.raw)?.abort(); + wsRuns.delete(ws.raw); }, message(ws, message) { try { @@ -320,7 +322,7 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { const payload = applyKey(base, ws.data); const abort = new AbortController(); - wsRuns.set(ws, abort); + wsRuns.set(ws.raw, abort); // Two arguments rather than a chained catch: a chain would also // catch a throw from the success callback and report it to the @@ -339,14 +341,14 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { abort.signal ).then( (result) => { - wsRuns.delete(ws); + wsRuns.delete(ws.raw); ws.send({ event: "complete", data: result, }); }, (error) => { - wsRuns.delete(ws); + wsRuns.delete(ws.raw); ws.send({ event: "error", data: toErrorPayload(error), diff --git a/platform/test/ws-cancel.test.ts b/platform/test/ws-cancel.test.ts new file mode 100644 index 00000000..45e5ee0e --- /dev/null +++ b/platform/test/ws-cancel.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from "bun:test"; + +import setup from "../src/server"; +import { InstanceAuth } from "../src/auth/instance-auth"; + +// Closing a websocket has to stop the run behind it, and the only honest way +// to check that is to look for the child in the process table. A unit test +// cannot see it: Elysia builds a fresh wrapper object per event, so the +// bookkeeping that connects `close` back to the run it should abort is exactly +// the part that broke, silently, with every other test still green. +const port = 9877; + +const auth = new InstanceAuth({ lookup: () => null, hasGlobalKey: true }); +await setup(port, auth); + +const childCount = async () => { + const proc = Bun.spawn(["pgrep", "-f", "entry.py echo"], { + stdout: "pipe", + stderr: "ignore", + }); + const out = await new Response(proc.stdout).text(); + return out.split("\n").filter(Boolean).length; +}; + +const settle = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe("closing a websocket", () => { + it("stops the run behind it", async () => { + const socket = new WebSocket(`ws://localhost:${port}/services/echo`); + + await new Promise((open) => + socket.addEventListener("open", () => open()) + ); + + socket.send(JSON.stringify({ event: "start", data: { x: 1 } })); + + // Long enough for the child to exist, short enough that it is still + // running: spawning python through poetry takes the best part of a second. + await settle(250); + expect(await childCount()).toBeGreaterThan(0); + + socket.close(); + + // Short: echo finishes on its own in about a second, so a longer wait + // would see zero children whether or not closing the socket did anything. + await settle(400); + expect(await childCount()).toBe(0); + }, 20_000); +}); From b0c9c789424546f40c0567a2f36834e73bd4f458 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Sun, 16 Aug 2026 22:09:32 +0000 Subject: [PATCH 19/23] Restore the proof that the right value reaches the payload Masking echo took away the only test of this. The five instance-auth rows used to read the substituted value back off the response; with everything coming back "[REDACTED]" a swap that wrote the wrong value, or forwarded the caller's own, would look exactly like a correct one. applyResolvedKey is now its own exported function and asserted directly, so the substitution is checked without a service having to hand a value back to prove it. Paired with the InstanceAuth rows that pin which resolution a credential produces, that covers the ground the round trip did. Checked by breaking it both ways: forwarding the caller's credential fails two of the five, and blanking the field rather than dropping it fails one. --- platform/src/middleware/services.ts | 74 +++++++++++++++++------------ platform/test/apply-key.test.ts | 70 +++++++++++++++++++++++++++ 2 files changed, 114 insertions(+), 30 deletions(-) create mode 100644 platform/test/apply-key.test.ts diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index 505ddbe6..677e6f88 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -7,7 +7,7 @@ import describeModules, { type ModuleDescription, } from "../util/describe-modules"; import { isApolloError, toErrorPayload } from "../util/errors"; -import type { InstanceAuth } from "../auth/instance-auth"; +import type { InstanceAuth, KeyResolution } from "../auth/instance-auth"; const textEncoder = new TextEncoder(); @@ -54,6 +54,44 @@ const callService = ( } }; +/** Write the resolved key onto an outgoing payload. + * + * Exported so a test can assert what lands on the payload without a service + * reflecting it back: that was how this was covered before, and masking the + * reflection took the proof with it. + * + * The switch is explicit so the inbound-credential-never-forwarded invariant + * is structural rather than positional: a known client's stored key is + * swapped in (useKey), a request with no key of its own drops the field so + * python falls back to the global one (useGlobal), and an internal + * apollo() hop is forwarded exactly as received (passthrough). + */ +export const applyResolvedKey = ( + payload: Record, + resolution: KeyResolution +): Record => { + switch (resolution.kind) { + case "useKey": + payload.api_key = resolution.key; + break; + case "useGlobal": + delete payload.api_key; + break; + case "passthrough": + break; + default: { + // A new KeyResolution tag must be a compile error here, not a silent + // forward of the inbound credential. + const _exhaustive: never = resolution; + throw new Error( + `unhandled KeyResolution: ${(resolution as { kind: string }).kind}` + ); + } + } + + return payload; +}; + // The in-flight run for each open websocket, so closing the socket can stop // it. Keyed on ws.raw, not ws: Elysia builds a fresh wrapper per event, so the // object the close handler receives is not the one the message handler saw, @@ -65,35 +103,11 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { console.log("Loading routes:"); const modules = await describeModules(path.resolve("./services")); - // Apply the resolved key to an outgoing payload with an explicit switch so the - // inbound-credential-never-forwarded invariant is structural, not positional: a - // known client's stored key is swapped in (useKey), a NULL stored key (or a request - // with no api_key) drops the field so Python uses the global key (useGlobal), and an - // internal apollo() hop forwards the body exactly as received (passthrough). `ctx` is - // the upgrade-time context that carries lightningClient/internalCall: on POST the - // route ctx, on WS the captured ws.data, never a fresh per-message one. - const applyKey = (payload: Record, ctx: any) => { - const resolution = auth.resolveKey(ctx); - switch (resolution.kind) { - case "useKey": - payload.api_key = resolution.key; - break; - case "useGlobal": - delete payload.api_key; - break; - case "passthrough": - break; - default: { - // Exhaustiveness guard: a new KeyResolution tag must be a compile error - // here, not a silent forward of the inbound credential. - const _exhaustive: never = resolution; - throw new Error( - `unhandled KeyResolution: ${(resolution as { kind: string }).kind}` - ); - } - } - return payload; - }; + // `ctx` is the upgrade-time context carrying lightningClient/internalCall: + // on POST the route ctx, on WS the captured ws.data, never a fresh + // per-message one. + const applyKey = (payload: Record, ctx: any) => + applyResolvedKey(payload, auth.resolveKey(ctx)); const buildPayload = (ctx: any) => applyKey({ ...(ctx.body ?? {}), session_id: ctx.uuid }, ctx); diff --git a/platform/test/apply-key.test.ts b/platform/test/apply-key.test.ts new file mode 100644 index 00000000..ec446077 --- /dev/null +++ b/platform/test/apply-key.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from "bun:test"; + +import { applyResolvedKey } from "../src/middleware/services"; + +// What lands on the payload a service is about to run with. +// +// This used to be covered end to end: echo reflected its input, so a test +// could POST a credential and read the substituted value back off the +// response. Masking that reflection was right, and it took the proof with it — +// with everything coming back "[REDACTED]", a swap that wrote the wrong value, +// or forwarded the caller's own, would look identical to a correct one. +// +// So the assertion moved to the substitution itself. Paired with the +// InstanceAuth tests that pin which resolution a given credential produces, +// the two halves cover the same ground the round trip did, without a service +// having to hand a value back to prove it. +describe("applyResolvedKey", () => { + const CALLER = "sk-ant-the-caller-sent-this"; + const STORED = "sk-ant-what-the-server-substitutes"; + + it("substitutes the stored value for a known client", () => { + const payload = applyResolvedKey( + { x: 1, api_key: CALLER }, + { kind: "useKey", key: STORED } + ); + + expect(payload.api_key).toBe(STORED); + expect(payload.x).toBe(1); + }); + + // The invariant the whole auth layer exists for. + it("never forwards what the caller sent", () => { + const payload = applyResolvedKey( + { api_key: CALLER }, + { kind: "useKey", key: STORED } + ); + + expect(JSON.stringify(payload)).not.toContain(CALLER); + }); + + // Dropped rather than blanked: python falls back to the global key only when + // the field is absent. + it("drops the field entirely when the global key should serve", () => { + const payload = applyResolvedKey( + { x: 1, api_key: CALLER }, + { kind: "useGlobal" } + ); + + expect("api_key" in payload).toBe(false); + expect(payload.x).toBe(1); + }); + + it("leaves an internal hop's body exactly as received", () => { + const forwarded = "sk-ant-forwarded-by-an-internal-call"; + const payload = applyResolvedKey( + { api_key: forwarded }, + { kind: "passthrough" } + ); + + expect(payload.api_key).toBe(forwarded); + }); + + // The default branch is an exhaustiveness guard: a new resolution kind must + // fail loudly rather than fall through and forward the inbound credential. + it("refuses a resolution it does not recognise", () => { + expect(() => + applyResolvedKey({ api_key: CALLER }, { kind: "brand-new" } as never) + ).toThrow(/unhandled KeyResolution/); + }); +}); From e7a18c6a934307151a3fe09895abb832a444c47a Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Mon, 17 Aug 2026 13:38:33 +0100 Subject: [PATCH 20/23] copy logger before walking it --- services/util.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/services/util.py b/services/util.py index bdaa8d6b..77c39234 100644 --- a/services/util.py +++ b/services/util.py @@ -114,11 +114,14 @@ def install_log_masking() -> None: A filter on a handler covers every record reaching that stream whatever logger produced it - which a filter on a logger does not. + + loggerDict is copied rather than walked live: a thread creating a logger + resizes it mid-walk, which raises RuntimeError. """ root = logging.getLogger() known = [root, *( logger - for logger in root.manager.loggerDict.values() + for logger in list(root.manager.loggerDict.values()) if isinstance(logger, logging.Logger) )] From 9559cd35ffb7526aea606ac48097caa308f5711c Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Mon, 17 Aug 2026 14:24:12 +0100 Subject: [PATCH 21/23] handle killed processes and unparseable output --- platform/src/bridge.ts | 21 +++++++++++++++++++-- platform/src/util/errors.ts | 27 +++++++++++++++++++++++++++ platform/test/util/errors.test.ts | 18 ++++++++++++++++++ 3 files changed, 64 insertions(+), 2 deletions(-) diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index 9d957481..7071a6b0 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -5,7 +5,9 @@ import { rm } from "node:fs/promises"; import { getInternalToken } from "./auth/internal-token"; import { emptyResult, + malformedResult, subprocessFailed, + subprocessKilled, subprocessSpawnFailed, } from "./util/errors"; import pkg from "../../package.json"; @@ -104,7 +106,7 @@ export const run = async ( onLog?.(line); }); - proc.on("close", async (code) => { + proc.on("close", async (code, signal) => { // Clean up readline interfaces immediately to prevent race conditions rl.close(); rl2.close(); @@ -122,13 +124,28 @@ export const run = async ( console.error(e); } + // Checked before the exit code and the output: a killed process reports + // a null code and usually a truncated file, so either later check would + // misdiagnose it + if (signal) { + console.error("Python process killed by signal", signal); + return reject(subprocessKilled(scriptName, signal)); + } + if (code) { console.error("Python process exited with code", code); return reject(subprocessFailed(scriptName, code)); } if (text) { - return resolve(JSON.parse(text)); + // A parse error must be caught here: thrown, it escapes the close + // handler and the promise never settles + try { + return resolve(JSON.parse(text)); + } catch (e) { + console.error("Unparseable output from pythonland", e); + return reject(malformedResult(scriptName)); + } } // entry.py writes a result on every path it completes, including its own diff --git a/platform/src/util/errors.ts b/platform/src/util/errors.ts index cfd7fcbc..237cc0cd 100644 --- a/platform/src/util/errors.ts +++ b/platform/src/util/errors.ts @@ -80,6 +80,33 @@ export function subprocessSpawnFailed( ); } +/** The service process was killed by a signal - OOM, or SIGTERM during a + * deploy. A killed process reports a null exit code, so without this case it + * would be misread as having finished. */ +export function subprocessKilled( + service: string, + signal: string +): ApolloThrowable { + return new ApolloThrowable( + 500, + "SUBPROCESS_KILLED", + `Service "${service}" was killed by signal ${signal}`, + { service, signal } + ); +} + +/** The service exited cleanly but its output isn't valid JSON. Without this + * case the parse error escapes the close handler and the request never + * settles. */ +export function malformedResult(service: string): ApolloThrowable { + return new ApolloThrowable( + 502, + "MALFORMED_RESULT", + `Service "${service}" produced a result that could not be parsed`, + { service } + ); +} + /** The service exited cleanly but wrote nothing. entry.py writes a result on * every path it completes, so an empty file means the run died. */ export function emptyResult(service: string): ApolloThrowable { diff --git a/platform/test/util/errors.test.ts b/platform/test/util/errors.test.ts index 3f317a48..6633d530 100644 --- a/platform/test/util/errors.test.ts +++ b/platform/test/util/errors.test.ts @@ -4,7 +4,9 @@ import { ApolloThrowable, emptyResult, isApolloError, + malformedResult, subprocessFailed, + subprocessKilled, subprocessSpawnFailed, } from "../../src/util/errors"; @@ -58,4 +60,20 @@ describe("subprocess failures", () => { expect(details?.cause).toBe("ENOENT"); }); + + // OOM or a deploy's SIGTERM: the process reports a null exit code, so the + // signal is the only honest diagnosis. + it("names the signal when the process was killed", () => { + const error = subprocessKilled("embed_docsite", "SIGKILL"); + + expect(error.type).toBe("SUBPROCESS_KILLED"); + expect(error.details?.signal).toBe("SIGKILL"); + }); + + it("reports unparseable output as a bad gateway", () => { + const error = malformedResult("echo"); + + expect(error.type).toBe("MALFORMED_RESULT"); + expect(error.code).toBe(502); + }); }); From bee229a738696475eaf8666fff8819c776c64ae4 Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Mon, 17 Aug 2026 14:28:15 +0100 Subject: [PATCH 22/23] Fix websocket cancellation, exempt index writers --- platform/src/middleware/services.ts | 38 +++++++++--- platform/test/middleware/ws-cancel.test.ts | 72 ++++++++++++++++++++++ services/test_slow/test_slow.py | 24 ++++++++ 3 files changed, 127 insertions(+), 7 deletions(-) create mode 100644 platform/test/middleware/ws-cancel.test.ts create mode 100644 services/test_slow/test_slow.py diff --git a/platform/src/middleware/services.ts b/platform/src/middleware/services.ts index a11af9e0..ab6a3d11 100644 --- a/platform/src/middleware/services.ts +++ b/platform/src/middleware/services.ts @@ -37,6 +37,12 @@ export const heartbeatIntervalMs = (): number => { : HEARTBEAT_INTERVAL_MS; }; +// Killing these mid-run corrupts shared state that other services read: +// embed_docsite fills a fresh timestamped namespace that search_docsite treats +// as live the moment it is newest, and load_adaptor_docs commits its delete +// before its insert. They run to completion even when the caller has gone. +const NON_CANCELLABLE = new Set(["embed_docsite", "load_adaptor_docs"]); + const callService = ( m: ModuleDescription, port: number, @@ -45,6 +51,10 @@ const callService = ( onEvent?: (evt: string, payload: any) => void, signal?: AbortSignal ) => { + if (NON_CANCELLABLE.has(m.name)) { + signal = undefined; + } + if (m.type === "py") { return run(m.name, port, payload as any, onLog, onEvent, signal); } else { @@ -55,8 +65,11 @@ const callService = ( }; // The in-flight run for each open websocket, so closing the socket can stop -// it. Keyed on the socket itself and deleted as soon as the run settles, so a -// closed socket holds nothing. +// it. Keyed on ws.data, NOT on ws: Elysia builds a fresh ElysiaWS wrapper for +// every callback, so the object message() sees is never the object close() +// sees and a map keyed on it can never hit. ws.data is the upgrade context, +// created once per connection and carried on every wrapper. Deleted as soon +// as the run settles, so a closed socket holds nothing. const wsRuns = new WeakMap(); export default async (app: Elysia, port: number, auth: InstanceAuth) => { @@ -293,8 +306,8 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { // going away leaves the child generating, which is the cost the whole // change exists to stop. close(ws) { - wsRuns.get(ws)?.abort(); - wsRuns.delete(ws); + wsRuns.get(ws.data)?.abort(); + wsRuns.delete(ws.data); }, message(ws, message) { try { @@ -319,8 +332,13 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { const base: Record = { ...(message.data ?? {}) }; const payload = applyKey(base, ws.data); + // A second start frame on the same socket would otherwise + // orphan the first run: its controller leaves the map, so the + // close handler could never reach it again. + wsRuns.get(ws.data)?.abort(); + const abort = new AbortController(); - wsRuns.set(ws, abort); + wsRuns.set(ws.data, abort); // Two arguments rather than a chained catch: a chain would also // catch a throw from the success callback and report it to the @@ -339,14 +357,20 @@ export default async (app: Elysia, port: number, auth: InstanceAuth) => { abort.signal ).then( (result) => { - wsRuns.delete(ws); + // Only if it is still ours: a later start frame may have + // replaced this run's controller with its own. + if (wsRuns.get(ws.data) === abort) { + wsRuns.delete(ws.data); + } ws.send({ event: "complete", data: result, }); }, (error) => { - wsRuns.delete(ws); + if (wsRuns.get(ws.data) === abort) { + wsRuns.delete(ws.data); + } ws.send({ event: "error", data: toErrorPayload(error), diff --git a/platform/test/middleware/ws-cancel.test.ts b/platform/test/middleware/ws-cancel.test.ts new file mode 100644 index 00000000..52aa932a --- /dev/null +++ b/platform/test/middleware/ws-cancel.test.ts @@ -0,0 +1,72 @@ +import { afterAll, describe, expect, it } from "bun:test"; + +import setup from "../../src/server"; +import { InstanceAuth } from "../../src/auth/instance-auth"; + +// The bridge tests hand run() an AbortSignal directly, which proves the kill +// but not the wiring: nothing between the socket and the signal is exercised. +// That gap hid a real bug - Elysia rebuilds its ws wrapper for every callback, +// so a run keyed on the wrapper in message() was unreachable from close() and +// no websocket run was ever cancelled. This suite drives a real socket. +const port = 9874; + +const auth = new InstanceAuth({ lookup: () => null, hasGlobalKey: true }); +await setup(port, auth); + +// No `pgrep -c`: BSD pgrep has no count flag, and its usage error goes to +// stderr, so asking for one reads as "no processes". +const childCount = async () => { + const proc = Bun.spawn(["pgrep", "-f", "entry.py test_slow"], { + stdout: "pipe", + stderr: "pipe", + }); + const out = await new Response(proc.stdout).text(); + return out.trim().split("\n").filter(Boolean).length; +}; + +const waitForChildrenToClear = async (limitMs = 20_000) => { + const deadline = Date.now() + limitMs; + while (Date.now() < deadline) { + if ((await childCount()) === 0) { + return true; + } + await Bun.sleep(250); + } + return false; +}; + +// A failed run must not leave a two-minute child holding up the suite. +afterAll(() => { + Bun.spawnSync(["pkill", "-f", "entry.py test_slow"]); +}); + +describe("cancelling over a real websocket", () => { + it("kills the python child when the socket closes", async () => { + const socket = new WebSocket(`ws://localhost:${port}/services/test_slow`); + + // The service announces itself on stdout once the interpreter is truly + // inside main(); matching the process list any earlier hits the poetry + // wrapper and proves nothing. + const started = new Promise((resolve) => { + socket.addEventListener("message", ({ data }) => { + const evt = JSON.parse(String(data)); + if (evt.event === "event" && evt.type === "probe_started") { + resolve(); + } + }); + }); + + socket.addEventListener("open", () => { + socket.send( + JSON.stringify({ event: "start", data: { sleep_for: 120 } }) + ); + }); + + await started; + expect(await childCount()).toBeGreaterThan(0); + + socket.close(); + + expect(await waitForChildrenToClear()).toBe(true); + }, 90_000); +}); diff --git a/services/test_slow/test_slow.py b/services/test_slow/test_slow.py new file mode 100644 index 00000000..42f382f3 --- /dev/null +++ b/services/test_slow/test_slow.py @@ -0,0 +1,24 @@ +"""A mounted service that does nothing slowly, so route-level cancellation can +be tested over real connections. + +The bridge's own cancellation tests use the unmounted _cancel_probe and hand +run() a signal directly. That cannot see route wiring - it missed the +websocket close handler looking runs up under the wrong key - so this one is +mounted, like test_errors, and driven through a real socket. + +It announces itself before sleeping because the spawned command is +`poetry run python ...`: the process list matches on the poetry wrapper +seconds before the interpreter has booted, so waiting on that would prove +nothing. +""" + +import time + + +def main(data_dict: dict) -> dict: + print("EVENT:probe_started:{}", flush=True) # noqa: T201 + + seconds = data_dict.get("sleep_for", 1) + time.sleep(seconds) + + return {"slept": seconds} From 5f13f4f978f0ef84459188962285cd552f871706 Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Mon, 17 Aug 2026 16:11:55 +0100 Subject: [PATCH 23/23] drop the duplicate ws-cancel test --- platform/test/ws-cancel.test.ts | 49 --------------------------------- 1 file changed, 49 deletions(-) delete mode 100644 platform/test/ws-cancel.test.ts diff --git a/platform/test/ws-cancel.test.ts b/platform/test/ws-cancel.test.ts deleted file mode 100644 index 45e5ee0e..00000000 --- a/platform/test/ws-cancel.test.ts +++ /dev/null @@ -1,49 +0,0 @@ -import { describe, expect, it } from "bun:test"; - -import setup from "../src/server"; -import { InstanceAuth } from "../src/auth/instance-auth"; - -// Closing a websocket has to stop the run behind it, and the only honest way -// to check that is to look for the child in the process table. A unit test -// cannot see it: Elysia builds a fresh wrapper object per event, so the -// bookkeeping that connects `close` back to the run it should abort is exactly -// the part that broke, silently, with every other test still green. -const port = 9877; - -const auth = new InstanceAuth({ lookup: () => null, hasGlobalKey: true }); -await setup(port, auth); - -const childCount = async () => { - const proc = Bun.spawn(["pgrep", "-f", "entry.py echo"], { - stdout: "pipe", - stderr: "ignore", - }); - const out = await new Response(proc.stdout).text(); - return out.split("\n").filter(Boolean).length; -}; - -const settle = (ms: number) => new Promise((r) => setTimeout(r, ms)); - -describe("closing a websocket", () => { - it("stops the run behind it", async () => { - const socket = new WebSocket(`ws://localhost:${port}/services/echo`); - - await new Promise((open) => - socket.addEventListener("open", () => open()) - ); - - socket.send(JSON.stringify({ event: "start", data: { x: 1 } })); - - // Long enough for the child to exist, short enough that it is still - // running: spawning python through poetry takes the best part of a second. - await settle(250); - expect(await childCount()).toBeGreaterThan(0); - - socket.close(); - - // Short: echo finishes on its own in about a second, so a longer wait - // would see zero children whether or not closing the socket did anything. - await settle(400); - expect(await childCount()).toBe(0); - }, 20_000); -});