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

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");

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();

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,

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<string, SessionEntry>();
+  let current = new Map<string, SessionEntry>();
+  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,

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,

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,

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<typeof createServer> | undefined;
+      let first: Awaited<ReturnType<typeof startGatewayWithClient>> | undefined;
+      let second: Awaited<ReturnType<typeof startGatewayWithClient>> | 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<void>((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<void>((resolve) => {
+            providerServer?.close(() => resolve());
+          });
+        }
+        envSnapshot.restore();
+        clearRuntimeConfigSnapshot();
+        clearConfigCache();
+        clearSessionStoreCacheForTest();
+      }
+    },
+  );
+});