diff --git a/extensions/codex/src/app-server/run-attempt-context.ts b/extensions/codex/src/app-server/run-attempt-context.ts index 7b4f83214ded..13f456ceb195 100644 --- a/extensions/codex/src/app-server/run-attempt-context.ts +++ b/extensions/codex/src/app-server/run-attempt-context.ts @@ -64,6 +64,7 @@ export async function prepareCodexAttemptContext( sessionFile: activeSessionFile, sessionId: activeSessionId, sessionKey: contextSessionKey, + sessionTarget: params.sessionTarget, }; const historyState = { messages: diff --git a/extensions/codex/src/app-server/run-attempt.test.ts b/extensions/codex/src/app-server/run-attempt.test.ts index 0579cd159f12..4df5785b2270 100644 --- a/extensions/codex/src/app-server/run-attempt.test.ts +++ b/extensions/codex/src/app-server/run-attempt.test.ts @@ -19,7 +19,10 @@ import { registerPluginCommand } from "openclaw/plugin-sdk/plugin-runtime"; import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime"; import { GPT5_BEHAVIOR_CONTRACT as CODEX_GPT5_BEHAVIOR_CONTRACT } from "openclaw/plugin-sdk/provider-model-shared"; import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime"; -import { readSessionTranscriptEvents } from "openclaw/plugin-sdk/session-transcript-runtime"; +import { + appendSessionTranscriptMessageByIdentity, + readSessionTranscriptEvents, +} from "openclaw/plugin-sdk/session-transcript-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; import WebSocket from "ws"; import { defaultCodexAppInventoryCache } from "./app-inventory-cache.js"; @@ -264,6 +267,24 @@ async function attachSqliteSessionTarget( }); } +async function appendSqliteHistoryMessage( + params: EmbeddedRunAttemptParams, + message: ReturnType | ReturnType, +): Promise { + const target = params.sessionTarget; + if (!target?.agentId || !target.sessionId || !target.sessionKey || !target.storePath) { + throw new Error("expected complete SQLite session target"); + } + await appendSessionTranscriptMessageByIdentity({ + agentId: target.agentId, + sessionId: target.sessionId, + sessionKey: target.sessionKey, + storePath: target.storePath, + message, + now: message.timestamp, + }); +} + async function readTranscriptMessagesByIdentity( params: EmbeddedRunAttemptParams, ): Promise>> { @@ -2538,6 +2559,40 @@ describe("runCodexAppServerAttempt", () => { expect(inputText).toContain("Current user request:"); expect(inputText).toContain("make the default webpage openclaw"); }); + it("projects canonical SQLite continuity when starting without a native thread binding", async () => { + const sessionId = "session-sqlite-fresh-continuity"; + const sessionFile = `agent:main:${sessionId}`; + const storePath = path.join(tempDir, "sqlite-fresh-continuity.sqlite"); + const workspaceDir = path.join(tempDir, "workspace-sqlite-fresh-continuity"); + const params = createParams(sessionFile, workspaceDir); + await attachSqliteSessionTarget(params, storePath, sessionId); + await appendSqliteHistoryMessage( + params, + userMessage("canonical SQLite startup question", Date.now()), + ); + await appendSqliteHistoryMessage( + params, + assistantMessage("canonical SQLite startup answer", Date.now() + 1), + ); + params.prompt = "continue the canonical SQLite startup"; + const harness = createStartedThreadHarness(); + + const run = runCodexAppServerAttempt(params); + await harness.waitForMethod("turn/start"); + await harness.completeTurn({ threadId: "thread-1", turnId: "turn-1" }); + await run; + + const turnStart = harness.requests.find((request) => request.method === "turn/start"); + const inputText = + (turnStart?.params as { input?: Array<{ text?: string }> } | undefined)?.input?.[0]?.text ?? + ""; + expect(harness.requests.map((request) => request.method)).toContain("thread/start"); + expect(inputText).toContain("OpenClaw assembled context for this turn:"); + expect(inputText).toContain("canonical SQLite startup question"); + expect(inputText).toContain("canonical SQLite startup answer"); + expect(inputText).toContain("Current user request:"); + expect(inputText).toContain("continue the canonical SQLite startup"); + }); it("keeps large fresh-thread continuity under the Codex turn/start input limit", async () => { const { sessionFile, workspaceDir } = createRunPaths(); const sessionManager = openFileBackedSessionManagerForTest(sessionFile); @@ -2809,6 +2864,51 @@ describe("runCodexAppServerAttempt", () => { expect(inputText).toContain("Current user request:"); expect(inputText).toContain("is the previous message trustworthy?"); }); + it("projects newer canonical SQLite continuity when a resumed binding is stale", async () => { + const sessionId = "session-sqlite-resume-continuity"; + const sessionFile = `agent:main:${sessionId}`; + const storePath = path.join(tempDir, "sqlite-resume-continuity.sqlite"); + const workspaceDir = path.join(tempDir, "workspace-sqlite-resume-continuity"); + const params = createParams(sessionFile, workspaceDir); + await attachSqliteSessionTarget(params, storePath, sessionId); + await writeExistingBinding(sessionFile, workspaceDir, { dynamicToolsFingerprint: "[]" }); + const binding = await readCodexAppServerBinding(sessionFile); + const bindingUpdatedAt = Date.parse(binding?.historyCoveredThrough ?? ""); + if (!Number.isFinite(bindingUpdatedAt)) { + throw new Error("expected valid Codex binding timestamp"); + } + await appendSqliteHistoryMessage( + params, + userMessage("old canonical SQLite native-owned context", bindingUpdatedAt - 2_000), + ); + await appendSqliteHistoryMessage( + params, + userMessage("new canonical SQLite resume question", bindingUpdatedAt + 1_000), + ); + await appendSqliteHistoryMessage( + params, + assistantMessage("new canonical SQLite resume answer", bindingUpdatedAt + 2_000), + ); + params.prompt = "continue the canonical SQLite resume"; + const harness = createResumeHarness(); + + const run = runCodexAppServerAttempt(params); + await harness.waitForMethod("turn/start"); + await harness.completeTurn({ threadId: "thread-existing", turnId: "turn-1" }); + await run; + + const turnStart = harness.requests.find((request) => request.method === "turn/start"); + const inputText = + (turnStart?.params as { input?: Array<{ text?: string }> } | undefined)?.input?.[0]?.text ?? + ""; + expect(harness.requests.map((request) => request.method)).toContain("thread/resume"); + expect(inputText).toContain("OpenClaw assembled context for this turn:"); + expect(inputText).not.toContain("old canonical SQLite native-owned context"); + expect(inputText).toContain("new canonical SQLite resume question"); + expect(inputText).toContain("new canonical SQLite resume answer"); + expect(inputText).toContain("Current user request:"); + expect(inputText).toContain("continue the canonical SQLite resume"); + }); it("does not project Codex mirrored transcript echoes as stale binding continuity", async () => { const { sessionFile, workspaceDir } = createRunPaths(); await writeExistingBinding(sessionFile, workspaceDir, { dynamicToolsFingerprint: "[]" }); @@ -3600,7 +3700,7 @@ describe("runCodexAppServerAttempt", () => { it("captures the complete mirrored branch through a settled tool-result boundary", async () => { const storePath = path.join(tempDir, "settled-finalization-context.sqlite"); const sessionId = "session-settled-finalization-context"; - const sessionFile = `sqlite:main:${sessionId}:${storePath}`; + const sessionFile = `agent:main:${sessionId}`; const workspaceDir = path.join(tempDir, "workspace-settled-finalization-context"); const harness = createStartedThreadHarness(); const params = createParams(sessionFile, workspaceDir); diff --git a/extensions/codex/src/app-server/session-history.test.ts b/extensions/codex/src/app-server/session-history.test.ts index 4bfb11458888..da3e0e10d5cf 100644 --- a/extensions/codex/src/app-server/session-history.test.ts +++ b/extensions/codex/src/app-server/session-history.test.ts @@ -67,6 +67,12 @@ function mirroredTarget(sessionFile: string) { async function writeSqliteSession(params: { storedSessionFile?: string } = {}): Promise<{ marker: string; sessionKey: string; + sessionTarget: { + agentId: string; + sessionId: string; + sessionKey: string; + storePath: string; + }; }> { const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-session-history-sqlite-")); tempDirs.push(dir); @@ -96,7 +102,7 @@ async function writeSqliteSession(params: { storedSessionFile?: string } = {}): ...scope, message: { role: "assistant", content: "sqlite answer", timestamp: 2 }, }); - return { marker, sessionKey }; + return { marker, sessionKey, sessionTarget: scope }; } describe("readCodexMirroredSessionHistoryMessages", () => { @@ -169,6 +175,57 @@ describe("readCodexMirroredSessionHistoryMessages", () => { ]); }); + it("replays SQLite history from the canonical typed session target", async () => { + const { sessionKey, sessionTarget } = await writeSqliteSession({ + storedSessionFile: "agent:main:codex-sqlite", + }); + + await expect( + readCodexMirroredSessionHistoryMessages({ + agentId: "main", + sessionFile: sessionKey, + sessionId: "codex-sqlite-session", + sessionKey, + sessionTarget, + }), + ).resolves.toMatchObject([ + { role: "user", content: "sqlite prompt" }, + { role: "assistant", content: "sqlite answer" }, + ]); + }); + + it.each([ + ["agent id", { agentId: "other" }], + ["session id", { sessionId: "another-session" }], + ["session key", { sessionKey: "agent:main:another-session" }], + ])("fails closed when the typed target has a mismatched %s", async (_label, targetPatch) => { + const { marker, sessionKey, sessionTarget } = await writeSqliteSession(); + + await expect( + readCodexMirroredSessionHistoryMessages({ + agentId: "main", + sessionFile: marker, + sessionId: "codex-sqlite-session", + sessionKey, + sessionTarget: { ...sessionTarget, ...targetPatch }, + }), + ).resolves.toEqual([]); + }); + + it("fails closed when the typed session target is incomplete", async () => { + const { sessionKey, sessionTarget } = await writeSqliteSession(); + + await expect( + readCodexMirroredSessionHistoryMessages({ + agentId: "main", + sessionFile: sessionKey, + sessionId: "codex-sqlite-session", + sessionKey, + sessionTarget: { ...sessionTarget, storePath: undefined }, + }), + ).resolves.toEqual([]); + }); + it("resolves SQLite marker history when the caller has no session key", async () => { const { marker } = await writeSqliteSession(); diff --git a/extensions/codex/src/app-server/session-history.ts b/extensions/codex/src/app-server/session-history.ts index 71867344fab7..a8ad4dfa433d 100644 --- a/extensions/codex/src/app-server/session-history.ts +++ b/extensions/codex/src/app-server/session-history.ts @@ -16,18 +16,22 @@ import { resolveTranscriptSessionKeyBySessionId, type SqliteSessionFileMarker, } from "openclaw/plugin-sdk/session-store-runtime"; -import { readSessionTranscriptEvents } from "openclaw/plugin-sdk/session-transcript-runtime"; +import { + readSessionTranscriptEvents, + type SessionTranscriptTargetParams, +} from "openclaw/plugin-sdk/session-transcript-runtime"; import { sanitizeCodexHistoryImagePayloads } from "./image-payload-sanitizer.js"; function isMissingFileError(error: unknown): boolean { return Boolean(error && typeof error === "object" && "code" in error && error.code === "ENOENT"); } -type CodexMirroredSessionHistoryTarget = { +export type CodexMirroredSessionHistoryTarget = { agentId?: string; sessionFile: string; sessionId: string; sessionKey?: string; + sessionTarget?: Partial; }; /** Returns sanitized session-context messages for a Codex mirrored session file. */ @@ -78,6 +82,26 @@ export async function readCodexMirroredSessionHistoryMessages( async function readCodexMirroredSessionEntries( target: CodexMirroredSessionHistoryTarget, ): Promise { + if (target.sessionTarget) { + const { agentId, sessionId, sessionKey, storePath } = target.sessionTarget; + if ( + !agentId || + !sessionId || + !sessionKey || + !storePath || + sessionId !== target.sessionId || + (target.agentId !== undefined && agentId !== target.agentId) || + (target.sessionKey !== undefined && sessionKey !== target.sessionKey) + ) { + return []; + } + return (await readSessionTranscriptEvents({ + agentId, + sessionId, + sessionKey, + storePath, + })) as SessionEntry[]; + } const sqliteMarker = parseSqliteSessionFileMarker(target.sessionFile); if (sqliteMarker) { if ( diff --git a/extensions/codex/src/app-server/settled-turn-context.ts b/extensions/codex/src/app-server/settled-turn-context.ts index b570fb6dbd93..7fee60f29c57 100644 --- a/extensions/codex/src/app-server/settled-turn-context.ts +++ b/extensions/codex/src/app-server/settled-turn-context.ts @@ -4,7 +4,10 @@ import { type AgentMessage, } from "openclaw/plugin-sdk/agent-harness-runtime"; import type { EmbeddedRunAttemptResult } from "./attempt-terminal.js"; -import { readCodexMirroredSessionHistoryMessages } from "./session-history.js"; +import { + readCodexMirroredSessionHistoryMessages, + type CodexMirroredSessionHistoryTarget, +} from "./session-history.js"; import { serializeCodexMirrorSourceEvidence } from "./transcript-mirror-attestation.js"; import { readMirrorIdentity } from "./upstream-prompt-provenance.js"; @@ -109,15 +112,13 @@ function buildCodexSettledTurnFinalizationContext(params: { } /** Reads and freezes the current active transcript branch after mirroring has settled. */ -export async function captureCodexSettledTurnFinalizationContext(params: { - agentId?: string; - sessionFile: string; - sessionId: string; - sessionKey?: string; - mirroredMessages: readonly AgentMessage[]; - settledMessages: readonly AgentMessage[]; - turnId: string; -}): Promise { +export async function captureCodexSettledTurnFinalizationContext( + params: CodexMirroredSessionHistoryTarget & { + mirroredMessages: readonly AgentMessage[]; + settledMessages: readonly AgentMessage[]; + turnId: string; + }, +): Promise { try { const historyMessages = await readCodexMirroredSessionHistoryMessages(params); if (!historyMessages) {