From a2a0e932c871a02f1c72fcd8236e2f5c4b3fb358 Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Mon, 20 Jul 2026 15:41:31 -0700 Subject: [PATCH] fix: preserve transcript cursors across embedded turns (#111949) * fix(agents): preserve transcript generation on flush (clawdbot-85c) * fix(agents): preserve SQLite transcript parents (clawdbot-85c.5) * test(agents): prove cursor continuity through Gateway (clawdbot-85c.7) --- src/agents/cli-runner.ts | 10 +- .../run/attempt-transcript-helpers.ts | 6 +- ...attempt-transcript-persistence.e2e.test.ts | 119 ++++++++ ...mpt.spawn-workspace.context-engine.test.ts | 2 +- .../attempt.spawn-workspace.test-support.ts | 6 +- src/agents/sessions/session-manager-core.ts | 9 + .../sessions/session-manager-persistence.ts | 17 +- src/agents/sessions/session-manager.test.ts | 66 ++++ src/agents/sessions/session-manager.ts | 5 + test/embedded-transcript-cursor.e2e.test.ts | 289 ++++++++++++++++++ 10 files changed, 509 insertions(+), 20 deletions(-) create mode 100644 src/agents/embedded-agent-runner/run/attempt-transcript-persistence.e2e.test.ts create mode 100644 test/embedded-transcript-cursor.e2e.test.ts diff --git a/src/agents/cli-runner.ts b/src/agents/cli-runner.ts index 40c8cfd61013..97b5e494527a 100644 --- a/src/agents/cli-runner.ts +++ b/src/agents/cli-runner.ts @@ -184,14 +184,6 @@ export async function isCliBindingFlushed( return false; } -function flushSessionManagerTranscript(sessionManager: SessionManager): void { - ( - sessionManager as unknown as { - replacePersistedTranscript?: () => void; - } - ).replacePersistedTranscript?.(); -} - async function assertSuccessfulCliRuntimeBindingCurrent( context: PreparedCliRunContext, ): Promise { @@ -887,7 +879,7 @@ export async function runPreparedCliAgent( sessionManager.appendMessage( redactedUserMessage as Parameters[0], ); - flushSessionManagerTranscript(sessionManager); + sessionManager.flushPendingPersistence(); } catch (err) { log.warn( `before_agent_run block: failed to persist redacted CLI user message: ${formatErrorMessage( diff --git a/src/agents/embedded-agent-runner/run/attempt-transcript-helpers.ts b/src/agents/embedded-agent-runner/run/attempt-transcript-helpers.ts index 8bbc64d67221..0791527c6946 100644 --- a/src/agents/embedded-agent-runner/run/attempt-transcript-helpers.ts +++ b/src/agents/embedded-agent-runner/run/attempt-transcript-helpers.ts @@ -22,11 +22,7 @@ import type { EmbeddedRunAttemptParams } from "./types.js"; type AttemptSessionManager = ReturnType; export function flushSessionManagerTranscript(sessionManager: AttemptSessionManager): void { - ( - sessionManager as unknown as { - replacePersistedTranscript?: () => void; - } - ).replacePersistedTranscript?.(); + sessionManager.flushPendingPersistence(); } export function repairAttemptToolUseResultPairing( diff --git a/src/agents/embedded-agent-runner/run/attempt-transcript-persistence.e2e.test.ts b/src/agents/embedded-agent-runner/run/attempt-transcript-persistence.e2e.test.ts new file mode 100644 index 000000000000..5be465dcb761 --- /dev/null +++ b/src/agents/embedded-agent-runner/run/attempt-transcript-persistence.e2e.test.ts @@ -0,0 +1,119 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { readSessionTranscriptRawDelta } from "openclaw/plugin-sdk/session-transcript-runtime"; +import { afterEach, describe, expect, it } from "vitest"; +import { + appendTranscriptMessage, + upsertSessionEntry, +} from "../../../config/sessions/session-accessor.js"; +import { formatSqliteSessionFileMarker } from "../../../config/sessions/sqlite-marker.js"; +import { SessionManager } from "../../sessions/session-manager.js"; +import { flushSessionManagerTranscript } from "./attempt-transcript-helpers.js"; + +const tempPaths: string[] = []; + +async function makeTempDir(): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-attempt-transcript-")); + tempPaths.push(dir); + return dir; +} + +function buildAssistantMessage(text: string) { + return { + role: "assistant" as const, + content: [{ type: "text" as const, text }], + api: "openai-responses" as const, + provider: "openai", + model: "test-model", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + stopReason: "stop" as const, + timestamp: Date.now(), + }; +} + +describe("embedded attempt transcript persistence", () => { + afterEach(async () => { + await Promise.all( + tempPaths.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })), + ); + }); + + it("resumes a raw cursor after append-only attempt settlement", async () => { + const dir = await makeTempDir(); + const storePath = path.join(dir, "sessions.json"); + const target = { + agentId: "main", + sessionId: "embedded-generation", + sessionKey: "agent:main:embedded-generation", + storePath, + }; + const marker = formatSqliteSessionFileMarker(target); + await upsertSessionEntry(target, { + sessionFile: marker, + sessionId: target.sessionId, + updatedAt: 1, + }); + await appendTranscriptMessage(target, { + cwd: dir, + eventId: "first-user", + message: { role: "user", content: "first turn" }, + now: 1, + }); + + const bootstrap = await readSessionTranscriptRawDelta({ + ...target, + maxBytes: 100_000, + maxEvents: 100, + }); + expect(bootstrap.kind).toBe("page"); + if (bootstrap.kind !== "page") { + throw new Error(`expected bootstrap page, got ${bootstrap.kind}`); + } + + const sessionManager = SessionManager.open(marker, dir, dir); + sessionManager.appendMessage({ + role: "user", + content: "second turn", + timestamp: Date.now(), + }); + sessionManager.appendMessage(buildAssistantMessage("second answer")); + + // Production settlement invokes this barrier immediately before afterTurn. + flushSessionManagerTranscript(sessionManager); + + const resumed = await readSessionTranscriptRawDelta({ + ...target, + cursor: bootstrap.cursor, + maxBytes: 100_000, + maxEvents: 100, + }); + expect(resumed.kind).toBe("page"); + if (resumed.kind !== "page") { + throw new Error(`expected append page, got ${resumed.kind}`); + } + expect( + resumed.events + .map((row) => row.event) + .filter((event): event is { message: { content: unknown }; type: "message" } => + Boolean( + event && typeof event === "object" && "type" in event && event.type === "message", + ), + ) + .map((event) => event.message.content), + ).toEqual(["second turn", [{ type: "text", text: "second answer" }]]); + }); +}); diff --git a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts index 73f158d62db0..8969d1a93cee 100644 --- a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts @@ -2803,7 +2803,7 @@ describe("runEmbeddedAttempt context engine sessionKey forwarding", () => { const afterTurn = vi.fn(async () => { events.push("afterTurn"); }); - hoisted.sessionManager.replacePersistedTranscript.mockImplementation(() => { + hoisted.sessionManager.flushPendingPersistence.mockImplementation(() => { events.push("flush"); }); diff --git a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts index 8e645d88ca2f..c8d47012b941 100644 --- a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts +++ b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts @@ -74,7 +74,7 @@ type SessionManagerMocks = { appendCustomEntry: UnknownMock; appendSessionInfo: UnknownMock; appendLabelChange: UnknownMock; - replacePersistedTranscript: UnknownMock; + flushPendingPersistence: UnknownMock; flushPendingToolResults: UnknownMock; clearPendingToolResults: UnknownMock; clearNextUserMessagePersistenceSuppression: UnknownMock; @@ -242,7 +242,7 @@ const hoisted = vi.hoisted((): AttemptSpawnWorkspaceHoisted => { appendCustomEntry: vi.fn(), appendSessionInfo: vi.fn(), appendLabelChange: vi.fn(), - replacePersistedTranscript: vi.fn(), + flushPendingPersistence: vi.fn(), flushPendingToolResults: vi.fn(), clearPendingToolResults: vi.fn(), clearNextUserMessagePersistenceSuppression: vi.fn(), @@ -1126,7 +1126,7 @@ export function resetEmbeddedAttemptHarness( hoisted.sessionManager.appendCustomEntry.mockReset(); hoisted.sessionManager.appendSessionInfo.mockReset(); hoisted.sessionManager.appendLabelChange.mockReset(); - hoisted.sessionManager.replacePersistedTranscript.mockReset(); + hoisted.sessionManager.flushPendingPersistence.mockReset(); if (params.subscribeImpl) { hoisted.subscribeEmbeddedAgentSessionMock.mockImplementation(params.subscribeImpl); } diff --git a/src/agents/sessions/session-manager-core.ts b/src/agents/sessions/session-manager-core.ts index faed9b191724..df86bc976592 100644 --- a/src/agents/sessions/session-manager-core.ts +++ b/src/agents/sessions/session-manager-core.ts @@ -582,6 +582,15 @@ export class SessionManagerCore { } } + /** Makes pending append-oriented persistence durable without replacing SQLite transcripts. */ + protected flushPendingPersistence(): void { + if (!this.shouldPersist || this.sqlitePersistence || this.flushed || !this.sessionFile) { + return; + } + this.replacePersistedTranscript(); + this.flushed = true; + } + isPersisted(): boolean { return this.shouldPersist; } diff --git a/src/agents/sessions/session-manager-persistence.ts b/src/agents/sessions/session-manager-persistence.ts index af6143e2a59b..39fd8c2e06b4 100644 --- a/src/agents/sessions/session-manager-persistence.ts +++ b/src/agents/sessions/session-manager-persistence.ts @@ -227,7 +227,7 @@ export class SessionManagerPersistence extends SessionManagerCore { appendTranscriptEventSync(scope, entry); return; } - const result = appendTranscriptMessageSync(scope, { + const appendOptions = { cwd: this.cwd, eventId: entry.id, ...(options?.config ? { config: options.config } : {}), @@ -235,7 +235,20 @@ export class SessionManagerPersistence extends SessionManagerCore { message: entry.message, now: Date.parse(entry.timestamp), parentId: entry.parentId, - }); + } satisfies Parameters[1]; + let result = appendTranscriptMessageSync(scope, appendOptions); + if (result && !result.appended && result.messageId !== entry.id) { + // SessionManager has already adopted this event ID as the next parent. A + // pre-persisted user turn may share its idempotency key, but dropping the + // canonical node would leave every later descendant dangling in SQLite. + result = appendTranscriptMessageSync(scope, { + ...appendOptions, + idempotencyLookup: "caller-checked", + }); + } + if (result && result.messageId !== entry.id) { + throw new Error(`Session transcript parent entry was not persisted: ${entry.id}`); + } if ( options?.idempotencyLookup === "caller-checked" && (!result?.appended || result.messageId !== entry.id) diff --git a/src/agents/sessions/session-manager.test.ts b/src/agents/sessions/session-manager.test.ts index 63f5eb014af8..fa72b007ce0a 100644 --- a/src/agents/sessions/session-manager.test.ts +++ b/src/agents/sessions/session-manager.test.ts @@ -9,6 +9,7 @@ import { appendTranscriptMessage, loadSessionEntry, loadTranscriptEvents, + readTranscriptRawDelta, upsertSessionEntry, } from "../../config/sessions/session-accessor.js"; import { formatSqliteSessionFileMarker } from "../../config/sessions/sqlite-marker.js"; @@ -45,6 +46,24 @@ describe("SessionManager.open", () => { ); }); + it("flushes a pending initial file transcript before later appends", async () => { + const dir = await makeTempDir(); + const sessionFile = path.join(dir, "pending-session.jsonl"); + const sessionManager = SessionManager.open(sessionFile, dir, dir); + + sessionManager.appendMessage({ role: "user", content: "question", timestamp: Date.now() }); + await expect(fs.stat(sessionFile)).rejects.toMatchObject({ code: "ENOENT" }); + + sessionManager.flushPendingPersistence(); + sessionManager.appendMessage(buildAssistantMessage("answer")); + + expect( + loadEntriesFromFile(sessionFile) + .filter((entry) => entry.type === "message") + .map((entry) => ("content" in entry.message ? entry.message.content : undefined)), + ).toEqual(["question", [{ type: "text", text: "answer" }]]); + }); + it("opens SQLite markers without creating marker-named files and persists assistant replies", async () => { const dir = await makeTempDir(); const storePath = path.join(dir, "sessions.json"); @@ -151,6 +170,53 @@ describe("SessionManager.open", () => { ); }); + it("persists a deduped runtime user entry before its SQLite descendants", async () => { + const dir = await makeTempDir(); + const storePath = path.join(dir, "sessions.json"); + const sessionId = "sqlite-runtime-user-parent"; + const sessionKey = "agent:main:dashboard:sqlite-runtime-user-parent"; + const scope = { agentId: "main", sessionId, sessionKey, storePath }; + const marker = formatSqliteSessionFileMarker(scope); + const userMessage = { + role: "user" as const, + content: "question", + idempotencyKey: "runtime-user-parent:user", + timestamp: 1, + }; + await upsertSessionEntry(scope, { sessionFile: marker, sessionId, updatedAt: 1 }); + await appendTranscriptMessage(scope, { + cwd: dir, + eventId: "pre-persisted-user", + message: userMessage, + now: 1, + }); + const bootstrap = readTranscriptRawDelta(scope, { maxBytes: 10_000, maxEvents: 100 }); + expect(bootstrap.kind).toBe("page"); + if (bootstrap.kind !== "page") { + throw new Error(`expected bootstrap page, got ${bootstrap.kind}`); + } + + const sessionManager = SessionManager.open(marker, dir, dir); + const runtimeUserId = sessionManager.appendMessage(userMessage); + const assistantId = sessionManager.appendMessage(buildAssistantMessage("answer")); + const resumed = readTranscriptRawDelta(scope, { + cursor: bootstrap.cursor, + maxBytes: 10_000, + maxEvents: 100, + }); + + expect(resumed.kind).toBe("page"); + if (resumed.kind !== "page") { + throw new Error(`expected append page, got ${resumed.kind}`); + } + expect(resumed.events.map((row) => (row.event as { id?: string }).id)).toEqual([ + runtimeUserId, + assistantId, + ]); + const assistantEvent = resumed.events.at(1)?.event as { parentId?: string } | undefined; + expect(assistantEvent?.parentId).toBe(runtimeUserId); + }); + it("preserves root-to-leaf ordering across session branches", () => { const entries = [ { diff --git a/src/agents/sessions/session-manager.ts b/src/agents/sessions/session-manager.ts index 4950352da6f6..e23e1f008153 100644 --- a/src/agents/sessions/session-manager.ts +++ b/src/agents/sessions/session-manager.ts @@ -96,6 +96,11 @@ export class SessionManager extends SessionManagerBranching { super.clearPreservedOpaqueFileEntries(); } + /** Makes pending append-oriented persistence durable without rewriting committed entries. */ + override flushPendingPersistence(): void { + super.flushPendingPersistence(); + } + override isPersisted(): boolean { return super.isPersisted(); } diff --git a/test/embedded-transcript-cursor.e2e.test.ts b/test/embedded-transcript-cursor.e2e.test.ts new file mode 100644 index 000000000000..ddc4f56e9edb --- /dev/null +++ b/test/embedded-transcript-cursor.e2e.test.ts @@ -0,0 +1,289 @@ +// E2E: ordinary embedded Gateway turns preserve raw transcript cursor continuity. +import { randomUUID } from "node:crypto"; +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import path from "node:path"; +import { readSessionTranscriptRawDelta } from "openclaw/plugin-sdk/session-transcript-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { loadSessionEntry } from "../src/config/sessions/session-accessor.js"; +import type { OpenClawConfig } from "../src/config/types.openclaw.js"; +import { connectGatewayClient, disconnectGatewayClient } from "../src/gateway/test-helpers.e2e.js"; +import { + createOpenClawTestInstance, + type OpenClawTestInstance, +} from "./helpers/openclaw-test-instance.js"; + +const TEST_TIMEOUT_MS = 180_000; +const MODEL_REF = "cursor-settlement/cursor-settlement"; +const SESSION_KEY = "agent:main:cursor-settlement-e2e"; +const GATEWAY_TOKEN_OPTION = "token"; + +type MockModelServer = { + baseUrl: string; + close: () => Promise; +}; + +const instances: OpenClawTestInstance[] = []; +const modelServers: MockModelServer[] = []; + +afterEach(async () => { + await Promise.allSettled(instances.splice(0).map((instance) => instance.cleanup())); + await Promise.allSettled(modelServers.splice(0).map((server) => server.close())); +}); + +describe("embedded transcript cursor settlement", () => { + it( + "resumes a public raw cursor after a real append-only Gateway turn", + { timeout: TEST_TIMEOUT_MS }, + async () => { + const modelServer = await startMockModelServer(); + modelServers.push(modelServer); + const instance = await createOpenClawTestInstance({ + name: "embedded-transcript-cursor", + config: createTestConfig(modelServer.baseUrl), + env: { OPENCLAW_SKIP_PROVIDERS: undefined }, + }); + instances.push(instance); + await instance.startGateway(); + + const client = await connectGatewayClient({ + url: instance.url, + [GATEWAY_TOKEN_OPTION]: instance.gatewayToken, + role: "operator", + scopes: ["operator.admin", "operator.read", "operator.write"], + }); + try { + await runAgentTurn(client, instance, "first cursor turn"); + const storePath = path.join(instance.state.sessionsDir("main"), "sessions.json"); + const sessionId = await waitForSessionId(storePath); + const target = { agentId: "main", sessionId, sessionKey: SESSION_KEY, storePath }; + const bootstrap = await readSessionTranscriptRawDelta({ + ...target, + maxBytes: 1_000_000, + maxEvents: 100, + }); + expect(bootstrap.kind, instance.logs()).toBe("page"); + if (bootstrap.kind !== "page") { + throw new Error(`expected bootstrap page, got ${bootstrap.kind}`); + } + expect(bootstrap.hasMore).toBe(false); + + await runAgentTurn(client, instance, "second cursor turn"); + const resumed = await readSessionTranscriptRawDelta({ + ...target, + cursor: bootstrap.cursor, + maxBytes: 1_000_000, + maxEvents: 100, + }); + + expect(resumed.kind, `${JSON.stringify(resumed)}\n${instance.logs()}`).toBe("page"); + if (resumed.kind !== "page") { + throw new Error(`expected resumed page, got ${resumed.kind}`); + } + expect(resumed.hasMore).toBe(false); + const messageTexts = resumed.events.flatMap((row) => readMessageText(row.event)); + expect(messageTexts).toContain("second cursor turn"); + expect(messageTexts).toContain("cursor settlement response 2"); + expect(messageTexts).not.toContain("first cursor turn"); + expect(messageTexts).not.toContain("cursor settlement response 1"); + } finally { + await disconnectGatewayClient(client); + } + }, + ); +}); + +function createTestConfig(baseUrl: string): OpenClawConfig { + return { + plugins: { slots: { memory: "none" } }, + agents: { + defaults: { + heartbeat: { every: "0m" }, + model: { primary: MODEL_REF }, + models: { [MODEL_REF]: { agentRuntime: { id: "openclaw" } } }, + skipBootstrap: true, + skills: [], + }, + }, + tools: { profile: "minimal" }, + models: { + mode: "replace", + providers: { + "cursor-settlement": { + baseUrl: `${baseUrl}/v1`, + apiKey: "test-token-placeholder", + api: "openai-responses", + request: { allowPrivateNetwork: true }, + models: [ + { + id: "cursor-settlement", + name: "cursor-settlement", + api: "openai-responses", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128_000, + maxTokens: 4_096, + }, + ], + }, + }, + }, + }; +} + +async function runAgentTurn( + client: Awaited>, + instance: OpenClawTestInstance, + message: string, +): Promise { + const requestedRunId = randomUUID(); + const started = await client.request<{ runId?: string; status?: string }>("agent", { + sessionKey: SESSION_KEY, + message, + deliver: false, + idempotencyKey: requestedRunId, + }); + expect(started.status).toBe("accepted"); + const completed = await client.request<{ error?: unknown; status?: string }>( + "agent.wait", + { runId: started.runId ?? requestedRunId, timeoutMs: 120_000 }, + { timeoutMs: 125_000 }, + ); + expect(completed.status, `${JSON.stringify(completed)}\n${instance.logs()}`).toBe("ok"); +} + +async function waitForSessionId(storePath: string): Promise { + let sessionId: string | undefined; + await vi.waitFor( + () => { + sessionId = loadSessionEntry({ + agentId: "main", + readConsistency: "latest", + sessionKey: SESSION_KEY, + storePath, + })?.sessionId; + expect(sessionId).toBeTruthy(); + }, + { interval: 20, timeout: 30_000 }, + ); + if (!sessionId) { + throw new Error(`session id was not persisted for ${SESSION_KEY}`); + } + return sessionId; +} + +function readMessageText(event: unknown): string[] { + if (!event || typeof event !== "object" || (event as { type?: unknown }).type !== "message") { + return []; + } + const message = (event as { message?: { content?: unknown } }).message; + if (typeof message?.content === "string") { + return [message.content]; + } + if (!Array.isArray(message?.content)) { + return []; + } + return message.content.flatMap((part) => + part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string" + ? [(part as { text: string }).text] + : [], + ); +} + +async function startMockModelServer(): Promise { + let responseCount = 0; + const server = createServer((request, response) => { + void (async () => { + const url = new URL(request.url ?? "/", "http://127.0.0.1"); + if (request.method === "GET" && url.pathname === "/v1/models") { + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ data: [{ id: "cursor-settlement", object: "model" }] })); + return; + } + if (request.method !== "POST" || url.pathname !== "/v1/responses") { + response.writeHead(404).end(); + return; + } + await drainRequest(request); + responseCount += 1; + writeModelResponse(response, responseCount); + })().catch((error) => { + response.writeHead(500, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: { message: String(error) } })); + }); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("cursor settlement model server did not bind"); + } + return { + baseUrl: `http://127.0.0.1:${address.port}`, + close: async () => { + server.closeAllConnections(); + await new Promise((resolve) => { + server.close(() => resolve()); + }); + }, + }; +} + +async function drainRequest(request: IncomingMessage): Promise { + for await (const chunk of request) { + // Consume the body before replying so the embedded transport completes cleanly. + void chunk; + } +} + +function writeModelResponse(response: ServerResponse, sequence: number): void { + const text = `cursor settlement response ${sequence}`; + const message = { + type: "message", + id: `cursor-settlement-message-${sequence}`, + role: "assistant", + status: "completed", + content: [{ type: "output_text", text, annotations: [] }], + }; + const events = [ + { + type: "response.output_item.added", + output_index: 0, + item: { ...message, status: "in_progress", content: [] }, + }, + { + type: "response.output_text.delta", + item_id: message.id, + output_index: 0, + content_index: 0, + delta: text, + }, + { + type: "response.output_text.done", + item_id: message.id, + output_index: 0, + content_index: 0, + text, + }, + { type: "response.output_item.done", output_index: 0, item: message }, + { + type: "response.completed", + response: { + id: `cursor-settlement-response-${sequence}`, + status: "completed", + output: [message], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }, + }, + ]; + response.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + response.end( + `${events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join("")}data: [DONE]\n\n`, + ); +}