From ea54060223134c103ce396f914c735b707f7c4b2 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 19 Jul 2026 00:11:10 -0700 Subject: [PATCH] feat(codex): fork upstream-linked sessions at a message via thread/fork (#111149) * feat(codex): fork upstream-linked sessions at a message via thread/fork * fix(gateway): fail closed for rewind and branch switch on upstream-linked sessions * fix(codex): fail closed on first-message forks, image-only prompts, and orphan archival * fix(codex): baseline retained history and reject paginated threads on upstream fork * fix(codex): validate the full fork prefix and fail closed across crash windows * fix(codex): treat all non-text inputs as unverifiable in fork drift checks * fix(codex): support first-message forks as empty-history upstream cuts * fix(codex): reject source-id reuse and unverifiable hidden inputs in fork boundaries * refactor(codex): materialize upstream forks from verified thread read-back * fix(codex): satisfy strict type lanes and knip for upstream fork --- extensions/codex/harness.ts | 24 ++ extensions/codex/index.ts | 2 + .../app-server/protocol-validators.test.ts | 19 ++ .../src/app-server/protocol-validators.ts | 16 + extensions/codex/src/app-server/protocol.ts | 1 + .../src/app-server/session-history-import.ts | 64 ++++ .../app-server/upstream-fork-boundary.test.ts | 205 +++++++++++ .../src/app-server/upstream-fork-boundary.ts | 321 ++++++++++++++++++ .../app-server/upstream-session-fork.test.ts | 281 +++++++++++++++ .../src/app-server/upstream-session-fork.ts | 219 ++++++++++++ extensions/codex/src/session-catalog-types.ts | 3 + extensions/codex/src/session-catalog.ts | 53 +-- scripts/plugin-sdk-surface-report.mjs | 7 +- src/agents/harness/types.ts | 44 +++ .../server-methods/sessions-rewind.test.ts | 131 ++++++- src/gateway/server-methods/sessions-rewind.ts | 134 ++++++-- src/plugin-sdk/agent-harness-runtime.ts | 3 + src/plugin-sdk/session-catalog.ts | 4 + src/sessions/session-upstream-links.ts | 6 +- 19 files changed, 1487 insertions(+), 50 deletions(-) create mode 100644 extensions/codex/src/app-server/session-history-import.ts create mode 100644 extensions/codex/src/app-server/upstream-fork-boundary.test.ts create mode 100644 extensions/codex/src/app-server/upstream-fork-boundary.ts create mode 100644 extensions/codex/src/app-server/upstream-session-fork.test.ts create mode 100644 extensions/codex/src/app-server/upstream-session-fork.ts diff --git a/extensions/codex/harness.ts b/extensions/codex/harness.ts index 0b24afc264cc..a82d400addfd 100644 --- a/extensions/codex/harness.ts +++ b/extensions/codex/harness.ts @@ -8,7 +8,9 @@ import type { ContextEngineHostCapability, } from "openclaw/plugin-sdk/agent-harness-runtime"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; import type { CodexAppServerBindingStore } from "./src/app-server/session-binding.js"; +import type { CodexSessionCatalogControl } from "./src/session-catalog-types.js"; // `codex` is legacy input only until Part 2 doctor migration rewrites stored refs. // New runtime identity uses the `openai` provider. @@ -50,7 +52,9 @@ export function createCodexAppServerAgentHarness(options: { pluginConfig?: unknown; resolvePluginConfig?: () => unknown; resolveConfig?: () => OpenClawConfig | undefined; + runtime?: PluginRuntime; bindingStore: CodexAppServerBindingStore; + sessionCatalogControl?: CodexSessionCatalogControl; }): AgentHarness { const harnessRuntimeId = options?.id ?? "codex"; const normalizedHarnessRuntimeId = harnessRuntimeId.trim().toLowerCase(); @@ -59,6 +63,8 @@ export function createCodexAppServerAgentHarness(options: { id.trim().toLowerCase(), ), ); + const sessionCatalogControl = options.sessionCatalogControl; + const sessionRuntime = options.runtime; const harness: CodexAppServerAgentHarness = { id: harnessRuntimeId, label: options?.label ?? "Codex agent harness", @@ -69,6 +75,24 @@ export function createCodexAppServerAgentHarness(options: { visibleReplies: "message_tool", }, authBootstrap: "harness", + ...(sessionCatalogControl && sessionRuntime + ? { + sessionFork: { + upstreamKinds: ["codex-app-server"] as const, + fork: async (params) => { + const { forkCodexUpstreamSession } = + await import("./src/app-server/upstream-session-fork.js"); + return await forkCodexUpstreamSession(params, { + bindingStore: options.bindingStore, + control: sessionCatalogControl, + harnessRuntimeId, + resolveConfig: options.resolveConfig, + runtime: sessionRuntime, + }); + }, + }, + } + : {}), authBinding: { fingerprint: async (params) => { const { fingerprintCodexAppServerAuthBinding } = diff --git a/extensions/codex/index.ts b/extensions/codex/index.ts index 31949c194eb2..748409520a99 100644 --- a/extensions/codex/index.ts +++ b/extensions/codex/index.ts @@ -150,8 +150,10 @@ export default definePluginEntry({ api.registerAgentHarness( createCodexAppServerAgentHarness({ bindingStore, + sessionCatalogControl, resolveConfig: resolveCurrentConfig, resolvePluginConfig: resolveCurrentPluginConfig, + runtime: api.runtime, }), ); api.registerMediaUnderstandingProvider( diff --git a/extensions/codex/src/app-server/protocol-validators.test.ts b/extensions/codex/src/app-server/protocol-validators.test.ts index c3a1cfcaa2cf..89ad8e01ae0d 100644 --- a/extensions/codex/src/app-server/protocol-validators.test.ts +++ b/extensions/codex/src/app-server/protocol-validators.test.ts @@ -1,6 +1,7 @@ // Codex tests cover protocol validators plugin behavior. import { describe, expect, it } from "vitest"; import { + assertCodexThreadForkParams, readCodexModelListResponse, readCodexTurn, assertCodexThreadStartResponse, @@ -52,6 +53,24 @@ describe("Codex thread response validators", () => { }); }); +describe("assertCodexThreadForkParams", () => { + it("accepts the experimental beforeTurnId boundary", () => { + expect( + assertCodexThreadForkParams({ + threadId: "thread-1", + beforeTurnId: "turn-2", + excludeTurns: true, + }), + ).toMatchObject({ beforeTurnId: "turn-2" }); + }); + + it("rejects a non-string beforeTurnId", () => { + expect(() => assertCodexThreadForkParams({ threadId: "thread-1", beforeTurnId: 2 })).toThrow( + "Invalid Codex app-server thread/fork params", + ); + }); +}); + describe("assertCodexThreadStartResponse", () => { it("accepts response with both id and sessionId", () => { const response = makeMinimalResponse(); diff --git a/extensions/codex/src/app-server/protocol-validators.ts b/extensions/codex/src/app-server/protocol-validators.ts index 169c06bee590..2f9d939a37c1 100644 --- a/extensions/codex/src/app-server/protocol-validators.ts +++ b/extensions/codex/src/app-server/protocol-validators.ts @@ -16,6 +16,7 @@ import type { CodexErrorNotification, CodexModelListResponse, CodexThreadForkResponse, + CodexThreadForkParams, CodexThreadResumeResponse, CodexThreadStartResponse, CodexTurn, @@ -236,6 +237,21 @@ export function assertCodexThreadForkResponse(value: unknown): CodexThreadForkRe return assertCodexShape(validateThreadStartResponse, normalized, "thread/fork response"); } +/** Asserts the experimental beforeTurnId request field before it crosses the app-server boundary. */ +export function assertCodexThreadForkParams(value: unknown): CodexThreadForkParams { + if ( + !isRecord(value) || + typeof value.threadId !== "string" || + !value.threadId.trim() || + (value.beforeTurnId !== undefined && + value.beforeTurnId !== null && + typeof value.beforeTurnId !== "string") + ) { + throw new Error("Invalid Codex app-server thread/fork params"); + } + return value as CodexThreadForkParams; +} + /** Asserts and normalizes a Codex thread/resume response. */ export function assertCodexThreadResumeResponse(value: unknown): CodexThreadResumeResponse { const normalized = normalizeWithDefaults(threadResumeResponseSchema, value); diff --git a/extensions/codex/src/app-server/protocol.ts b/extensions/codex/src/app-server/protocol.ts index 70bd1cc11a26..d315c0955558 100644 --- a/extensions/codex/src/app-server/protocol.ts +++ b/extensions/codex/src/app-server/protocol.ts @@ -165,6 +165,7 @@ export type CodexThreadStartResponse = { export type CodexThreadForkParams = JsonObject & { threadId: string; lastTurnId?: string | null; + beforeTurnId?: string | null; path?: string | null; model?: string | null; modelProvider?: string | null; diff --git a/extensions/codex/src/app-server/session-history-import.ts b/extensions/codex/src/app-server/session-history-import.ts new file mode 100644 index 000000000000..0931044f5731 --- /dev/null +++ b/extensions/codex/src/app-server/session-history-import.ts @@ -0,0 +1,64 @@ +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; +import { resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; +import type { CodexThread } from "./protocol.js"; +import { importCodexThreadHistoryToTranscript } from "./transcript-mirror.js"; + +type CreatedCodexImportedSession = Awaited< + ReturnType +>; + +/** Creates a session whose transcript is derived from one verified Codex thread snapshot. */ +export async function createImportedCodexSession(params: { + runtime: PluginRuntime; + config: OpenClawConfig; + key: string; + agentId: string; + thread: CodexThread; + throughTurnId: string | null; + recoverMatchingInitialEntry?: true; + initialEntry: { + agentHarnessId: string; + modelSelectionLocked?: true; + pluginExtensions?: CreatedCodexImportedSession["entry"]["pluginExtensions"]; + }; + afterImport: ( + created: CreatedCodexImportedSession, + ) => Promise<{ pluginExtensions: CreatedCodexImportedSession["entry"]["pluginExtensions"] }>; +}): Promise { + const label = params.thread.name?.trim() || undefined; + const spawnedCwd = params.thread.cwd?.trim() || undefined; + const createParams = { + cfg: params.config, + key: params.key, + agentId: params.agentId, + ...(label ? { label } : {}), + ...(spawnedCwd ? { spawnedCwd } : {}), + initialEntry: params.initialEntry, + afterCreate: async (entry: CreatedCodexImportedSession) => { + // Post-flip the mirror targets SQLite rows; resolve the agent's store + // path instead of trusting the legacy sessionFile locator marker. + const storePath = resolveStorePath(params.config.session?.store, { + agentId: entry.agentId, + }); + await importCodexThreadHistoryToTranscript({ + thread: params.thread, + throughTurnId: params.throughTurnId, + storePath, + sessionId: entry.sessionId, + sessionKey: entry.key, + agentId: entry.agentId, + ...(spawnedCwd ? { cwd: spawnedCwd } : {}), + modelProvider: params.thread.modelProvider, + config: params.config, + }); + return await params.afterImport(entry); + }, + }; + return params.recoverMatchingInitialEntry + ? await params.runtime.agent.session.createSessionEntry({ + ...createParams, + recoverMatchingInitialEntry: true, + }) + : await params.runtime.agent.session.createSessionEntry(createParams); +} diff --git a/extensions/codex/src/app-server/upstream-fork-boundary.test.ts b/extensions/codex/src/app-server/upstream-fork-boundary.test.ts new file mode 100644 index 000000000000..78fa5205a33f --- /dev/null +++ b/extensions/codex/src/app-server/upstream-fork-boundary.test.ts @@ -0,0 +1,205 @@ +import type { SessionTranscriptMessageEntry } from "openclaw/plugin-sdk/session-transcript-runtime"; +import { describe, expect, it, vi } from "vitest"; +import type { CodexThreadItem, CodexTurn } from "./protocol.js"; +import { resolveCodexUpstreamForkBoundary } from "./upstream-fork-boundary.js"; + +const transcriptMocks = vi.hoisted(() => ({ + readVisibleEntries: vi.fn(), +})); + +vi.mock("openclaw/plugin-sdk/session-transcript-runtime", () => ({ + readVisibleSessionTranscriptMessageEntries: transcriptMocks.readVisibleEntries, +})); + +function item(type: string, overrides: Record = {}): CodexThreadItem { + return { id: `${type}-item`, type, ...overrides } as CodexThreadItem; +} + +function user(text: string): CodexThreadItem { + return item("userMessage", { content: [{ type: "text", text, textElements: [] }] }); +} + +function turn(id: string, items: CodexThreadItem[], overrides: Partial = {}): CodexTurn { + return { id, status: "completed", items, ...overrides }; +} + +async function resolveFromTurns(params: { + turns: readonly CodexTurn[]; + userMessageOrdinal: number; + localPrefixTexts: readonly (string | undefined)[]; +}) { + const entries: SessionTranscriptMessageEntry[] = params.localPrefixTexts.map((text, index) => ({ + entryId: `entry-${index}`, + parentId: index > 0 ? `entry-${index - 1}` : null, + seq: index, + role: "user", + message: { + role: "user", + content: text ?? [{ type: "image", data: "", mimeType: "image/png" }], + timestamp: index, + }, + })); + transcriptMocks.readVisibleEntries.mockResolvedValue(entries); + const result = await resolveCodexUpstreamForkBoundary({ + agentId: "main", + sessionId: "session-1", + sessionKey: "agent:main:upstream", + storePath: "/tmp/does-not-matter", + entryId: `entry-${params.userMessageOrdinal}`, + threadId: "thread-1", + control: { + readThread: vi.fn(async () => ({ id: "thread-1" })), + listTurnPage: vi.fn(async () => ({ data: [...params.turns] })), + } as unknown as Parameters[0]["control"], + }); + return result.ok ? { ok: true as const, boundary: result.boundary } : result; +} + +describe("resolveCodexUpstreamForkBoundaryFromTurns", () => { + it("maps the local user ordinal to the upstream turn", async () => { + const result = await resolveFromTurns({ + turns: [turn("turn-1", [user("one")]), turn("turn-2", [user("two")])], + userMessageOrdinal: 1, + localPrefixTexts: ["one", "two"], + }); + + expect(result).toEqual({ + ok: true, + boundary: { + beforeTurnId: "turn-2", + targetTurnId: "turn-2", + retainedMarker: { turnId: "turn-1", userMessageCount: 1 }, + }, + }); + }); + + it("cuts before the first turn with an empty retained baseline", async () => { + const result = await resolveFromTurns({ + turns: [turn("turn-1", [user("one")])], + userMessageOrdinal: 0, + localPrefixTexts: ["one"], + }); + expect(result).toEqual({ + ok: true, + boundary: { + beforeTurnId: "turn-1", + targetTurnId: "turn-1", + retainedMarker: { turnId: null, userMessageCount: 0 }, + }, + }); + }); + + it("rejects a selected steer message", async () => { + const result = await resolveFromTurns({ + turns: [turn("turn-1", [user("one"), user("steer")])], + userMessageOrdinal: 1, + localPrefixTexts: ["one", "steer"], + }); + + expect(result).toMatchObject({ ok: false, code: "steer-message" }); + }); + + it("skips prompts inside review spans", async () => { + const result = await resolveFromTurns({ + turns: [ + turn("turn-review", [ + item("enteredReviewMode"), + user("hidden review prompt"), + item("exitedReviewMode"), + ]), + turn("turn-2", [user("visible")]), + ], + userMessageOrdinal: 0, + localPrefixTexts: ["visible"], + }); + + expect(result).toEqual({ + ok: true, + boundary: { + beforeTurnId: "turn-2", + targetTurnId: "turn-2", + retainedMarker: { turnId: "turn-review", userMessageCount: 1 }, + }, + }); + }); + + it("rejects an in-progress target turn", async () => { + const result = await resolveFromTurns({ + turns: [turn("turn-1", [user("one")], { status: "inProgress" })], + userMessageOrdinal: 0, + localPrefixTexts: ["one"], + }); + + expect(result).toMatchObject({ ok: false, code: "in-progress-turn" }); + }); + + it("rejects local and upstream text drift", async () => { + const result = await resolveFromTurns({ + turns: [turn("turn-1", [user("persisted")])], + userMessageOrdinal: 0, + localPrefixTexts: ["local mirror"], + }); + + expect(result).toMatchObject({ ok: false, code: "drift-mismatch" }); + }); + + it("rejects equal targets over divergent prefixes", async () => { + const result = await resolveFromTurns({ + turns: [turn("turn-1", [user("upstream-old")]), turn("turn-2", [user("target")])], + userMessageOrdinal: 1, + localPrefixTexts: ["local-old", "target"], + }); + + expect(result).toMatchObject({ ok: false, code: "drift-mismatch" }); + }); + + it("rejects upstream messages carrying semantic non-text inputs", async () => { + const result = await resolveFromTurns({ + turns: [ + turn("turn-1", [ + item("userMessage", { + content: [ + { type: "text", text: "one", textElements: [] }, + { type: "skill", name: "reviewer" }, + ], + }), + ]), + turn("turn-2", [user("target")]), + ], + userMessageOrdinal: 1, + localPrefixTexts: ["one", "target"], + }); + + expect(result).toMatchObject({ ok: false, code: "drift-mismatch" }); + }); + + it("rejects prefixes whose content identity cannot be verified", async () => { + const result = await resolveFromTurns({ + turns: [turn("turn-1", [user("one")]), turn("turn-2", [user("target")])], + userMessageOrdinal: 1, + localPrefixTexts: [undefined, "target"], + }); + + expect(result).toMatchObject({ ok: false, code: "drift-mismatch" }); + }); +}); + +describe("resolveCodexUpstreamForkBoundary", () => { + it("rejects paginated-history threads before reading turns", async () => { + const readThread = vi.fn(async () => ({ id: "thread-1", historyMode: "paginated" })); + const result = await resolveCodexUpstreamForkBoundary({ + agentId: "main", + sessionId: "session-1", + sessionKey: "agent:main:upstream", + storePath: "/tmp/does-not-matter", + entryId: "entry-1", + threadId: "thread-1", + control: { readThread } as unknown as Parameters< + typeof resolveCodexUpstreamForkBoundary + >[0]["control"], + }); + + expect(result).toMatchObject({ ok: false, code: "upstream-unavailable" }); + expect(readThread).toHaveBeenCalledWith("thread-1", false); + }); +}); diff --git a/extensions/codex/src/app-server/upstream-fork-boundary.ts b/extensions/codex/src/app-server/upstream-fork-boundary.ts new file mode 100644 index 000000000000..e57c49dd3335 --- /dev/null +++ b/extensions/codex/src/app-server/upstream-fork-boundary.ts @@ -0,0 +1,321 @@ +import { readVisibleSessionTranscriptMessageEntries } from "openclaw/plugin-sdk/session-transcript-runtime"; +import type { CodexSessionCatalogControl } from "../session-catalog-types.js"; +import type { CodexThreadItem, CodexTurn } from "./protocol.js"; + +type CodexUpstreamForkBoundaryFailureCode = + | "steer-message" + | "in-progress-turn" + | "drift-mismatch" + | "upstream-unavailable"; + +type CodexUpstreamForkBoundary = { + beforeTurnId: string; + targetTurnId: string; + /** Baseline for the forked thread: the last retained turn (null when the cut is + * before the first turn), so the upstream monitor does not replay retained + * history as fresh external activity. */ + retainedMarker: { turnId: string | null; userMessageCount: number }; +}; + +type CodexUpstreamForkBoundaryResult = + | { ok: true; boundary: CodexUpstreamForkBoundary; editorText?: string } + | { ok: false; code: CodexUpstreamForkBoundaryFailureCode; message: string }; + +const TURN_PAGE_LIMIT = 100; + +type UserInput = { + type?: unknown; + text?: unknown; + textElements?: unknown; + url?: unknown; + path?: unknown; +}; + +function failure( + code: CodexUpstreamForkBoundaryFailureCode, + message: string, +): CodexUpstreamForkBoundaryResult { + return { ok: false, code, message }; +} + +function asInputs(item: CodexThreadItem): UserInput[] { + return Array.isArray(item.content) ? (item.content as UserInput[]) : []; +} + +function userMessageDisplay(item: CodexThreadItem): { + text: string; + visible: boolean; + hasUnverifiableInput: boolean; +} { + let text = ""; + let hasTextElement = false; + let hasImage = false; + // Any non-text input (images, skills, mentions, future variants) has no canonical + // cross-system identity; its presence makes the message unverifiable for drift checks. + let hasUnverifiableInput = false; + for (const input of asInputs(item)) { + if (input.type === "text") { + if (typeof input.text === "string") { + text += input.text; + } + hasTextElement ||= Array.isArray(input.textElements) && input.textElements.length > 0; + } else { + hasUnverifiableInput = true; + hasImage ||= input.type === "image" || input.type === "localImage"; + } + } + return { + text, + visible: Boolean(text.trim()) || hasTextElement || hasImage, + hasUnverifiableInput, + }; +} + +function isHiddenNestedReviewTurn(previous: CodexTurn | undefined, turn: CodexTurn): boolean { + if ( + previous?.status !== "completed" || + turn.status !== "interrupted" || + turn.completedAt != null || + !previous.items.some((item) => item.type === "enteredReviewMode") || + !previous.items.some((item) => item.type === "exitedReviewMode") + ) { + return false; + } + const userMessages = turn.items.filter((item) => item.type === "userMessage"); + const [firstUserMessage, secondUserMessage] = userMessages; + if (!firstUserMessage || !secondUserMessage || userMessages.length !== 2) { + return false; + } + return JSON.stringify(asInputs(firstUserMessage)) === JSON.stringify(asInputs(secondUserMessage)); +} + +function localMessageText(content: unknown): string | undefined { + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return undefined; + } + // Non-text blocks (images/attachments) have no canonical cross-system identity; + // undefined marks the message unverifiable so boundary resolution fails closed. + const texts: string[] = []; + for (const block of content) { + if (!block || typeof block !== "object" || Array.isArray(block)) { + return undefined; + } + const typed = block as { type?: unknown; text?: unknown }; + if (typed.type !== "text" || typeof typed.text !== "string") { + return undefined; + } + texts.push(typed.text); + } + return texts.join(""); +} + +function resolveCodexUpstreamForkBoundaryFromTurns(params: { + turns: readonly CodexTurn[]; + userMessageOrdinal: number; + /** Canonical text for every visible local user message through the target ordinal; + * undefined marks content (images/attachments) whose identity cannot be verified. */ + localPrefixTexts: readonly (string | undefined)[]; +}): CodexUpstreamForkBoundaryResult { + let visibleUserMessagesSeen = 0; + let reviewMode = false; + for (const [turnIndex, turn] of params.turns.entries()) { + const hiddenNestedReviewTurn = isHiddenNestedReviewTurn(params.turns[turnIndex - 1], turn); + let userMessagesInTurn = 0; + for (const item of turn.items) { + if (item.type === "enteredReviewMode") { + reviewMode = true; + continue; + } + if (item.type === "exitedReviewMode") { + reviewMode = false; + continue; + } + if (item.type !== "userMessage") { + continue; + } + const isSteer = userMessagesInTurn > 0; + userMessagesInTurn += 1; + if (reviewMode || hiddenNestedReviewTurn) { + continue; + } + const display = userMessageDisplay(item); + // Unverifiable inputs fail closed even when display-invisible: a skipped + // skill/mention-only message would silently desync ordinals against the mirror. + if (display.hasUnverifiableInput) { + return failure( + "drift-mismatch", + "A message before the fork point contains images or attachments that cannot be verified across OpenClaw and Codex. Fork from a text-only span instead.", + ); + } + if (!display.visible) { + continue; + } + const ordinal = visibleUserMessagesSeen; + if (ordinal > params.userMessageOrdinal) { + break; + } + // The local transcript is only a mirror; every prefix message must match, not just + // the target — equal tails over different prefixes would bind divergent histories. + const localText = params.localPrefixTexts[ordinal]; + if (localText === undefined) { + return failure( + "drift-mismatch", + "A message before the fork point contains images or attachments that cannot be verified across OpenClaw and Codex. Fork from a text-only span instead.", + ); + } + if (display.text !== localText) { + return failure( + "drift-mismatch", + "The local conversation no longer matches the Codex thread. Refresh the session and try again.", + ); + } + if (ordinal !== params.userMessageOrdinal) { + visibleUserMessagesSeen += 1; + continue; + } + if (isSteer) { + return failure( + "steer-message", + "This message steered an existing Codex turn and cannot be forked independently. Fork from the turn's first message instead.", + ); + } + if (turn.status === "inProgress") { + return failure( + "in-progress-turn", + "This Codex turn is still in progress. Wait for it to finish, then try forking again.", + ); + } + // beforeTurnId at the first turn yields a valid empty-history fork upstream + // (codex-rs thread_fork_inner has no minimum-turn guard), matching the empty + // local mirror prefix. + const retained = turnIndex > 0 ? params.turns[turnIndex - 1] : undefined; + return { + ok: true, + boundary: { + beforeTurnId: turn.id, + targetTurnId: turn.id, + retainedMarker: retained + ? { + turnId: retained.id, + userMessageCount: retained.items.filter( + (retainedItem) => retainedItem.type === "userMessage", + ).length, + } + : { turnId: null, userMessageCount: 0 }, + }, + }; + } + } + return failure( + "drift-mismatch", + "The message could not be matched to the Codex thread. Refresh the session and try again.", + ); +} + +export async function listCodexUpstreamTurns( + control: CodexSessionCatalogControl, + threadId: string, +): Promise { + const turns: CodexTurn[] = []; + const seenCursors = new Set(); + let cursor: string | undefined; + for (;;) { + const page = await control.listTurnPage({ + threadId, + limit: TURN_PAGE_LIMIT, + sortDirection: "asc", + itemsView: "full", + ...(cursor ? { cursor } : {}), + }); + turns.push(...page.data); + const nextCursor = page.nextCursor?.trim() || undefined; + if (!nextCursor) { + return turns; + } + if (seenCursors.has(nextCursor)) { + throw new Error("Codex returned a repeated thread/turns/list cursor"); + } + seenCursors.add(nextCursor); + cursor = nextCursor; + } +} + +export async function resolveCodexUpstreamForkBoundary(params: { + agentId: string; + sessionId: string; + sessionKey: string; + storePath: string; + entryId: string; + threadId: string; + control: CodexSessionCatalogControl; +}): Promise { + try { + // Paginated-history threads reject itemsView "full" turn reads (thread/items/list + // is required); fork support for them is future work — fail closed with intent. + const thread = await params.control.readThread(params.threadId, false); + if (thread.historyMode === "paginated") { + return failure( + "upstream-unavailable", + "This Codex thread uses paginated history, which cannot be forked from OpenClaw yet.", + ); + } + const entries = await readVisibleSessionTranscriptMessageEntries({ + agentId: params.agentId, + sessionId: params.sessionId, + sessionKey: params.sessionKey, + storePath: params.storePath, + }); + const visibleUserEntries = entries.filter((entry) => entry.role === "user"); + const userMessageOrdinal = visibleUserEntries.findIndex( + (entry) => entry.entryId === params.entryId, + ); + if (userMessageOrdinal < 0) { + return failure( + "drift-mismatch", + "The local message could not be mapped to the Codex thread. Refresh the session and try again.", + ); + } + const localPrefixTexts = visibleUserEntries + .slice(0, userMessageOrdinal + 1) + .map((entry) => + localMessageText("content" in entry.message ? entry.message.content : undefined), + ); + const turns = await listCodexUpstreamTurns(params.control, params.threadId); + const resolved = resolveCodexUpstreamForkBoundaryFromTurns({ + turns, + userMessageOrdinal, + localPrefixTexts, + }); + return resolved.ok + ? { ...resolved, editorText: localPrefixTexts[userMessageOrdinal] } + : resolved; + } catch { + return failure( + "upstream-unavailable", + "The Codex thread could not be read. Check that Codex is available, then try again.", + ); + } +} + +export function precheckCodexUpstreamForkBoundary(params: { + boundary: CodexUpstreamForkBoundary; + turns: readonly CodexTurn[]; +}): CodexUpstreamForkBoundaryResult { + const target = params.turns.find((turn) => turn.id === params.boundary.targetTurnId); + if (!target) { + return failure( + "upstream-unavailable", + "The Codex thread changed before it could be forked. Refresh the session and try again.", + ); + } + if (target.status === "inProgress") { + return failure( + "in-progress-turn", + "This Codex turn is still in progress. Wait for it to finish, then try forking again.", + ); + } + return { ok: true, boundary: params.boundary }; +} diff --git a/extensions/codex/src/app-server/upstream-session-fork.test.ts b/extensions/codex/src/app-server/upstream-session-fork.test.ts new file mode 100644 index 000000000000..4988f045c99c --- /dev/null +++ b/extensions/codex/src/app-server/upstream-session-fork.test.ts @@ -0,0 +1,281 @@ +import { createPluginRuntimeMock } from "openclaw/plugin-sdk/plugin-test-runtime"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { CodexSessionCatalogControl } from "../session-catalog-types.js"; +import type { CodexThreadForkParams, CodexTurn } from "./protocol.js"; +import type { CodexAppServerBindingStore } from "./session-binding.js"; + +const boundaryMocks = vi.hoisted(() => ({ + listTurns: vi.fn(), +})); +const linkMocks = vi.hoisted(() => ({ + delete: vi.fn(), + upsert: vi.fn(), +})); +const transcriptMocks = vi.hoisted(() => ({ + importHistory: vi.fn(), +})); + +const boundary = { + beforeTurnId: "turn-2", + targetTurnId: "turn-2", + retainedMarker: { turnId: "turn-1", userMessageCount: 1 }, +} as const; + +vi.mock("openclaw/plugin-sdk/session-catalog", async (importOriginal) => ({ + ...(await importOriginal()), + deleteSessionUpstreamLink: linkMocks.delete, + upsertSessionUpstreamLink: linkMocks.upsert, +})); + +vi.mock("./transcript-mirror.js", () => ({ + importCodexThreadHistoryToTranscript: transcriptMocks.importHistory, +})); + +vi.mock("./upstream-fork-boundary.js", () => ({ + resolveCodexUpstreamForkBoundary: vi.fn(async () => ({ + ok: true, + boundary, + editorText: "edit me", + })), + listCodexUpstreamTurns: boundaryMocks.listTurns, + precheckCodexUpstreamForkBoundary: vi.fn(() => ({ ok: true, boundary })), +})); + +import { forkCodexUpstreamSession } from "./upstream-session-fork.js"; + +function turn(id: string, text: string): CodexTurn { + return { + id, + status: "completed", + items: [ + { + aggregatedOutput: null, + changes: [], + command: null, + cwd: null, + id: `${id}-user`, + name: null, + query: null, + server: null, + status: null, + text: "", + title: null, + tool: null, + content: [{ type: "text", text, textElements: [] }], + type: "userMessage", + }, + ], + }; +} + +function forkResponse(threadId = "thread-forked") { + return { + approvalPolicy: "never", + approvalsReviewer: "user", + cwd: "/tmp", + model: "gpt-5.4", + modelProvider: "openai", + sandbox: { type: "dangerFullAccess" }, + thread: { + id: threadId, + sessionId: "session-forked", + cliVersion: "0.143.0", + createdAt: 1715299200, + updatedAt: 1715299200, + cwd: "/tmp", + ephemeral: false, + modelProvider: "openai", + preview: "forked thread", + source: "appServer", + status: { type: "notLoaded" }, + turns: [], + }, + }; +} + +function forkParams() { + return { + targetKey: "agent:main:dashboard:forked", + source: { + agentId: "main", + sessionId: "session-source", + sessionKey: "agent:main:source", + storePath: "/tmp/sessions.db", + entryId: "entry-2", + }, + upstream: { + catalogId: "codex", + hostId: "gateway:local", + kind: "codex-app-server" as const, + threadId: "thread-source", + ref: { connectionFingerprint: "fingerprint", threadId: "thread-source" }, + }, + }; +} + +type ForkThreadStub = (params: CodexThreadForkParams) => Promise; + +function forkControl(forkThread: ForkThreadStub = vi.fn(async () => forkResponse())) { + const archiveThread = vi.fn(async () => undefined); + const control = { + archiveThread, + connectionFingerprint: "fingerprint", + forkThread, + } as unknown as CodexSessionCatalogControl; + control.withPinnedConnection = async (run) => await run(control); + return { archiveThread, control, forkThread }; +} + +beforeEach(() => { + boundaryMocks.listTurns.mockReset(); + linkMocks.delete.mockReset(); + linkMocks.upsert.mockReset().mockReturnValue(true); + transcriptMocks.importHistory.mockReset().mockResolvedValue({ + importedMessages: 1, + omittedMessages: 0, + }); +}); + +describe("forkCodexUpstreamSession", () => { + it("verifies the cut, imports the fork history, then links before binding", async () => { + const retainedTurn = turn("turn-1", "one"); + boundaryMocks.listTurns + .mockResolvedValueOnce([turn("turn-2", "edit me")]) + .mockResolvedValueOnce([retainedTurn]); + const { archiveThread, control, forkThread } = forkControl(); + const events: string[] = []; + linkMocks.upsert.mockImplementation(() => { + events.push("link"); + return true; + }); + const mutate = vi.fn(async () => { + events.push("bind"); + return true; + }); + const runtime = createPluginRuntimeMock(); + const createSessionEntry = vi.mocked(runtime.agent.session.createSessionEntry); + + const result = await forkCodexUpstreamSession(forkParams(), { + bindingStore: { mutate } as unknown as CodexAppServerBindingStore, + control, + harnessRuntimeId: "codex-custom", + resolveConfig: () => ({}), + runtime, + }); + + expect(forkThread).toHaveBeenCalledWith({ + threadId: "thread-source", + beforeTurnId: "turn-2", + excludeTurns: true, + }); + expect(boundaryMocks.listTurns).toHaveBeenLastCalledWith(control, "thread-forked"); + expect(transcriptMocks.importHistory).toHaveBeenCalledWith( + expect.objectContaining({ + sessionKey: "agent:main:dashboard:forked", + thread: expect.objectContaining({ id: "thread-forked", turns: [retainedTurn] }), + throughTurnId: "turn-1", + }), + ); + expect(linkMocks.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + marker: { turnId: "turn-1", userMessageCount: 1 }, + sessionKey: "agent:main:dashboard:forked", + threadId: "thread-forked", + }), + ); + expect(runtime.agent.session.createSessionEntry).toHaveBeenCalledWith( + expect.objectContaining({ + initialEntry: expect.objectContaining({ agentHarnessId: "codex-custom" }), + }), + ); + expect(createSessionEntry.mock.calls[0]?.[0]).not.toHaveProperty("recoverMatchingInitialEntry"); + expect(events).toEqual(["link", "bind"]); + expect(result).toEqual({ + status: "created", + key: "agent:main:dashboard:forked", + editorText: "edit me", + }); + expect(archiveThread).not.toHaveBeenCalled(); + }); + + it("archives a fork whose read-back history proves beforeTurnId was ignored", async () => { + boundaryMocks.listTurns + .mockResolvedValueOnce([turn("turn-2", "edit me")]) + .mockResolvedValueOnce([turn("turn-1", "one"), turn("turn-2", "edit me")]); + const { archiveThread, control } = forkControl(); + const runtime = createPluginRuntimeMock(); + + const result = await forkCodexUpstreamSession(forkParams(), { + bindingStore: { mutate: vi.fn() } as unknown as CodexAppServerBindingStore, + control, + harnessRuntimeId: "codex", + runtime, + }); + + expect(result).toMatchObject({ + status: "failed", + code: "upstream-unavailable", + message: expect.stringContaining("Codex version"), + }); + expect(archiveThread).toHaveBeenCalledWith("thread-forked"); + expect(runtime.agent.session.createSessionEntry).not.toHaveBeenCalled(); + expect(linkMocks.upsert).not.toHaveBeenCalled(); + }); + + it("cleans the link and archives the fork when binding materialization fails", async () => { + boundaryMocks.listTurns + .mockResolvedValueOnce([turn("turn-2", "edit me")]) + .mockResolvedValueOnce([turn("turn-1", "one")]); + const { archiveThread, control } = forkControl(); + const mutate = vi.fn(async () => false); + + const result = await forkCodexUpstreamSession(forkParams(), { + bindingStore: { mutate } as unknown as CodexAppServerBindingStore, + control, + harnessRuntimeId: "codex", + runtime: createPluginRuntimeMock(), + }); + + expect(result).toMatchObject({ status: "failed", code: "upstream-unavailable" }); + expect(linkMocks.delete).toHaveBeenCalledWith("agent:main:dashboard:forked", "main"); + expect(mutate).toHaveBeenLastCalledWith(expect.anything(), { + kind: "clear", + threadId: "thread-forked", + }); + expect(archiveThread).toHaveBeenCalledWith("thread-forked"); + }); + + it("archives a recoverable orphan id when the fork response is invalid", async () => { + boundaryMocks.listTurns.mockResolvedValueOnce([turn("turn-2", "edit me")]); + const { archiveThread, control } = forkControl( + vi.fn(async () => ({ thread: { id: "thread-orphan" } })), + ); + + const result = await forkCodexUpstreamSession(forkParams(), { + bindingStore: {} as CodexAppServerBindingStore, + control, + harnessRuntimeId: "codex", + runtime: createPluginRuntimeMock(), + }); + + expect(result).toMatchObject({ status: "failed", code: "upstream-unavailable" }); + expect(archiveThread).toHaveBeenCalledWith("thread-orphan"); + }); + + it("rejects a fork response that reuses the source thread id", async () => { + boundaryMocks.listTurns.mockResolvedValueOnce([turn("turn-2", "edit me")]); + const { archiveThread, control } = forkControl( + vi.fn(async () => forkResponse("thread-source")), + ); + + const result = await forkCodexUpstreamSession(forkParams(), { + bindingStore: { mutate: vi.fn() } as unknown as CodexAppServerBindingStore, + control, + harnessRuntimeId: "codex", + runtime: createPluginRuntimeMock(), + }); + + expect(result).toMatchObject({ status: "failed", code: "upstream-unavailable" }); + expect(archiveThread).not.toHaveBeenCalled(); + }); +}); diff --git a/extensions/codex/src/app-server/upstream-session-fork.ts b/extensions/codex/src/app-server/upstream-session-fork.ts new file mode 100644 index 000000000000..8de6ad24bdad --- /dev/null +++ b/extensions/codex/src/app-server/upstream-session-fork.ts @@ -0,0 +1,219 @@ +import type { + AgentHarnessSessionForkParams, + AgentHarnessSessionForkResult, +} from "openclaw/plugin-sdk/agent-harness-runtime"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; +import { + deleteSessionUpstreamLink, + upsertSessionUpstreamLink, +} from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import type { CodexSessionCatalogControl } from "../session-catalog-types.js"; +import { codexLastTerminalTurnId, codexUpstreamBaseline } from "../session-upstream-marker.js"; +import { assertCodexThreadForkResponse } from "./protocol-validators.js"; +import type { CodexThread, CodexThreadForkResponse } from "./protocol.js"; +import { sessionBindingIdentity, type CodexAppServerBindingStore } from "./session-binding.js"; +import { createImportedCodexSession } from "./session-history-import.js"; +import { + listCodexUpstreamTurns, + precheckCodexUpstreamForkBoundary, + resolveCodexUpstreamForkBoundary, +} from "./upstream-fork-boundary.js"; + +function readConnectionFingerprint(ref: unknown): string | undefined { + if (!isRecord(ref)) { + return undefined; + } + return typeof ref.connectionFingerprint === "string" && ref.connectionFingerprint.trim() + ? ref.connectionFingerprint + : undefined; +} + +function normalizeTurnId(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +export async function forkCodexUpstreamSession( + params: AgentHarnessSessionForkParams, + options: { + bindingStore: CodexAppServerBindingStore; + control: CodexSessionCatalogControl; + harnessRuntimeId: string; + resolveConfig?: () => OpenClawConfig | undefined; + runtime: PluginRuntime; + }, +): Promise { + try { + return await options.control.withPinnedConnection(async (control) => { + let linked = false; + let bindingIdentity: ReturnType | undefined; + const compensateFork = async (forkedThreadId: string) => { + if (bindingIdentity) { + await options.bindingStore + .mutate(bindingIdentity, { kind: "clear", threadId: forkedThreadId }) + .catch(() => undefined); + } + if (linked) { + deleteSessionUpstreamLink(params.targetKey, params.source.agentId); + } + await control.archiveThread(forkedThreadId).catch(() => undefined); + }; + const sourceFingerprint = readConnectionFingerprint(params.upstream.ref); + if ( + params.upstream.kind !== "codex-app-server" || + !sourceFingerprint || + sourceFingerprint !== control.connectionFingerprint + ) { + return { + status: "failed", + code: "upstream-unavailable", + message: + "This Codex thread is not available on the current connection. Reconnect to its host and try again.", + }; + } + const resolved = await resolveCodexUpstreamForkBoundary({ + ...params.source, + threadId: params.upstream.threadId, + control, + }); + if (!resolved.ok) { + return { status: "failed", code: resolved.code, message: resolved.message }; + } + const liveTurns = await listCodexUpstreamTurns(control, params.upstream.threadId); + const precheck = precheckCodexUpstreamForkBoundary({ + boundary: resolved.boundary, + turns: liveTurns, + }); + if (!precheck.ok) { + return { status: "failed", code: precheck.code, message: precheck.message }; + } + // beforeTurnId is experimental; the initialized shared client explicitly negotiates it. + const rawResponse = await control.forkThread({ + threadId: params.upstream.threadId, + beforeTurnId: resolved.boundary.beforeTurnId, + excludeTurns: true, + }); + let response: CodexThreadForkResponse; + try { + response = assertCodexThreadForkResponse(rawResponse); + } catch (error) { + const orphanThreadId = + isRecord(rawResponse.thread) && typeof rawResponse.thread.id === "string" + ? rawResponse.thread.id.trim() + : ""; + // A malformed response cannot be trusted to name a NEW thread; never archive an + // id that matches the source conversation. + if (orphanThreadId && orphanThreadId !== params.upstream.threadId) { + await control.archiveThread(orphanThreadId).catch(() => undefined); + } + throw error; + } + const threadId = response.thread.id.trim(); + if (!threadId) { + throw new Error("Codex thread/fork response did not include a thread id"); + } + // A contract-violating response reusing the source id would bind (and later + // archive) the original conversation; reject identity reuse outright. + if (threadId === params.upstream.threadId) { + throw new Error("Codex thread/fork response reused the source thread id"); + } + const forkedThreadId = threadId; + try { + const connectionFingerprint = control.connectionFingerprint; + if (!connectionFingerprint) { + throw new Error("Codex fork connection did not include a fingerprint"); + } + const forkedTurns = await listCodexUpstreamTurns(control, threadId); + const expectedLastTurnId = resolved.boundary.retainedMarker.turnId; + const actualLastTurnId = forkedTurns.at(-1)?.id ?? null; + // Boundary resolution already verified the source prefix; this read-back tail identity + // detects app-server versions that ignored the exclusive beforeTurnId cut. + if (actualLastTurnId !== expectedLastTurnId) { + await compensateFork(forkedThreadId); + return { + status: "failed", + code: "upstream-unavailable", + message: + "This Codex version does not support message-level forks. Update Codex, reconnect, and try again.", + }; + } + const forkedThread: CodexThread = { ...response.thread, turns: forkedTurns }; + const throughTurnId = codexLastTerminalTurnId(forkedThread, normalizeTurnId) ?? null; + const marker = codexUpstreamBaseline(forkedThread, normalizeTurnId); + const config = options.resolveConfig?.() ?? {}; + const created = await createImportedCodexSession({ + runtime: options.runtime, + config, + key: params.targetKey, + agentId: params.source.agentId, + thread: forkedThread, + throughTurnId, + initialEntry: { + agentHarnessId: options.harnessRuntimeId, + modelSelectionLocked: true, + }, + afterImport: async (entry) => { + bindingIdentity = sessionBindingIdentity({ + agentId: entry.agentId, + sessionId: entry.sessionId, + sessionKey: entry.key, + config, + }); + // Link BEFORE bind: a crash cannot expose a bound session to local-only + // rewind/switch while its canonical upstream ownership is missing. + linked = upsertSessionUpstreamLink({ + sessionKey: entry.key, + agentId: entry.agentId, + catalogId: params.upstream.catalogId, + hostId: params.upstream.hostId, + threadId, + upstreamKind: params.upstream.kind, + upstreamRef: { connectionFingerprint, threadId }, + marker, + }); + if (!linked) { + throw new Error("Codex fork link could not be persisted"); + } + const attached = await options.bindingStore.mutate(bindingIdentity, { + kind: "set", + binding: { + threadId, + cwd: forkedThread.cwd ?? "", + model: response.model, + modelProvider: response.modelProvider ?? undefined, + historyCoveredThrough: new Date().toISOString(), + }, + }); + if (!attached) { + throw new Error("Codex session binding changed before the fork could be attached"); + } + return { pluginExtensions: entry.entry.pluginExtensions }; + }, + }); + return { + status: "created", + key: created.key, + ...(resolved.editorText !== undefined ? { editorText: resolved.editorText } : {}), + }; + } catch { + // thread/fork commits before local materialization. The guarded session initializer + // rolls back its row/transcript; this capability clears link/binding and archives the orphan. + await compensateFork(forkedThreadId); + return { + status: "failed", + code: "upstream-unavailable", + message: + "The Codex fork could not be verified or imported into a new session. Refresh sessions and try again.", + }; + } + }); + } catch { + return { + status: "failed", + code: "upstream-unavailable", + message: + "The Codex thread could not be forked. Check that Codex is available, then try again.", + }; + } +} diff --git a/extensions/codex/src/session-catalog-types.ts b/extensions/codex/src/session-catalog-types.ts index c72b86ad9405..2338faf9ccc8 100644 --- a/extensions/codex/src/session-catalog-types.ts +++ b/extensions/codex/src/session-catalog-types.ts @@ -1,5 +1,7 @@ import type { CodexThread, + CodexThreadForkParams, + CodexThreadForkResponse, CodexThreadListParams, CodexThreadListResponse, CodexThreadTurnsListParams, @@ -45,6 +47,7 @@ export type CodexSessionCatalogControl = { listPage(params: CodexSessionCatalogPageParams): Promise; listDescendantPage(params: CodexThreadListParams): Promise; listTurnPage(params: CodexThreadTurnsListParams): Promise; + forkThread(params: CodexThreadForkParams): Promise; readThread(threadId: string, includeTurns?: boolean): Promise; archiveThread(threadId: string): Promise; }; diff --git a/extensions/codex/src/session-catalog.ts b/extensions/codex/src/session-catalog.ts index 54f3032f7bfe..919513206310 100644 --- a/extensions/codex/src/session-catalog.ts +++ b/extensions/codex/src/session-catalog.ts @@ -11,13 +11,15 @@ import type { SessionCatalogHost, SessionCatalogProvider, } from "openclaw/plugin-sdk/session-catalog"; -import { resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { CODEX_CONTROL_METHODS } from "./app-server/capabilities.js"; import { resolveCodexSupervisionAppServerRuntimeOptions } from "./app-server/config.js"; import { buildCodexAppServerConnectionFingerprint } from "./app-server/plugin-app-cache-key.js"; +import { assertCodexThreadForkParams } from "./app-server/protocol-validators.js"; import type { CodexThread, + CodexThreadForkParams, + CodexThreadForkResponse, CodexThreadListParams, CodexThreadListResponse, CodexThreadTurnsListParams, @@ -31,12 +33,12 @@ import { type CodexAppServerPendingSupervisionBranch, type CodexAppServerThreadBinding, } from "./app-server/session-binding.js"; +import { createImportedCodexSession } from "./app-server/session-history-import.js"; import { getLeasedSharedCodexAppServerClient, releaseLeasedSharedCodexAppServerClient, } from "./app-server/shared-client.js"; import { assertCodexArchiveDescendantsUnowned } from "./app-server/thread-archive-guard.js"; -import { importCodexThreadHistoryToTranscript } from "./app-server/transcript-mirror.js"; import { codexControlRequest } from "./command-rpc.js"; import { adoptedSourceKey, @@ -123,6 +125,7 @@ type CodexSessionCatalogRequestSnapshot = { requestTimeoutMs: number; listThreads(params: CodexThreadListParams, timeoutMs: number): Promise; listThreadTurns(params: CodexThreadTurnsListParams): Promise; + forkThread(params: CodexThreadForkParams): Promise; readThread(threadId: string, includeTurns: boolean): Promise; archiveThread(threadId: string): Promise; }; @@ -213,6 +216,9 @@ function createCodexSessionCatalogControlFromRequests(params: { const response = await params.createRequestSnapshot().listThreadTurns(listParams); return response; }, + async forkThread(forkParams) { + return await params.createRequestSnapshot().forkThread(forkParams); + }, async archiveThread(threadId) { await params.createRequestSnapshot().archiveThread(threadId); }, @@ -257,6 +263,13 @@ export function createCodexSessionCatalogControl(params: { listParams, requestOptions, ), + forkThread: async (forkParams) => + await codexControlRequest( + pluginConfig, + CODEX_CONTROL_METHODS.forkThread, + assertCodexThreadForkParams(forkParams), + requestOptions, + ), archiveThread: async (threadId) => { await codexControlRequest( pluginConfig, @@ -307,6 +320,14 @@ export function createCodexSessionCatalogControl(params: { config: runtimeConfig, timeoutMs: runtime.requestTimeoutMs, }), + forkThread: async (forkParams) => + await requestCodexAppServerClientJson({ + client, + method: CODEX_CONTROL_METHODS.forkThread, + requestParams: assertCodexThreadForkParams(forkParams), + config: runtimeConfig, + timeoutMs: runtime.requestTimeoutMs, + }), archiveThread: async (threadId) => { await requestCodexAppServerClientJson({ client, @@ -866,17 +887,17 @@ async function createOrReuseAdoptedSession(params: { let createdBindingIdentity: ReturnType | undefined; let createdPendingBinding: CodexAppServerPendingSupervisionBranch | undefined; try { - const label = params.sourceThread.name?.trim() || undefined; const spawnedCwd = params.sourceThread.cwd?.trim() || undefined; const pendingLastTurnId = codexLastTerminalTurnId(params.sourceThread, boundCatalogSessionId); const marker: CodexSupervisionMarker = { sourceThreadId: params.sourceThread.id }; - const created = await params.api.runtime.agent.session.createSessionEntry({ - cfg: params.config, + const created = await createImportedCodexSession({ + runtime: params.api.runtime, + config: params.config, key: adoptionSessionKey(params.sourceThread.id), agentId: resolveDefaultAgentId(params.config), + thread: params.sourceThread, + throughTurnId: pendingLastTurnId ?? null, recoverMatchingInitialEntry: true, - ...(label ? { label } : {}), - ...(spawnedCwd ? { spawnedCwd } : {}), initialEntry: { agentHarnessId: "codex", modelSelectionLocked: true, @@ -890,28 +911,12 @@ async function createOrReuseAdoptedSession(params: { }, }, }, - afterCreate: async (entry) => { + afterImport: async (entry) => { createdBindingIdentity = sessionBindingIdentity({ sessionId: entry.sessionId, sessionKey: entry.key, config: params.config, }); - // Post-flip the mirror targets SQLite rows; resolve the agent's store - // path instead of trusting the legacy sessionFile locator marker. - const storePath = resolveStorePath(params.config.session?.store, { - agentId: entry.agentId, - }); - await importCodexThreadHistoryToTranscript({ - thread: params.sourceThread, - throughTurnId: pendingLastTurnId ?? null, - storePath, - sessionId: entry.sessionId, - sessionKey: entry.key, - agentId: entry.agentId, - ...(spawnedCwd ? { cwd: spawnedCwd } : {}), - modelProvider: params.sourceThread.modelProvider, - config: params.config, - }); createdPendingBinding = { sourceThreadId: params.sourceThread.id, connectionFingerprint: params.connectionFingerprint, diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index 2b52779737f2..c91259aa7d41 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -282,7 +282,9 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { // +9: shared ingress monitor factory and lifecycle/result contracts across // channel-outbound and its two deprecated compatibility barrels. // +1: SwarmConfig exposes the tools.swarm contract through config-types. - 8178, + // +3: harness sessionFork capability params, result, and failure-code contracts. + // +2: upstream-link registry write/delete for harness-owned session forks. + 8183, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( @@ -327,7 +329,8 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { // +1: bounded raw transcript cursor reader. // +1: bounded visible transcript cursor reader. // +3: shared ingress monitor factory across channel-outbound and compat mirrors. - 4546, + // +2: upstream-link registry write/delete for harness-owned session forks. + 4548, env, ), publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/agents/harness/types.ts b/src/agents/harness/types.ts index 55cb35bae3ab..dae125974da0 100644 --- a/src/agents/harness/types.ts +++ b/src/agents/harness/types.ts @@ -128,6 +128,42 @@ export type AgentHarnessResetParams = { reason?: "new" | "reset" | "idle" | "daily" | "compaction" | "deleted" | "unknown"; }; +export type AgentHarnessSessionForkFailureCode = + | "steer-message" + | "in-progress-turn" + | "drift-mismatch" + | "upstream-unavailable"; + +export type AgentHarnessSessionForkParams = { + targetKey: string; + source: { + agentId: string; + sessionId: string; + sessionKey: string; + storePath: string; + entryId: string; + }; + upstream: { + catalogId: string; + hostId: string; + kind: import("../../plugins/session-catalog.js").SessionUpstreamKind; + threadId: string; + ref: import("../../plugins/session-catalog.js").SessionUpstreamJsonValue; + }; +}; + +export type AgentHarnessSessionForkResult = + | { + status: "created"; + key: string; + editorText?: string; + } + | { + status: "failed"; + code: AgentHarnessSessionForkFailureCode; + message: string; + }; + export type AgentHarnessResultClassification = | "ok" | NonNullable; @@ -188,6 +224,13 @@ type AgentHarnessSessionLifecycleCapability = { dispose?(): Promise | void; }; +type AgentHarnessSessionForkCapability = { + sessionFork?: { + upstreamKinds: readonly import("../../plugins/session-catalog.js").SessionUpstreamKind[]; + fork(params: AgentHarnessSessionForkParams): Promise; + }; +}; + type AgentHarnessRuntimeArtifactCapability = { /** Revalidate an artifact only at setup and persistent-operation boundaries. */ runtimeArtifact?: { @@ -225,6 +268,7 @@ export type AgentHarness = AgentHarnessRunCapability & AgentHarnessRuntimeArtifactCapability & AgentHarnessAuthBindingCapability & AgentHarnessProviderUsageCapability & + AgentHarnessSessionForkCapability & AgentHarnessSessionLifecycleCapability; export type RegisteredAgentHarness = { diff --git a/src/gateway/server-methods/sessions-rewind.test.ts b/src/gateway/server-methods/sessions-rewind.test.ts index 15c6997371b6..376bcbb893d7 100644 --- a/src/gateway/server-methods/sessions-rewind.test.ts +++ b/src/gateway/server-methods/sessions-rewind.test.ts @@ -8,16 +8,46 @@ import type { GatewayRequestContext, RespondFn } from "./types.js"; const mocks = vi.hoisted(() => ({ active: false, + capability: false, external: false, + upstreamFork: vi.fn(), queueClear: vi.fn(), })); +vi.mock("../../agents/harness/registry.js", () => ({ + listRegisteredAgentHarnesses: () => + mocks.capability + ? [ + { + harness: { + sessionFork: { + upstreamKinds: ["codex-app-server"], + fork: mocks.upstreamFork, + }, + }, + }, + ] + : [], +})); + vi.mock("../../auto-reply/reply/queue/cleanup.js", () => ({ clearSessionQueues: mocks.queueClear, })); vi.mock("../../sessions/session-upstream-links.js", () => ({ - readSessionUpstreamLink: () => (mocks.external ? { upstreamKind: "external" } : undefined), + readSessionUpstreamLink: () => + mocks.external + ? { + agentId: "main", + catalogId: "codex", + hostId: "gateway:local", + marker: { turnId: "turn-2", userMessageCount: 1 }, + sessionKey, + threadId: "thread-source", + upstreamKind: "codex-app-server", + upstreamRef: { connectionFingerprint: "fingerprint", threadId: "thread-source" }, + } + : undefined, })); vi.mock("./session-active-runs.js", () => { @@ -27,6 +57,7 @@ vi.mock("./session-active-runs.js", () => { import { appendTranscriptEvent, appendTranscriptMessage, + listSessionEntries, upsertSessionEntry, } from "../../config/sessions/session-accessor.js"; import { sessionsHandlers } from "./sessions.js"; @@ -36,7 +67,9 @@ const sessionKey = "agent:main:rewind-handler"; beforeEach(async () => { mocks.active = false; + mocks.capability = false; mocks.external = false; + mocks.upstreamFork.mockReset(); mocks.queueClear.mockReset(); vi.stubEnv("OPENCLAW_STATE_DIR", tempDirs.make("openclaw-rewind-handler-")); await upsertSessionEntry( @@ -222,6 +255,102 @@ describe("session message-cut methods", () => { } }); + it.each(["sessions.rewind", "sessions.branches.switch"] as const)( + "rejects %s for upstream-linked sessions even with a fork-capable harness", + async (method) => { + mocks.external = true; + mocks.capability = true; + const respond = await invoke(method, "user-entry"); + + expect(respond).toHaveBeenCalledWith( + false, + undefined, + expect.objectContaining({ + code: ErrorCodes.INVALID_REQUEST, + message: expect.stringContaining("external agent harness"), + }), + ); + expect(mocks.upstreamFork).not.toHaveBeenCalled(); + }, + ); + + it("delegates complete upstream fork materialization to the harness", async () => { + mocks.external = true; + mocks.capability = true; + mocks.upstreamFork.mockResolvedValue({ + status: "created", + key: "agent:main:dashboard:forked", + editorText: "edit me", + }); + + const respond = await invoke("sessions.fork", "user-entry"); + expect(respond).toHaveBeenCalledWith( + true, + { editorText: "edit me", sessionKey: "agent:main:dashboard:forked" }, + undefined, + ); + expect(mocks.upstreamFork).toHaveBeenCalledWith( + expect.objectContaining({ + source: expect.objectContaining({ entryId: "user-entry", sessionKey }), + targetKey: expect.stringMatching(/^agent:main:dashboard:/), + upstream: expect.objectContaining({ + catalogId: "codex", + hostId: "gateway:local", + kind: "codex-app-server", + threadId: "thread-source", + }), + }), + ); + }); + + it("does not mutate the local session when the upstream fork fails", async () => { + mocks.external = true; + mocks.capability = true; + mocks.upstreamFork.mockResolvedValue({ + status: "failed", + code: "upstream-unavailable", + message: "Codex is offline. Try again.", + }); + + const entryCount = listSessionEntries({ agentId: "main" }).length; + const respond = await invoke("sessions.fork", "user-entry"); + + expect(respond).toHaveBeenCalledWith( + false, + undefined, + expect.objectContaining({ + code: ErrorCodes.UNAVAILABLE, + details: { reason: "upstream-unavailable" }, + }), + ); + expect(listSessionEntries({ agentId: "main" })).toHaveLength(entryCount); + }); + + it.each(["steer-message", "in-progress-turn", "drift-mismatch"] as const)( + "passes through the %s boundary failure", + async (reason) => { + mocks.external = true; + mocks.capability = true; + mocks.upstreamFork.mockResolvedValue({ + status: "failed", + code: reason, + message: `boundary failed: ${reason}`, + }); + + const respond = await invoke("sessions.fork", "user-entry"); + + expect(respond).toHaveBeenCalledWith( + false, + undefined, + expect.objectContaining({ + code: ErrorCodes.INVALID_REQUEST, + details: { reason }, + message: `boundary failed: ${reason}`, + }), + ); + }, + ); + it("returns a typed error for unsupported transcript storage", async () => { await upsertSessionEntry( { agentId: "main", sessionKey }, diff --git a/src/gateway/server-methods/sessions-rewind.ts b/src/gateway/server-methods/sessions-rewind.ts index c067d2f36cbc..643ebd0cb650 100644 --- a/src/gateway/server-methods/sessions-rewind.ts +++ b/src/gateway/server-methods/sessions-rewind.ts @@ -7,6 +7,7 @@ import { validateSessionsRewindParams, } from "../../../packages/gateway-protocol/src/index.js"; import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; +import { listRegisteredAgentHarnesses } from "../../agents/harness/registry.js"; import { clearSessionQueues } from "../../auto-reply/reply/queue/cleanup.js"; import { forkSessionAtMessage, @@ -21,7 +22,10 @@ import { isCompetingSessionWorkAdmissionActive, runExclusiveSessionLifecycleMutation, } from "../../sessions/session-lifecycle-admission.js"; -import { readSessionUpstreamLink } from "../../sessions/session-upstream-links.js"; +import { + readSessionUpstreamLink, + type SessionUpstreamLink, +} from "../../sessions/session-upstream-links.js"; import { buildDashboardSessionKey, resolveRequestedSessionAgentId as resolveRequestedGlobalAgentId, @@ -42,6 +46,13 @@ type MessageCutAction = "fork" | "rewind" | "switch"; const EXTERNAL_CONVERSATION_ERROR = "Session history changes are unavailable because this session is owned by an external agent harness."; +function resolveUpstreamForkHarness(link: SessionUpstreamLink) { + const matches = listRegisteredAgentHarnesses().filter((entry) => + entry.harness.sessionFork?.upstreamKinds.includes(link.upstreamKind), + ); + return matches.length === 1 ? matches[0]?.harness.sessionFork : undefined; +} + export const sessionRewindHandlers: GatewayRequestHandlers = { "sessions.branches.list": async (options) => { if ( @@ -179,7 +190,10 @@ async function mutateSessionAtMessage( } const initialSessionId = initial.entry.sessionId; const initialLifecycleRevision = initial.entry.lifecycleRevision; - if (readSessionUpstreamLink(initial.canonicalKey, initial.target.agentId)) { + const initialUpstreamLink = readSessionUpstreamLink(initial.canonicalKey, initial.target.agentId); + // Only fork may cross to an upstream-owned conversation (it creates a new thread). + // Rewind and switch would mutate the shared upstream history in place; fail closed. + if (initialUpstreamLink && action !== "fork") { respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, EXTERNAL_CONVERSATION_ERROR)); return; } @@ -273,7 +287,8 @@ async function mutateSessionAtMessage( ); return; } - if (readSessionUpstreamLink(current.canonicalKey, current.target.agentId)) { + const upstreamLink = readSessionUpstreamLink(current.canonicalKey, current.target.agentId); + if (upstreamLink && action !== "fork") { respond( false, undefined, @@ -293,30 +308,107 @@ async function mutateSessionAtMessage( } const targetKey = action === "fork" ? buildDashboardSessionKey(current.target.agentId) : current.canonicalKey; - const result = await (action === "fork" - ? forkSessionAtMessage({ - agentId: current.target.agentId, - entryId, - sessionKey: current.canonicalKey, - sessionStoreKey: current.sessionStoreKey, - storePath: current.storePath, - targetKey, - }) - : action === "rewind" - ? rewindSessionToMessage({ + const upstreamForkHarness = upstreamLink + ? resolveUpstreamForkHarness(upstreamLink) + : undefined; + if (upstreamLink && !upstreamForkHarness) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, EXTERNAL_CONVERSATION_ERROR), + ); + return; + } + const upstreamFork = + upstreamLink && upstreamForkHarness + ? await upstreamForkHarness.fork({ + targetKey, + source: { + agentId: current.target.agentId, + sessionId: current.entry.sessionId, + sessionKey: current.canonicalKey, + storePath: current.storePath, + entryId, + }, + upstream: { + catalogId: upstreamLink.catalogId, + hostId: upstreamLink.hostId, + kind: upstreamLink.upstreamKind, + threadId: upstreamLink.threadId, + ref: upstreamLink.upstreamRef, + }, + }) + : undefined; + if (upstreamFork?.status === "failed") { + respond( + false, + undefined, + errorShape( + upstreamFork.code === "upstream-unavailable" + ? ErrorCodes.UNAVAILABLE + : ErrorCodes.INVALID_REQUEST, + upstreamFork.message, + { details: { reason: upstreamFork.code } }, + ), + ); + return; + } + if (upstreamFork?.status === "created") { + // Canonical fork lineage stays upstream. Linked sessions intentionally do not enter + // the local branch graph; branch listing/switching remains rejected for them above. + respond( + true, + { + sessionKey: upstreamFork.key, + ...(upstreamFork.editorText !== undefined + ? { editorText: upstreamFork.editorText } + : {}), + }, + undefined, + ); + emitSessionsChanged(context, { + sessionKey: upstreamFork.key, + ...(upstreamFork.key === "global" && requestedAgent.agentId + ? { agentId: requestedAgent.agentId } + : {}), + reason: "fork", + }); + return; + } + let result: SessionMessageCutMutationResult | SessionBranchSwitchMutationResult; + try { + result = await (action === "fork" + ? forkSessionAtMessage({ agentId: current.target.agentId, entryId, sessionKey: current.canonicalKey, sessionStoreKey: current.sessionStoreKey, storePath: current.storePath, + targetKey, }) - : switchSessionBranch({ - agentId: current.target.agentId, - leafEntryId: entryId, - sessionKey: current.canonicalKey, - sessionStoreKey: current.sessionStoreKey, - storePath: current.storePath, - })); + : action === "rewind" + ? rewindSessionToMessage({ + agentId: current.target.agentId, + entryId, + sessionKey: current.canonicalKey, + sessionStoreKey: current.sessionStoreKey, + storePath: current.storePath, + }) + : switchSessionBranch({ + agentId: current.target.agentId, + leafEntryId: entryId, + sessionKey: current.canonicalKey, + sessionStoreKey: current.sessionStoreKey, + storePath: current.storePath, + })); + } catch { + respond( + false, + undefined, + errorShape(ErrorCodes.UNAVAILABLE, `Failed to ${action} the local session. Try again.`), + ); + return; + } if (result.status !== "created") { respondMessageCutError(result, action, entryId, respond); return; diff --git a/src/plugin-sdk/agent-harness-runtime.ts b/src/plugin-sdk/agent-harness-runtime.ts index 938f0fe6e049..2b3a8dd4090a 100644 --- a/src/plugin-sdk/agent-harness-runtime.ts +++ b/src/plugin-sdk/agent-harness-runtime.ts @@ -51,6 +51,9 @@ export type { AgentHarnessSideQuestionParams, AgentHarnessSideQuestionResult, AgentHarnessResetParams, + AgentHarnessSessionForkFailureCode, + AgentHarnessSessionForkParams, + AgentHarnessSessionForkResult, AgentHarnessSupport, AgentHarnessSupportContext, } from "../agents/harness/types.js"; diff --git a/src/plugin-sdk/session-catalog.ts b/src/plugin-sdk/session-catalog.ts index 06cf8ddbb513..2e664230fbff 100644 --- a/src/plugin-sdk/session-catalog.ts +++ b/src/plugin-sdk/session-catalog.ts @@ -27,6 +27,10 @@ export type { SessionsCatalogReadParams, SessionsCatalogReadResult, } from "../../packages/gateway-protocol/src/schema/sessions-catalog.js"; +export { + deleteSessionUpstreamLink, + upsertSessionUpstreamLink, +} from "../sessions/session-upstream-links.js"; export { classifyClaudeCliHistoryMessage, classifyClaudeCliHistoryLine, diff --git a/src/sessions/session-upstream-links.ts b/src/sessions/session-upstream-links.ts index 25ee93b2a72f..11b33c793a9e 100644 --- a/src/sessions/session-upstream-links.ts +++ b/src/sessions/session-upstream-links.ts @@ -18,7 +18,7 @@ type SessionUpstreamDatabase = Pick< >; type SessionUpstreamLinkRow = Selectable; -type SessionUpstreamLink = { +export type SessionUpstreamLink = { sessionKey: string; agentId: string; catalogId: string; @@ -79,7 +79,7 @@ export function upsertSessionUpstreamLink( marker: SessionUpstreamJsonValue; }, options: OpenClawStateDatabaseOptions & { now?: number } = {}, -): void { +): boolean { const now = options.now ?? Date.now(); try { runOpenClawStateWriteTransaction(({ db }) => { @@ -141,8 +141,10 @@ export function upsertSessionUpstreamLink( ), ); }, options); + return true; } catch (error) { log.warn(`failed to upsert session upstream link: ${String(error)}`); + return false; } }