import test from "node:test"; import assert from "node:assert/strict"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-active-stream-lifecycle-")); process.env.DATA_DIR = TEST_DATA_DIR; const core = await import("../../src/lib/db/core.ts"); const usageHistory = await import("../../src/lib/usage/usageHistory.ts"); const callLogs = await import("../../src/lib/usage/callLogs.ts"); // Captured stream chunks carry a per-chunk arrival-time prefix ("[HH:MM:SS.mmm] ") // added by the request logger for streaming-latency observability (#5834). Strip it // before comparing the raw chunk payload. const stripChunkTs = (chunk: string): string => chunk.replace(/^\[\d{2}:\d{2}:\d{2}\.\d{3}\] /, ""); test.after(() => { core.resetDbInstance(); fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); }); // ─── Helper: Simulates /api/logs/[id] API route logic ────────────────────── // // The production route uses 3-tier lookup: // 1. DB (getCallLogById) // 2. completedDetails in-memory cache (getCompletedDetails) // 3. pendingById in-memory active requests (getPendingById) // // Each tier constructs the API response shape including pipelinePayloads. function buildApiResponseFromPending(id: string): Record | null { const active = usageHistory.getPendingById().get(id); if (!active) return null; const pipelinePayloads: Record = { clientRequest: active.clientRequest ?? null, providerRequest: active.providerRequest ?? null, providerResponse: active.providerResponse ?? null, clientResponse: active.clientResponse ?? null, streamChunks: active.streamChunks ?? null, }; return { id: active.id, timestamp: new Date(active.startedAt).toISOString(), method: "", path: active.clientEndpoint || "", status: 0, model: active.model, provider: active.provider, connectionId: active.connectionId, duration: Date.now() - active.startedAt, detailState: "in-flight", active: true, pipelinePayloads, hasPipelineDetails: true, }; } function buildApiResponseFromCompleted(id: string): Record | null { const completed = usageHistory.getCompletedDetails(); const inMem = completed.get(id); if (!inMem) return null; const pipelinePayloads: Record = { clientRequest: inMem.clientRequest ?? null, providerRequest: inMem.providerRequest ?? null, providerResponse: inMem.providerResponse ?? null, clientResponse: inMem.clientResponse ?? null, streamChunks: inMem.streamChunks ?? null, }; return { id: inMem.id, timestamp: new Date(inMem.startedAt).toISOString(), path: inMem.clientEndpoint || "", status: 0, model: inMem.model, provider: inMem.provider, connectionId: inMem.connectionId, duration: Date.now() - inMem.startedAt, detailState: "in-memory", active: false, pipelinePayloads, hasPipelineDetails: true, }; } // ─── Helper: Simulates frontend streamChunksText IIFE ────────────────────── function computeStreamChunksText( debugEnabled: boolean, pipelinePayloads: Record | null | undefined ): string | null { if (!debugEnabled || !pipelinePayloads?.streamChunks) return null; let chunks: unknown = pipelinePayloads.streamChunks; if (typeof chunks === "string") { try { chunks = JSON.parse(chunks); } catch { return chunks; } } if (chunks && typeof chunks === "object") { try { return Object.entries(chunks as Record) .map(([stage, arr]) => { const joined = Array.isArray(arr) ? (arr as string[]).join("") : String(arr); return `--- ${stage} ---\n${joined}`; }) .join("\n\n"); } catch { return JSON.stringify(chunks, null, 2); } } return null; } // ─── Helper: Simulates frontend openDetail state merge ───────────────────── function mergeDetailData( prev: Record | null, data: Record ): Record { const dataHasPipeline = data?.pipelinePayloads && Object.keys(data.pipelinePayloads || {}).length > 0; return { ...prev, ...data, pipelinePayloads: dataHasPipeline ? data.pipelinePayloads : prev?.pipelinePayloads, }; } // ─── Test: Full lifecycle ────────────────────────────────────────────────── test("streamChunks survive the full lifecycle: in-flight → completed → persisted", async () => { usageHistory.clearPendingRequests(); const model = "gpt-4"; const provider = "openai"; const connectionId = "conn-lifecycle-1"; // ── Phase 1: Track a pending request ── const requestId = usageHistory.trackPendingRequest(model, provider, connectionId, true, { clientRequest: { messages: [{ role: "user", content: "hello" }] }, providerRequest: { model: "gpt-4" }, clientEndpoint: "/v1/chat/completions", }); assert.ok(requestId, "trackPendingRequest should return a request ID"); assert.ok( usageHistory.getPendingById().has(requestId), "request ID should be in pendingById immediately" ); // ── Phase 2: Simulate streaming — chunks arrive gradually ── // Round 1: provider chunks usageHistory.updatePendingRequestStreamChunks(model, provider, connectionId, { provider: ['data: {"type":"message_start"}\n\n'], openai: [], client: [], }); // Verify via pendingById (what the API reads) const apiResponse1 = buildApiResponseFromPending(requestId); assert.ok(apiResponse1, "API should find in-flight request"); assert.ok(apiResponse1!.pipelinePayloads, "should have pipelinePayloads"); assert.ok( (apiResponse1!.pipelinePayloads as Record).streamChunks, "streamChunks should be present in API response" ); // Simulate frontend streamChunksText const text1 = computeStreamChunksText( true, apiResponse1!.pipelinePayloads as Record ); assert.ok(text1, "streamChunksText should be non-null with debugEnabled"); assert.ok(text1!.includes("message_start"), "streamChunksText should contain the chunk content"); // Verify frontend state merge const initialDetail = mergeDetailData(null, apiResponse1 as unknown as Record); assert.ok( (initialDetail.pipelinePayloads as Record).streamChunks, "streamChunks should survive the openDetail state merge" ); const text1after = computeStreamChunksText( true, initialDetail.pipelinePayloads as Record ); assert.ok(text1after, "streamChunksText should work after state merge"); assert.ok(text1after!.includes("message_start"), "content preserved after state merge"); // Round 2: openai chunks arrive (simulating proxy relay) usageHistory.updatePendingRequestStreamChunks(model, provider, connectionId, { provider: ['data: {"type":"message_start"}\n\n'], openai: ['data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n'], client: [], }); // Round 3: client (converted) chunk arrives usageHistory.updatePendingRequestStreamChunks(model, provider, connectionId, { provider: ['data: {"type":"message_start"}\n\n'], openai: ['data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n'], client: ['data: {"content":"Hello"}\n\n'], }); // Verify API sees the latest (shows all stages now) const apiResponse2 = buildApiResponseFromPending(requestId); const streamChunks2 = (apiResponse2!.pipelinePayloads as Record) .streamChunks as Record; assert.equal(streamChunks2.provider.length, 1, "provider chunks: 1"); assert.equal(streamChunks2.openai.length, 1, "openai chunks: 1"); assert.equal(streamChunks2.client.length, 1, "client chunks: 1"); // Verify frontend text includes all 3 stages const text2 = computeStreamChunksText( true, apiResponse2!.pipelinePayloads as Record ); assert.ok(text2!.includes("--- provider ---"), "should include provider stage"); assert.ok(text2!.includes("--- openai ---"), "should include openai stage"); assert.ok(text2!.includes("--- client ---"), "should include client stage"); // Verify debugEnabled=false hides the stream const textHidden = computeStreamChunksText( false, apiResponse2!.pipelinePayloads as Record ); assert.equal(textHidden, null, "streamChunksText should be null when debugEnabled=false"); // Verify null pipelinePayloads hides the stream const textNoPayload = computeStreamChunksText(true, null); assert.equal(textNoPayload, null, "streamChunksText should be null without pipelinePayloads"); const textNoChunks = computeStreamChunksText(true, {}); assert.equal(textNoChunks, null, "streamChunksText should be null without streamChunks key"); // ── Phase 3: Simulate request completion ── usageHistory.finalizeMostRecentPendingRequest(model, provider, connectionId, { status: 200, model, provider, clientResponse: { choices: [{ message: { content: "Hello" } }] }, }); // The request should be in completedDetails now assert.ok( !usageHistory.getPendingById().has(requestId), "request should be removed from pendingById after finalization" ); const completedResponse = buildApiResponseFromCompleted(requestId); assert.ok(completedResponse, "API should find request in completedDetails"); assert.ok( (completedResponse!.pipelinePayloads as Record).streamChunks, "streamChunks should be in completedDetails" ); // Verify frontend can render from completed response const completedText = computeStreamChunksText( true, completedResponse!.pipelinePayloads as Record ); assert.ok(completedText, "streamChunksText should work from completed data"); assert.ok(completedText!.includes("Hello"), "content preserved after completion"); // ── Phase 4: Persist to DB (simulates saveCallLog) ── await callLogs.saveCallLog({ id: requestId, method: "POST", path: "/v1/chat/completions", status: 200, model, requestedModel: model, provider, connectionId, duration: 5000, tokens: { in: 10, out: 20 }, requestBody: { messages: [{ role: "user", content: "hello" }] }, responseBody: { choices: [{ message: { content: "Hello" } }] }, error: null, sourceFormat: "openai", targetFormat: "openai", comboName: null, comboStepId: null, comboExecutionKey: null, tokensCompressed: null, cacheSource: "upstream", apiKeyId: null, apiKeyName: null, noLog: false, pipelinePayloads: completedResponse!.pipelinePayloads as Record, }); // Verify DB has the entry const dbEntry = await callLogs.getCallLogById(requestId); assert.ok(dbEntry, "should find persisted call log by the same ID"); assert.ok(dbEntry.pipelinePayloads, "persisted entry should have pipelinePayloads"); // The pipelinePayloads in the DB may have been compacted/truncated. // streamChunks may or may not be there depending on captureStreamChunks, // but the ID must match so the API can find it. assert.equal( (dbEntry as Record).id, requestId, "DB entry ID should match the original request ID" ); }); test("streamChunksText renders progressive updates correctly", () => { // Simulate the scenario where streamChunks grows between polls // Poll 1: first chunk arrives const payload1 = { streamChunks: { provider: ['data: {"content":"A"}\n\n'], openai: [], client: [] }, }; const text1 = computeStreamChunksText(true, payload1); assert.ok(text1, "poll 1 should produce text"); assert.ok(text1!.includes("A"), "poll 1 should show chunk A"); // Poll 2: second chunk appended const payload2 = { streamChunks: { provider: ['data: {"content":"A"}\n\n', 'data: {"content":"B"}\n\n'], openai: [], client: [], }, }; const text2 = computeStreamChunksText(true, payload2); assert.ok(text2!.includes("A"), "poll 2 should still show chunk A"); assert.ok(text2!.includes("B"), "poll 2 should show newly arrived chunk B"); // The joined text should be the concatenation assert.ok( text2!.includes('data: {"content":"A"}\n\ndata: {"content":"B"}'), "poll 2 joined text should contain both chunks" ); }); test("pooling effect state merge preserves streamChunks across updates", () => { // Simulate the polling useEffect state merge: // setDetailData(prev => ({...prev, ...data, pipelinePayloads: data?.pipelinePayloads || prev?.pipelinePayloads})) let detailData: Record | null = null; // openDetail initial fetch const initialFetch = { pipelinePayloads: { streamChunks: { provider: ['data: {"a":1}\n\n'] }, providerRequest: { model: "gpt-4" }, }, active: true, detailState: "in-flight", }; detailData = mergeDetailData(null, initialFetch); assert.ok( (detailData!.pipelinePayloads as Record).streamChunks, "initial merge should preserve streamChunks" ); // Poll response adds more chunks const pollResponse = { pipelinePayloads: { streamChunks: { provider: ['data: {"a":1}\n\n', 'data: {"b":2}\n\n'], }, providerRequest: { model: "gpt-4" }, }, active: true, detailState: "in-flight", }; detailData = mergeDetailData(detailData, pollResponse); const mergedChunks = (detailData!.pipelinePayloads as Record) .streamChunks as Record; assert.equal(mergedChunks.provider.length, 2, "merged should have 2 chunks"); assert.equal(mergedChunks.provider[1], 'data: {"b":2}\n\n', "merged should include new chunk"); // Simulate: polling response loses pipelinePayloads (null) // This should NOT overwrite the existing streamChunks const nullPayloadResponse = { pipelinePayloads: null, active: true, detailState: "in-flight", }; const afterNullPayload = mergeDetailData(detailData, nullPayloadResponse); assert.ok( (afterNullPayload.pipelinePayloads as Record).streamChunks, "streamChunks should survive null pipelinePayloads update" ); // Simulate: polling response has pipelinePayloads but NO streamChunks // This SHOULD overwrite (|| semantics) — but the production condition is: when // data?.pipelinePayloads is truthy, it replaces prev. If captureStreamChunks was false, // the data won't have streamChunks. This is expected behavior — the frontend doesn't // try to retain old streamChunks when new data explicitly lacks them. const noChunksPayloadResponse = { pipelinePayloads: { providerResponse: { status: 200 }, // no streamChunks key }, active: false, detailState: "ready", }; const afterNoChunks = mergeDetailData(detailData, noChunksPayloadResponse); const payloadAfter = afterNoChunks.pipelinePayloads as Record; assert.equal( payloadAfter.streamChunks, undefined, "streamChunks is lost when new pipelinePayloads lacks it and is truthy" ); // This demonstrates the || semantics: data.pipelinePayloads is truthy ({providerResponse: {...}}) // so it replaces prev.pipelinePayloads even though it lacks streamChunks. // The Event Stream section will not render after this update. // However, in practice this only happens after request completion when the polling // transitions to the persisted artifact which may have truncated streamChunks. // By that point the Event Stream section is no longer needed (streaming is done). }); test("pendingById references are live: push mutates the shared arrays visible to API", () => { usageHistory.clearPendingRequests(); const model = "gpt-4"; const provider = "openai"; const connectionId = "conn-ref-1"; const requestId = usageHistory.trackPendingRequest(model, provider, connectionId, true); // Simulate push() — store initial empty arrays usageHistory.updatePendingRequestStreamChunks(model, provider, connectionId, { provider: [], openai: [], client: [], }); // Get the pending detail reference const detailFromPending = usageHistory.getPendingById().get(requestId); assert.ok(detailFromPending, "detail should be in pendingById"); assert.ok(detailFromPending!.streamChunks, "streamChunks should be set to wrapper object"); // Simulate appendBoundedChunk — pushes to the shared array detailFromPending!.streamChunks!.provider.push('data: {"chunk":"live"}'); // Re-read from pendingById — should see the mutation const detailReRead = usageHistory.getPendingById().get(requestId); assert.equal( detailReRead!.streamChunks!.provider.length, 1, "mutation should be visible through pendingById" ); assert.equal( detailReRead!.streamChunks!.provider[0], 'data: {"chunk":"live"}', "mutated content should be visible through pendingById" ); // Verify via simulated API response const apiResp = buildApiResponseFromPending(requestId); const apiChunks = (apiResp!.pipelinePayloads as Record).streamChunks as Record< string, string[] >; assert.equal(apiChunks.provider.length, 1, "API should see the live mutation"); }); test("no connectionId in request logger does not break anything", async () => { const { createRequestLogger } = await import("../../open-sse/utils/requestLogger.ts"); usageHistory.clearPendingRequests(); usageHistory.trackPendingRequest("gpt-4", "openai", "conn-noop-1", true); // Logger without connectionId/model — push() bails, no crash const logger = await createRequestLogger("openai", "openai", "gpt-4", { enabled: true, captureStreamChunks: true, }); logger.appendProviderChunk('data: {"a":1}'); logger.appendOpenAIChunk('data: {"b":2}'); logger.appendConvertedChunk('data: {"c":3}'); // The pending detail should NOT have streamChunks (no connectionId match) const pending = usageHistory.getPendingRequests(); const detail = pending.details["conn-noop-1"]?.["gpt-4 (openai)"]?.[0]; assert.equal(detail.streamChunks, undefined, "streamChunks should not be set"); // Simulated API should return null for streamChunks const apiResp = buildApiResponseFromPending( // We need the actual ID. trackPendingRequest returns it now. (() => { usageHistory.clearPendingRequests(); const id = usageHistory.trackPendingRequest("gpt-4", "openai", "conn-noop-2", true); return id!; })() ); // This request had no streaming, so streamChunks is null if (apiResp) { const noChunks = computeStreamChunksText( true, apiResp.pipelinePayloads as Record ); assert.equal(noChunks, null, "no streamChunks means no event stream"); } }); test("createRequestLogger and trackPendingRequest with matching model propagate streamChunks", async () => { // The fix: chatCore.ts now passes `model` (not `effectiveModel`) to // createRequestLogger, so the modelKey matches what trackPendingRequest uses. const { createRequestLogger } = await import("../../open-sse/utils/requestLogger.ts"); usageHistory.clearPendingRequests(); const model = "gpt-4"; const provider = "openai"; const connectionId = "conn-match-1"; const requestId = usageHistory.trackPendingRequest(model, provider, connectionId, true, { clientRequest: { messages: [] }, }); // Use matching model (the fix) const logger = await createRequestLogger("openai", "openai", model, { enabled: true, captureStreamChunks: true, model, provider, connectionId, }); logger.appendProviderChunk('data: {"chunk":"hello"}'); logger.appendOpenAIChunk('data: {"choices":[{"delta":{"content":"hi"}}]}'); logger.appendConvertedChunk('data: {"content":"world"}'); const detail = usageHistory.getPendingById().get(requestId); assert.ok(detail, "pending detail should exist"); assert.ok(detail!.streamChunks, "streamChunks should be set"); assert.equal(detail!.streamChunks!.provider.length, 1); assert.equal(detail!.streamChunks!.openai.length, 1); assert.equal(detail!.streamChunks!.client.length, 1); const apiResp = buildApiResponseFromPending(requestId!); assert.ok(apiResp); const apiChunks = (apiResp!.pipelinePayloads as Record).streamChunks as Record< string, string[] >; assert.equal(stripChunkTs(apiChunks?.provider[0]), 'data: {"chunk":"hello"}'); }); test("finalizePendingRequestById completes the exact stream when same model requests overlap", () => { usageHistory.clearPendingRequests(); const model = "gpt-4"; const provider = "openai"; const connectionId = "conn-overlap-1"; const firstId = usageHistory.trackPendingRequest(model, provider, connectionId, true, { clientRequest: { messages: [{ role: "user", content: "first" }] }, }); const secondId = usageHistory.trackPendingRequest(model, provider, connectionId, true, { clientRequest: { messages: [{ role: "user", content: "second" }] }, }); const completed = usageHistory.finalizePendingRequestById(firstId, { providerResponse: { id: "first-response" }, clientResponse: { id: "first-response" }, }); assert.equal(completed, true); assert.ok(usageHistory.getCompletedDetails().has(firstId)); assert.equal(usageHistory.getCompletedDetails().has(secondId), false); assert.equal(usageHistory.getPendingById().has(firstId), false); assert.equal(usageHistory.getPendingById().has(secondId), true); const pending = usageHistory.getPendingRequests(); const details = pending.details[connectionId]?.[`${model} (${provider})`] ?? []; assert.equal(details.length, 1); assert.equal(details[0].id, secondId); assert.equal(pending.byModel[`${model} (${provider})`], 1); assert.equal(pending.byAccount[connectionId]?.[`${model} (${provider})`], 1); }); test("completedDetails cache evicts oldest entries when bounded", () => { usageHistory.clearPendingRequests(); const model = "gpt-4"; const provider = "openai"; const connectionId = "conn-completed-bound"; const ids: string[] = []; for (let i = 0; i < 260; i++) { const id = usageHistory.trackPendingRequest(model, provider, connectionId, true); ids.push(id!); const completed = usageHistory.finalizePendingRequestById(id, { clientResponse: { choices: [{ message: { content: `done ${i}` } }] }, }); assert.equal(completed, true); } assert.ok( usageHistory.getCompletedDetails().size <= 256, "completedDetails should remain bounded" ); assert.equal(usageHistory.getCompletedDetails().has(ids[0]), false); assert.equal(usageHistory.getCompletedDetails().has(ids[ids.length - 1]), true); }); test("streamChunks in completedDetails survives beyond the logs polling window", async () => { usageHistory.clearPendingRequests(); const model = "gpt-4"; const provider = "openai"; const connectionId = "conn-ttl-1"; const requestId = usageHistory.trackPendingRequest(model, provider, connectionId, true, { clientRequest: { messages: [] }, }); // Simulate streaming then completion usageHistory.updatePendingRequestStreamChunks(model, provider, connectionId, { provider: ['data: {"chunk":"final"}\n\n'], openai: [], client: [], }); usageHistory.finalizeMostRecentPendingRequest(model, provider, connectionId, { status: 200, model, provider, }); // Should be in completedDetails assert.ok( usageHistory.getCompletedDetails().has(requestId), "should be in completedDetails after finalization" ); const completedResp = buildApiResponseFromCompleted(requestId); assert.ok(completedResp, "API response should be available from completedDetails"); assert.ok( (completedResp!.pipelinePayloads as Record).streamChunks, "streamChunks should be in completedDetails" ); // Save to DB as the normal durable path, but keep the completedDetails cache // long enough that a slow Logs-page poll does not see the row disappear // between pending removal and DB/detail refresh. await callLogs.saveCallLog({ id: requestId, method: "POST", path: "/v1/chat/completions", status: 200, model, requestedModel: model, provider, connectionId, duration: 3000, tokens: { in: 5, out: 10 }, requestBody: {}, responseBody: {}, error: null, sourceFormat: "openai", targetFormat: "openai", comboName: null, comboStepId: null, comboExecutionKey: null, tokensCompressed: null, cacheSource: "upstream", apiKeyId: null, apiKeyName: null, noLog: false, pipelinePayloads: completedResp!.pipelinePayloads as Record, }); // Verify DB has it const dbEntry1 = await callLogs.getCallLogById(requestId); assert.ok(dbEntry1, "DB should have the entry after saveCallLog"); // This used to expire after 5 seconds. Keep it visible beyond that window so // a slow client poll can still resolve the completed row/details. await new Promise((r) => setTimeout(r, 5100)); assert.ok( usageHistory.getCompletedDetails().has(requestId), "completedDetails should still be available after a 5-second polling gap" ); const dbEntry2 = await callLogs.getCallLogById(requestId); assert.ok(dbEntry2, "DB should still have the entry after the polling gap"); assert.equal( (dbEntry2 as Record).id, requestId, "DB entry ID should match the original request ID" ); });