diff --git a/README.md b/README.md index 7850073..0a27c24 100644 --- a/README.md +++ b/README.md @@ -191,7 +191,7 @@ curl -o .opencode/plugins/llm-proxy.js \ | `OPENCODE_LLM_PROXY_TOOL_BRIDGE_POOL_SIZE` | `8` | Max concurrent in-flight requests using [tool calling](#tool-calling). | | `OPENCODE_LLM_PROXY_TOOL_BRIDGE_ACQUIRE_TIMEOUT_MS` | `10000` | Maximum wait for a tool-bridge slot, from 1 to 3,600,000 ms. | | `OPENCODE_LLM_PROXY_TOOL_BRIDGE_MAX_QUEUE` | `32` | Maximum tool-calling requests waiting for a bridge slot, from 0 to 10,000; excess requests receive `429`. | -| `OPENCODE_LLM_PROXY_KEEP_SESSIONS` | `false` | Set to `true` to retain temporary OpenCode sessions; otherwise they are deleted after use. | +| `OPENCODE_LLM_PROXY_KEEP_SESSIONS` | `false` | Set to `true` to retain temporary OpenCode sessions; otherwise they are deleted after use. Kept sessions are still swept after 24h idle (see [Session lifecycle](#session-lifecycle-safe-teardown)). | | `OPENCODE_LLM_PROXY_MODEL_ALIASES` | `{}` | JSON object mapping aliases to a model ID string or ordered array of fallback model IDs. | | `OPENCODE_LLM_PROXY_METRICS_ENABLED` | `false` | Set to `true` to expose the authenticated Prometheus endpoint at `GET /metrics`. | | `OPENCODE_LLM_PROXY_REMOTE_MEDIA_ENABLED` | `false` | Set to `true` to fetch remote media URLs and convert them to embedded data URLs. Leave disabled unless required. | @@ -206,6 +206,15 @@ Use `x-opencode-variant` to select an OpenCode model variant for a request. The Generation `temperature`, top-p (`top_p`/`topP`), and top-k (`topK`) values are validated and applied through the plugin's `chat.params` hook. Maximum-token fields (`max_tokens`, `max_completion_tokens`, `max_output_tokens`, and Gemini `maxOutputTokens`) are accepted where clients require them, but the current OpenCode SDK cannot enforce them. OpenAI and Anthropic requests reject unsupported controls (`stop`, `seed`, `frequency_penalty`, `presence_penalty`, `logprobs`, and `n`) with `400` instead of silently ignoring them. +### Session lifecycle (safe teardown) + +Each proxied request runs in a throwaway OpenCode session titled `Proxy: ` (**"Proxy: " is a reserved title prefix** — do not name your own sessions with it). When a request completes normally, the session is deleted immediately. When a request is abandoned mid-turn (client disconnect, timeout, or error), the session is **intentionally left in place**: deleting it immediately races the OpenCode server's final persist of the aborted turn, which surfaces as `FOREIGN KEY constraint failed` errors in the server log (observed at thousands per day under bulk clients that cancel slow streams, e.g. RAG pipelines with aggressive timeouts). + +Leaked sessions are reaped by a background sweep at plugin start and every 6 hours: `Proxy:`-titled sessions idle for more than 24 hours are deleted (long-idle deletes are race-free). Notes: + +- The first start after upgrading to this behavior will burn down any previously leaked backlog in one pass. +- With `OPENCODE_LLM_PROXY_KEEP_SESSIONS=true`, kept sessions carry the same title and are subject to the same 24h reap — if you need a longer inspection window, inspect within a day or copy what you need. + ```bash OPENCODE_LLM_PROXY_HOST=0.0.0.0 \ OPENCODE_LLM_PROXY_TOKEN=my-secret \ diff --git a/index.js b/index.js index 1561d4d..65d4d91 100644 --- a/index.js +++ b/index.js @@ -621,15 +621,99 @@ function validateUnsupportedControls(request) { } } -async function deleteSession(client, sessionID, keepSessions) { +// Bound the best-effort delete: client.session.delete() accepts no abort signal +// in older SDK shapes, and on the happy path it runs INSIDE the request's +// finally — a hung SDK connection there would block the response forever and +// hold the concurrency slot (observed as a full limiter wedge: /models keeps +// answering while every completion hangs). Race it with a short timer; on +// timeout give up — sweepStaleProxySessions reaps the session later anyway. +// Read per-call (not at module load) so tests and runtime config changes apply. +function deleteTimeoutMs() { + return integerEnv("OPENCODE_LLM_PROXY_DELETE_TIMEOUT_MS", 5000, { min: 100, max: 60000 }) +} + +async function deleteSession(client, sessionID, keepSessions, signal) { if (keepSessions || !sessionID || typeof client.session.delete !== "function") return + const attempt = client.session.delete({ path: { id: sessionID }, signal }) try { - await client.session.delete({ path: { id: sessionID } }) + await Promise.race([ + attempt, + new Promise((_, reject) => { + const timer = setTimeout(() => reject(new Error("session delete timed out")), deleteTimeoutMs()) + timer.unref?.() + signal?.addEventListener("abort", () => { + clearTimeout(timer) + reject(new Error("session delete aborted")) + }, { once: true }) + }), + ]) } catch { - // Best-effort cleanup for compatibility with older OpenCode clients. + // Best-effort cleanup: gave up (timeout/abort) — the stale-session sweep reaps it. } } +// Safe teardown: sessions abandoned mid-turn (client abort, error, timeout) are +// intentionally leaked at abandon time and reaped by this sweep once they are +// long idle. Deleting them immediately races the server's final persist of the +// aborted turn: the delete commits while the turn's last message insert is still +// in flight, and the insert then fails with +// "FOREIGN KEY constraint failed: insert into message" against the deleted +// session row. Under bulk clients that cancel slow streams (RAG pipelines with +// aggressive timeouts, retry loops), this produced thousands of constraint +// errors per day. Deleting only settled sessions mirrors how long-lived SDK +// consumers (e.g. chat bots that keep one session per conversation) avoid the +// race entirely. +// +// NOTE: "Proxy: " is a reserved title prefix — this plugin titles its throwaway +// sessions `Proxy: ` at creation. Do not name real sessions with this +// prefix: idle ones get reaped here. Internal helper — exported for tests. +export async function sweepStaleProxySessions(client, maxAgeMs = 24 * 60 * 60 * 1000) { + if (typeof client.session.list !== "function") return + let sessions + try { + const result = await client.session.list() + sessions = result.data ?? [] + } catch (error) { + await safeLog(client, "warn", "OpenAI proxy stale-session sweep could not list sessions", { + error: error instanceof Error ? error.message : String(error), + }) + return + } + const cutoff = Date.now() - maxAgeMs + let reaped = 0 + for (const s of sessions) { + if (!s?.title?.startsWith("Proxy:")) continue + const updated = Number(s.time_updated ?? s.timeUpdated ?? 0) + // Fail safe: only treat as a stale ms-epoch timestamp. Rejects seconds-epoch + // values, NaN, and missing fields (those sessions are left for the next sweep + // rather than risk mass-deleting fresh ones on a units mismatch). + if (!(updated > 1e12) || updated >= cutoff) continue + try { + await client.session.delete({ path: { id: s.id } }) + reaped++ + } catch { + // skip; a later sweep retries + } + } + if (reaped > 0) { + await safeLog(client, "info", "OpenAI proxy stale-session sweep", { reaped }) + } +} + +const SWEEP_INTERVAL_MS = 6 * 60 * 60 * 1000 + +// Run the sweep at startup and every SWEEP_INTERVAL_MS so long-lived processes +// still reap leaked sessions. The timer is unref'd: it never keeps the process +// alive on its own. +function scheduleSessionSweeps(client) { + void sweepStaleProxySessions(client).catch(() => {}) + const timer = setInterval(() => { + void sweepStaleProxySessions(client).catch(() => {}) + }, SWEEP_INTERVAL_MS) + timer.unref?.() + return timer +} + function setGenerationControls(sessionID, controls) { if (!sessionID || !controls || Object.keys(controls).length === 0) return const state = getState() @@ -665,6 +749,7 @@ async function executePrompt(client, _request, model, messages, system, callerTo const tools = await getDisabledTools(client) let sessionID + let settled = false // safe teardown: only delete sessions whose turn fully settled try { const session = await client.session.create({ body: { title: `Proxy: ${model.id}` }, signal: options.signal }) sessionID = session.data.id @@ -687,10 +772,14 @@ async function executePrompt(client, _request, model, messages, system, callerTo if (!content && completion.data.info?.error) throw new Error(completion.data.info.error.message ?? "Model call failed.") + settled = true return { content, structured, toolCalls: [], completion, request: _request, sessionID } } finally { clearGenerationControls(sessionID) - await deleteSession(client, sessionID, options.keepSessions) + // Deleting an aborted/errored session here races the server's final persist + // of the aborted turn (FOREIGN KEY constraint failures under bulk cancels). + // Leak it instead; sweepStaleProxySessions reaps it once long idle. + if (settled) await deleteSession(client, sessionID, options.keepSessions, options.signal) } } @@ -1277,7 +1366,9 @@ async function runAgentTurn(client, model, messages, system, callerTools, onChun } } } catch (error) { - await deleteSession(client, sessionID, options.keepSessions) + // Safe teardown: no delete on abort/error — it races the server's final + // persist of the aborted turn (FOREIGN KEY failures under bulk cancels). + // The session leaks inert; sweepStaleProxySessions reaps it once long idle. throw error } finally { removeAbortListener() @@ -1297,7 +1388,7 @@ async function runAgentTurn(client, model, messages, system, callerTools, onChun })) if (errorMessage && toolCalls.length === 0) { - await deleteSession(client, sessionID, options.keepSessions) + // Safe teardown: leak-on-error (was: delete then throw). throw new Error(errorMessage) } @@ -1308,7 +1399,7 @@ async function runAgentTurn(client, model, messages, system, callerTools, onChun const messagesResult = await client.session.messages({ path: { id: sessionID }, signal: options.signal }) assistantEntry = (messagesResult.data ?? []).filter((m) => m.info?.role === "assistant").at(-1) } catch (error) { - await deleteSession(client, sessionID, options.keepSessions) + // Safe teardown: leak-on-error (was: delete then throw). throw error } const assistantInfo = assistantEntry?.info @@ -1330,7 +1421,10 @@ async function runAgentTurn(client, model, messages, system, callerTools, onChun finish: toolCalls.length > 0 ? "tool_calls" : assistantInfo?.finish, structured: assistantInfo?.structured, } - await deleteSession(client, sessionID, options.keepSessions) + // Safe teardown: only delete when the turn completed without a server-side + // error (errorMessage with partial tool calls still lands here — the turn did + // not settle cleanly, so leak it for the sweep instead). + if (!errorMessage) await deleteSession(client, sessionID, options.keepSessions, options.signal) return result } @@ -2754,6 +2848,10 @@ export const OpenAIProxyPlugin = async ({ client }) => { state.started = true state.server = server + // Safe teardown: reap stale leaked "Proxy:" sessions now and periodically + // (fire-and-forget; never blocks or fails startup). + state.sweepTimer = scheduleSessionSweeps(client) + await safeLog(client, "info", "OpenAI proxy server started", { hostname, port, diff --git a/index.test.js b/index.test.js index c27c36e..413d959 100644 --- a/index.test.js +++ b/index.test.js @@ -1173,7 +1173,9 @@ test("completed sessions are deleted when the client supports deletion", async ( })) assert.equal(response.status, 200) - assert.deepEqual(deleted, [{ path: { id: "sess-resp-1" } }]) + assert.equal(deleted.length, 1) + assert.equal(deleted[0].path.id, "sess-resp-1") + assert.ok(deleted[0].signal instanceof AbortSignal, "delete carries the request signal") }) test("structured output schema is forwarded and structured data is extracted", async () => { diff --git a/test/safe-teardown.test.js b/test/safe-teardown.test.js new file mode 100644 index 0000000..ead325f --- /dev/null +++ b/test/safe-teardown.test.js @@ -0,0 +1,357 @@ +import assert from "node:assert/strict" +import http from "node:http" +import test from "node:test" + +import { createProxyFetchHandler, sweepStaleProxySessions } from "../index.js" + +const TOKENS = { input: 2, output: 1, reasoning: 0, cache: { read: 0, write: 0 } } + +function createMockClient({ promptMode = "hang", streamEvents = null } = {}) { + const state = { created: [], deleted: [] } + const records = new Map() + let pendingSubscription = null + return { + state, + app: { log: async () => {} }, + tool: { ids: async () => ({ data: [] }) }, + config: { + providers: async () => ({ + data: { + providers: [{ id: "openai", models: { first: { id: "first" } } }], + }, + }), + }, + session: { + create: async () => { + const id = `session-${state.created.length + 1}` + state.created.push(id) + records.set(id, { ready: null }) + return { data: { id } } + }, + prompt: async ({ signal }) => { + if (promptMode === "resolve") { + return { + data: { + info: { finish: "stop", tokens: TOKENS }, + parts: [{ type: "text", text: "hello" }], + }, + } + } + await new Promise((_, reject) => { + if (signal?.aborted) return reject(signal.reason ?? new Error("aborted")) + signal?.addEventListener("abort", () => reject(signal.reason ?? new Error("aborted")), { once: true }) + }) + }, + promptAsync: async ({ path }) => { + pendingSubscription = path.id + return { data: true } + }, + messages: async ({ path }) => ({ + data: [{ + info: { role: "assistant", tokens: TOKENS, finish: "stop" }, + parts: [{ type: "text", text: "streamed answer" }], + }], + }), + list: async () => ({ data: [] }), + delete: async ({ path }) => { + state.deleted.push(path.id) + return { data: true } + }, + }, + event: { + subscribe: async ({ signal }) => { + // The plugin subscribes BEFORE calling promptAsync, so the session ID is + // resolved lazily at first next() — by then promptAsync has run. + let events = null + let index = 0 + return { + stream: { + async next() { + if (events === null) { + const sessionID = pendingSubscription + events = streamEvents ? streamEvents(sessionID) : [] + } + if (index < events.length) { + return { value: events[index++], done: false } + } + // Exhausted: hang until aborted (mimics a stream that never idles). + await new Promise((_, reject) => { + if (signal?.aborted) return reject(signal.reason ?? new Error("aborted")) + signal?.addEventListener("abort", () => reject(signal.reason ?? new Error("aborted")), { once: true }) + }) + }, + async return() { + return { done: true } + }, + [Symbol.asyncIterator]() { + return this + }, + }, + } + }, + }, + } +} + +const deltaThenIdle = (sessionID) => ([ + { + type: "message.part.delta", + properties: { sessionID, field: "text", delta: "streamed" }, + }, + { type: "session.idle", properties: { sessionID } }, +]) + +async function withServer(client, run) { + const handler = createProxyFetchHandler(client) + const server = http.createServer(async (incoming, outgoing) => { + const controller = new AbortController() + const abort = () => controller.abort(new Error("HTTP client disconnected")) + incoming.once("aborted", abort) + outgoing.once("close", () => { + if (!outgoing.writableEnded) abort() + }) + try { + const chunks = [] + for await (const chunk of incoming) chunks.push(chunk) + const address = server.address() + const request = new Request(`http://127.0.0.1:${address.port}${incoming.url}`, { + method: incoming.method, + headers: incoming.headers, + body: chunks.length ? Buffer.concat(chunks) : undefined, + signal: controller.signal, + }) + const response = await handler(request) + outgoing.writeHead(response.status, Object.fromEntries(response.headers)) + if (!response.body) return outgoing.end() + const reader = response.body.getReader() + const cancel = () => reader.cancel(new Error("HTTP client disconnected")).catch(() => {}) + outgoing.once("close", cancel) + while (true) { + const { done, value } = await reader.read() + if (done) break + if (!outgoing.write(value)) await new Promise((resolve) => outgoing.once("drain", resolve)) + } + outgoing.end() + } catch { + if (!outgoing.writableEnded) outgoing.end() + } + }) + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)) + const port = server.address().port + try { + await run(port) + } finally { + await new Promise((resolve) => server.close(resolve)) + } +} + +function postCompletion(port, { stream = false, timeoutMs = 1000 } = {}) { + const controller = new AbortController() + const promise = fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "openai/first", + messages: [{ role: "user", content: "hello" }], + stream, + }), + signal: controller.signal, + }).catch((error) => error) + setTimeout(() => controller.abort(new Error("client gave up")), timeoutMs) + return promise +} + +test("non-streaming success deletes the throwaway session", async () => { + const client = createMockClient({ promptMode: "resolve" }) + await withServer(client, async (port) => { + const response = await postCompletion(port, { timeoutMs: 5000 }) + assert.equal(response.status, 200) + await new Promise((resolve) => setTimeout(resolve, 50)) + assert.equal(client.state.created.length, 1) + assert.deepEqual(client.state.deleted, client.state.created) + }) +}) + +test("non-streaming client abort does NOT delete the session (FOREIGN KEY race)", async () => { + const client = createMockClient({ promptMode: "hang" }) + await withServer(client, async (port) => { + await postCompletion(port, { timeoutMs: 150 }) + await new Promise((resolve) => setTimeout(resolve, 100)) + assert.equal(client.state.created.length, 1) + assert.deepEqual(client.state.deleted, [], "aborted session must leak, not delete") + }) +}) + +test("keepSessions=true never deletes, even on success", async () => { + const client = createMockClient({ promptMode: "resolve" }) + process.env.OPENCODE_LLM_PROXY_KEEP_SESSIONS = "true" + try { + await withServer(client, async (port) => { + const response = await postCompletion(port, { timeoutMs: 5000 }) + assert.equal(response.status, 200) + await new Promise((resolve) => setTimeout(resolve, 50)) + assert.equal(client.state.created.length, 1) + assert.deepEqual(client.state.deleted, []) + }) + } finally { + delete process.env.OPENCODE_LLM_PROXY_KEEP_SESSIONS + } +}) + +const DAY = 24 * 60 * 60 * 1000 + +test("sweep reaps only stale Proxy: sessions", async () => { + const staleProxy = { id: "stale-proxy", title: "Proxy: openai/first", time_updated: Date.now() - 2 * DAY } + const freshProxy = { id: "fresh-proxy", title: "Proxy: openai/first", time_updated: Date.now() - 1000 } + const staleOther = { id: "stale-other", title: "My chat", time_updated: Date.now() - 2 * DAY } + const unknownAge = { id: "unknown-age", title: "Proxy: openai/first" } + const deleted = [] + const client = { + app: { log: async () => {} }, + session: { + list: async () => ({ data: [staleProxy, freshProxy, staleOther, unknownAge] }), + delete: async ({ path }) => { deleted.push(path.id); return { data: true } }, + }, + } + await sweepStaleProxySessions(client) + assert.deepEqual(deleted, ["stale-proxy"], "only the stale Proxy: session is reaped") +}) + +test("sweep is a no-op when session.list is unavailable or fails", async () => { + const noList = { app: { log: async () => {} }, session: {} } + await sweepStaleProxySessions(noList) // must not throw + let listed = false + const failing = { + app: { log: async () => {} }, + session: { + list: async () => { listed = true; throw new Error("boom") }, + delete: async () => { throw new Error("must not be called") }, + }, + } + await sweepStaleProxySessions(failing) // must not throw + assert.ok(listed) +}) + +test("sweep keeps going when an individual delete fails", async () => { + const DAY = 24 * 60 * 60 * 1000 + const deleted = [] + const client = { + app: { log: async () => {} }, + session: { + list: async () => ({ + data: [ + { id: "stale-1", title: "Proxy: openai/first", time_updated: Date.now() - 2 * DAY }, + { id: "stale-2", title: "Proxy: openai/first", time_updated: Date.now() - 2 * DAY }, + ], + }), + delete: async ({ path }) => { + if (path.id === "stale-1") throw new Error("delete failed") + deleted.push(path.id) + return { data: true } + }, + }, + } + await sweepStaleProxySessions(client) + assert.deepEqual(deleted, ["stale-2"], "one failing delete must not stop the sweep") +}) + +test("sweep ignores seconds-epoch timestamps (units mismatch fails safe)", async () => { + const DAY = 24 * 60 * 60 * 1000 + const deleted = [] + const client = { + app: { log: async () => {} }, + session: { + list: async () => ({ + data: [ + // seconds-epoch "2 days ago" — magnitude says seconds, not ms: keep. + { id: "seconds-epoch", title: "Proxy: openai/first", time_updated: Math.floor((Date.now() - 2 * DAY) / 1000) }, + ], + }), + delete: async ({ path }) => { deleted.push(path.id); return { data: true } }, + }, + } + await sweepStaleProxySessions(client) + assert.deepEqual(deleted, [], "seconds-epoch timestamps must not classify as stale ms-epoch") +}) + +test("hung session.delete does not block the response past the delete timeout", async () => { + process.env.OPENCODE_LLM_PROXY_DELETE_TIMEOUT_MS = "250" + try { + const state = { created: [], deleted: [] } + const client = { + state, + app: { log: async () => {} }, + tool: { ids: async () => ({ data: [] }) }, + config: { + providers: async () => ({ + data: { providers: [{ id: "openai", models: { first: { id: "first" } } }] }, + }), + }, + session: { + create: async () => { + const id = `session-${state.created.length + 1}` + state.created.push(id) + return { data: { id } } + }, + prompt: async () => ({ + data: { + info: { finish: "stop", tokens: TOKENS }, + parts: [{ type: "text", text: "hello" }], + }, + }), + list: async () => ({ data: [] }), + delete: () => new Promise(() => {}), // never resolves — hung SDK connection + }, + } + await withServer(client, async (port) => { + const t0 = Date.now() + const response = await postCompletion(port, { timeoutMs: 8000 }) + const elapsed = Date.now() - t0 + assert.equal(response.status, 200) + assert.ok(elapsed < 3000, `response took ${elapsed}ms; hung delete must not block it`) + assert.equal(client.state.created.length, 1) + assert.deepEqual(client.state.deleted, [], "hung delete means no completed delete") + }) + } finally { + delete process.env.OPENCODE_LLM_PROXY_DELETE_TIMEOUT_MS + } +}) + +test("streaming success still deletes the throwaway session", async () => { + const client = createMockClient({ streamEvents: deltaThenIdle }) + await withServer(client, async (port) => { + const response = await postCompletion(port, { stream: true, timeoutMs: 5000 }) + assert.equal(response.status, 200) + await response.arrayBuffer() + await new Promise((resolve) => setTimeout(resolve, 50)) + assert.equal(client.state.created.length, 1) + assert.deepEqual(client.state.deleted, client.state.created) + }) +}) + +test("streaming client abort does NOT delete the session (FOREIGN KEY race)", async () => { + const client = createMockClient({ streamEvents: () => [ + { type: "message.part.delta", properties: { sessionID: null, field: "text", delta: "partial" } }, + ] }) + await withServer(client, async (port) => { + await postCompletion(port, { stream: true, timeoutMs: 150 }) + await new Promise((resolve) => setTimeout(resolve, 150)) + assert.equal(client.state.created.length, 1) + assert.deepEqual(client.state.deleted, [], "aborted stream session must leak, not delete") + }) +}) + +test("streaming session.error does NOT delete the session", async () => { + const client = createMockClient({ streamEvents: (sessionID) => [ + { type: "session.error", properties: { sessionID, error: { message: "model exploded" } } }, + ] }) + await withServer(client, async (port) => { + const response = await postCompletion(port, { stream: true, timeoutMs: 5000 }).catch((e) => e) + // The proxy surfaces the error (HTTP 500 or a stream error event); either way + // the session must NOT be deleted. + assert.ok(response instanceof Error || response.status >= 400 || response.status === 200) + await new Promise((resolve) => setTimeout(resolve, 100)) + assert.equal(client.state.created.length, 1) + assert.deepEqual(client.state.deleted, [], "errored stream session must leak, not delete") + }) +})