From 5ec4d9f82a6bf05edb3bf33558f10f84ec6d317e Mon Sep 17 00:00:00 2001 From: Jacqueline Henriksen Date: Wed, 5 Aug 2026 15:30:30 -0700 Subject: [PATCH] fix(agents): seed fresh session identity before transcript append (#119473) * fix(agents): seed fresh session identity before transcript append * test(agents): prove fresh session persistence through gateway * fix(agents): seed session identity on first transcript write * fix(agents): atomically seed fresh session identity --- .../sessions/session-manager-persistence.ts | 9 + src/agents/sessions/session-manager.test.ts | 75 +++++++ src/config/sessions/session-accessor.entry.ts | 2 + .../sessions/session-accessor.sqlite-entry.ts | 29 +++ .../sessions/session-accessor.sqlite.ts | 1 + src/config/sessions/session-accessor.ts | 1 + test/pr119473-real-runtime-proof.e2e.test.ts | 197 ++++++++++++++++++ 7 files changed, 314 insertions(+) create mode 100644 test/pr119473-real-runtime-proof.e2e.test.ts diff --git a/src/agents/sessions/session-manager-persistence.ts b/src/agents/sessions/session-manager-persistence.ts index fcb279ec9ab1..d216e2c259b0 100644 --- a/src/agents/sessions/session-manager-persistence.ts +++ b/src/agents/sessions/session-manager-persistence.ts @@ -1,6 +1,7 @@ import { appendTranscriptEventSync, appendTranscriptMessageSync, + ensureSessionEntrySync, } from "../../config/sessions/session-accessor.js"; import { isSessionTranscriptSideAppendEntry } from "../../config/sessions/transcript-tree.js"; import { @@ -150,6 +151,14 @@ export class SessionManagerPersistence extends SessionManagerCore { } const scope = this.persistenceTarget; if (this.persistenceHeaderPending) { + if ( + !ensureSessionEntrySync(scope, { + sessionId: scope.sessionId, + updatedAt: Date.now(), + }) + ) { + throw new Error("Session transcript header was not persisted"); + } const header = this.fileEntries[0]; if (!header || header.type !== "session" || !appendTranscriptEventSync(scope, header)) { throw new Error("Session transcript header was not persisted"); diff --git a/src/agents/sessions/session-manager.test.ts b/src/agents/sessions/session-manager.test.ts index cc6eb380cf84..ff713d5a2b3c 100644 --- a/src/agents/sessions/session-manager.test.ts +++ b/src/agents/sessions/session-manager.test.ts @@ -306,6 +306,81 @@ describe("SessionManager.open", () => { expect(() => SessionManager.open(scope, dir)).not.toThrow(); }); + it("persists a fresh SQLite session header and first message", async () => { + const dir = await makeTempDir(); + const scope = { + agentId: "main", + sessionId: "sqlite-fresh-session", + sessionKey: "agent:main:sqlite-fresh-session", + storePath: path.join(dir, "sessions.json"), + }; + + expect(loadSessionEntry(scope)).toBeUndefined(); + const manager = SessionManager.open(scope, dir); + expect(loadSessionEntry(scope)).toBeUndefined(); + const messageId = manager.appendMessage({ + role: "user", + content: "first message", + timestamp: 1, + }); + + await expect(loadTranscriptEvents(scope)).resolves.toEqual([ + expect.objectContaining({ + id: scope.sessionId, + type: "session", + version: CURRENT_SESSION_VERSION, + }), + expect.objectContaining({ + id: messageId, + message: expect.objectContaining({ content: "first message", role: "user" }), + type: "message", + }), + ]); + expect(loadSessionEntry(scope)).toMatchObject({ sessionId: scope.sessionId }); + }); + + it("does not rewrite an existing session row when opening an empty transcript", async () => { + const dir = await makeTempDir(); + const scope = { + agentId: "main", + sessionId: "sqlite-empty-existing-row-target", + sessionKey: "agent:main:sqlite-empty-existing-row", + storePath: path.join(dir, "sessions.json"), + }; + await upsertSessionEntry(scope, { + sessionId: "sqlite-existing-row", + updatedAt: 123, + label: "preserved", + }); + const before = loadSessionEntry(scope); + + SessionManager.open(scope, dir); + + expect(loadSessionEntry(scope)).toEqual(before); + }); + + it("does not overwrite a rebound session row when the first append seeds its header", async () => { + const dir = await makeTempDir(); + const scope = { + agentId: "main", + sessionId: "sqlite-stale-appender", + sessionKey: "agent:main:sqlite-rebound-before-header", + storePath: path.join(dir, "sessions.json"), + }; + await upsertSessionEntry(scope, { + sessionId: "sqlite-current-owner", + updatedAt: 456, + label: "preserved", + }); + const before = loadSessionEntry(scope); + const manager = SessionManager.open(scope, dir); + + expect(() => + manager.appendMessage({ role: "user", content: "stale message", timestamp: 1 }), + ).toThrow("Session transcript header was not persisted"); + expect(loadSessionEntry(scope)).toEqual(before); + }); + it("rejects invalid entries before mutating in-memory state", () => { const manager = SessionManager.inMemory("/tmp"); const entriesBefore = manager.getEntries(); diff --git a/src/config/sessions/session-accessor.entry.ts b/src/config/sessions/session-accessor.entry.ts index dd463fb07805..b2df51e98050 100644 --- a/src/config/sessions/session-accessor.entry.ts +++ b/src/config/sessions/session-accessor.entry.ts @@ -12,6 +12,7 @@ import { clearPluginOwnedSessionState } from "./plugin-host-cleanup.js"; import { countSqliteSessionEntryRowsReadOnly as countSessionEntryRowsReadOnly, copySqliteSessionOwnedStateForCanonicalRepair as copySessionOwnedStateForCanonicalRepair, + ensureSqliteSessionEntrySync, hasSqliteSessionEntriesByStatusReadOnly as hasSessionEntriesByStatusReadOnly, listSqliteSessionGenerationIdsForCanonicalRepair as listSessionGenerationIdsForCanonicalRepair, listSqliteSessionChildEntriesReadOnly as listSessionChildEntriesReadOnly, @@ -62,6 +63,7 @@ export { clearPluginOwnedSessionState }; export { countSessionEntryRowsReadOnly, copySessionOwnedStateForCanonicalRepair, + ensureSqliteSessionEntrySync, hasSessionEntriesByStatusReadOnly, listSessionGenerationIdsForCanonicalRepair, listSessionChildEntriesReadOnly, diff --git a/src/config/sessions/session-accessor.sqlite-entry.ts b/src/config/sessions/session-accessor.sqlite-entry.ts index df43dcc826e2..7c91f4d8ec5d 100644 --- a/src/config/sessions/session-accessor.sqlite-entry.ts +++ b/src/config/sessions/session-accessor.sqlite-entry.ts @@ -448,6 +448,35 @@ export function replaceSqliteSessionEntrySync( emitCommittedSessionIdentityDiff(previous, current); } +/** Creates a missing session identity without replacing a concurrently owned row. */ +export function ensureSqliteSessionEntrySync( + scope: SessionAccessScope, + entry: SessionEntry, +): boolean { + const resolved = resolveSqliteScope(scope); + assertCanonicalSessionWriteScope(resolved); + let owned = false; + let previous = new Map(); + let current = new Map(); + runOpenClawAgentWriteTransaction((database) => { + const identityKeys = collectSessionEntryLookupKeys(database, resolved.sessionKey); + previous = readSqliteSessionIdentitySnapshot(database, identityKeys); + const existing = readSessionEntryRow(database, resolved.sessionKey)?.entry; + if (existing) { + owned = existing.sessionId === entry.sessionId; + current = previous; + return; + } + writeSessionEntry(database, resolved.sessionKey, entry); + current = readSqliteSessionIdentitySnapshot(database, identityKeys); + owned = current.get(resolved.sessionKey)?.sessionId === entry.sessionId; + }, toDatabaseOptions(resolved)); + if (current.size !== previous.size || owned) { + emitCommittedSessionIdentityDiff(previous, current); + } + return owned; +} + /** Patches one entry in the additive SQLite session store. */ export async function patchSqliteSessionEntry( scope: SessionAccessScope, diff --git a/src/config/sessions/session-accessor.sqlite.ts b/src/config/sessions/session-accessor.sqlite.ts index 14f9038da6d2..861c12943153 100644 --- a/src/config/sessions/session-accessor.sqlite.ts +++ b/src/config/sessions/session-accessor.sqlite.ts @@ -1,6 +1,7 @@ // Stable SQLite accessor surface. Domain owners live in the focused modules below. export { countSqliteSessionEntryRowsReadOnly, + ensureSqliteSessionEntrySync, hasSqliteSessionEntriesByStatusReadOnly, listSqliteSessionEntries, listSqliteSessionChildEntriesReadOnly, diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index 98ef3ac8cc21..1addd041235a 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -120,6 +120,7 @@ export type { } from "./session-accessor.entry-mutation.js"; export { countSessionEntryRowsReadOnly, + ensureSqliteSessionEntrySync as ensureSessionEntrySync, copySessionOwnedStateForCanonicalRepair, hasSessionEntriesByStatusReadOnly, listSessionGenerationIdsForCanonicalRepair, diff --git a/test/pr119473-real-runtime-proof.e2e.test.ts b/test/pr119473-real-runtime-proof.e2e.test.ts new file mode 100644 index 000000000000..945aa1a74696 --- /dev/null +++ b/test/pr119473-real-runtime-proof.e2e.test.ts @@ -0,0 +1,197 @@ +import fs from "node:fs/promises"; +import { createServer } from "node:http"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { clearConfigCache, clearRuntimeConfigSnapshot } from "../src/config/config.js"; +import { clearSessionStoreCacheForTest } from "../src/config/sessions/store-writer-state.js"; +import { + disconnectGatewayClient, + startGatewayWithClient, +} from "../src/gateway/test-helpers.e2e.js"; +import { buildMockOpenAiResponsesProvider } from "../src/gateway/test-openai-responses-model.js"; +import { captureEnv, setTestEnvValue } from "../src/test-utils/env.js"; + +const envKeys = [ + "HOME", + "OPENCLAW_STATE_DIR", + "OPENCLAW_CONFIG_PATH", + "OPENCLAW_GATEWAY_TOKEN", + "OPENCLAW_SKIP_CHANNELS", + "OPENCLAW_SKIP_GMAIL_WATCHER", + "OPENCLAW_SKIP_CRON", + "OPENCLAW_SKIP_CANVAS_HOST", + "OPENCLAW_SKIP_BROWSER_CONTROL_SERVER", + "OPENCLAW_SKIP_PROVIDERS", + "OPENCLAW_BUNDLED_PLUGINS_DIR", + "OPENCLAW_DISABLE_BUNDLED_PLUGINS", +] as const; + +describe("PR #119473 real gateway proof", () => { + let tempHome: string | undefined; + + afterEach(async () => { + if (tempHome) { + await fs.rm(tempHome, { recursive: true, force: true }); + tempHome = undefined; + } + }); + + it( + "persists the first turn of a fresh SQLite-backed session across gateway restart", + { timeout: 90_000 }, + async () => { + const envSnapshot = captureEnv([...envKeys]); + let providerServer: ReturnType | undefined; + let first: Awaited> | undefined; + let second: Awaited> | undefined; + + try { + tempHome = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pr119473-proof-")); + const stateDir = path.join(tempHome, ".openclaw"); + const workspaceDir = path.join(tempHome, "workspace"); + const configPath = path.join(stateDir, "openclaw.json"); + const bundledPluginsDir = path.join(tempHome, "bundled-plugins"); + await Promise.all([ + fs.mkdir(workspaceDir, { recursive: true }), + fs.mkdir(bundledPluginsDir, { recursive: true }), + fs.mkdir(path.dirname(configPath), { recursive: true }), + ]); + for (const [key, value] of Object.entries({ + HOME: tempHome, + OPENCLAW_STATE_DIR: stateDir, + OPENCLAW_CONFIG_PATH: configPath, + OPENCLAW_GATEWAY_TOKEN: "pr119473-proof-token", + OPENCLAW_SKIP_CHANNELS: "1", + OPENCLAW_SKIP_GMAIL_WATCHER: "1", + OPENCLAW_SKIP_CRON: "1", + OPENCLAW_SKIP_CANVAS_HOST: "1", + OPENCLAW_SKIP_BROWSER_CONTROL_SERVER: "1", + OPENCLAW_SKIP_PROVIDERS: "1", + OPENCLAW_BUNDLED_PLUGINS_DIR: bundledPluginsDir, + OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1", + })) { + setTestEnvValue(key, value); + } + + providerServer = createServer((_request, response) => { + response.writeHead(200, { "content-type": "text/event-stream" }); + const message = { + type: "message", + id: "pr119473-proof-message", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "PR119473_RUNTIME_OK", annotations: [] }], + }; + response.end( + [ + { + type: "response.output_item.added", + output_index: 0, + item: { ...message, status: "in_progress", content: [] }, + }, + { type: "response.output_item.done", output_index: 0, item: message }, + { + type: "response.completed", + response: { + status: "completed", + usage: { input_tokens: 10, output_tokens: 5, total_tokens: 15 }, + }, + }, + ] + .map((event) => `data: ${JSON.stringify(event)}\n\n`) + .concat("data: [DONE]\n\n") + .join(""), + ); + }); + await new Promise((resolve, reject) => { + providerServer?.once("error", reject); + providerServer?.listen(0, "127.0.0.1", resolve); + }); + const providerAddress = providerServer.address(); + if (!providerAddress || typeof providerAddress === "string") { + throw new Error("proof provider did not bind a loopback port"); + } + const provider = buildMockOpenAiResponsesProvider( + `http://127.0.0.1:${providerAddress.port}/v1`, + ); + const cfg = { + agents: { + defaults: { + workspace: workspaceDir, + skipBootstrap: true, + model: { primary: provider.modelRef }, + models: { + [provider.modelRef]: { params: { transport: "sse", openaiWsWarmup: false } }, + }, + }, + entries: { main: { default: true } }, + }, + models: { mode: "replace", providers: { [provider.providerId]: provider.config } }, + gateway: { auth: { mode: "token", token: "pr119473-proof-token" } }, + }; + const sessionKey = "agent:main:pr119473-fresh-runtime"; + const marker = "PR119473_FIRST_USER_MESSAGE"; + first = await startGatewayWithClient({ + cfg, + configPath, + token: "pr119473-proof-token", + clientDisplayName: "pr119473-proof-first-start", + }); + const started = await first.client.request<{ runId?: string; status?: string }>( + "chat.send", + { + sessionKey, + message: marker, + deliver: false, + idempotencyKey: "pr119473-proof-first-turn", + }, + ); + expect(started.status).toBe("started"); + expect(started.runId).toEqual(expect.any(String)); + await expect( + first.client.request( + "agent.wait", + { runId: started.runId, timeoutMs: 30_000 }, + { timeoutMs: 35_000 }, + ), + ).resolves.toMatchObject({ status: "ok" }); + await disconnectGatewayClient(first.client); + await first.server.close({ reason: "PR #119473 proof restart" }); + first = undefined; + + clearRuntimeConfigSnapshot(); + clearConfigCache(); + second = await startGatewayWithClient({ + cfg, + configPath, + token: "pr119473-proof-token", + clientDisplayName: "pr119473-proof-after-restart", + }); + const history = await second.client.request<{ messages?: unknown[] }>("chat.history", { + sessionKey, + limit: 20, + }); + expect(JSON.stringify(history.messages ?? [])).toContain(marker); + } finally { + if (first) { + await disconnectGatewayClient(first.client).catch(() => undefined); + await first.server.close().catch(() => undefined); + } + if (second) { + await disconnectGatewayClient(second.client).catch(() => undefined); + await second.server.close().catch(() => undefined); + } + if (providerServer?.listening) { + await new Promise((resolve) => { + providerServer?.close(() => resolve()); + }); + } + envSnapshot.restore(); + clearRuntimeConfigSnapshot(); + clearConfigCache(); + clearSessionStoreCacheForTest(); + } + }, + ); +});