mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
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
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
aae5b0f041
commit
ea54060223
@@ -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 } =
|
||||
|
||||
@@ -150,8 +150,10 @@ export default definePluginEntry({
|
||||
api.registerAgentHarness(
|
||||
createCodexAppServerAgentHarness({
|
||||
bindingStore,
|
||||
sessionCatalogControl,
|
||||
resolveConfig: resolveCurrentConfig,
|
||||
resolvePluginConfig: resolveCurrentPluginConfig,
|
||||
runtime: api.runtime,
|
||||
}),
|
||||
);
|
||||
api.registerMediaUnderstandingProvider(
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<PluginRuntime["agent"]["session"]["createSessionEntry"]>
|
||||
>;
|
||||
|
||||
/** 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<CreatedCodexImportedSession> {
|
||||
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);
|
||||
}
|
||||
@@ -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<string, unknown> = {}): 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> = {}): 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<typeof resolveCodexUpstreamForkBoundary>[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);
|
||||
});
|
||||
});
|
||||
@@ -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<CodexTurn[]> {
|
||||
const turns: CodexTurn[] = [];
|
||||
const seenCursors = new Set<string>();
|
||||
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<CodexUpstreamForkBoundaryResult> {
|
||||
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 };
|
||||
}
|
||||
@@ -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<unknown>;
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<AgentHarnessSessionForkResult> {
|
||||
try {
|
||||
return await options.control.withPinnedConnection(async (control) => {
|
||||
let linked = false;
|
||||
let bindingIdentity: ReturnType<typeof sessionBindingIdentity> | 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.",
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,7 @@
|
||||
import type {
|
||||
CodexThread,
|
||||
CodexThreadForkParams,
|
||||
CodexThreadForkResponse,
|
||||
CodexThreadListParams,
|
||||
CodexThreadListResponse,
|
||||
CodexThreadTurnsListParams,
|
||||
@@ -45,6 +47,7 @@ export type CodexSessionCatalogControl = {
|
||||
listPage(params: CodexSessionCatalogPageParams): Promise<CodexSessionCatalogPage>;
|
||||
listDescendantPage(params: CodexThreadListParams): Promise<CodexThreadListResponse>;
|
||||
listTurnPage(params: CodexThreadTurnsListParams): Promise<CodexThreadTurnsListResponse>;
|
||||
forkThread(params: CodexThreadForkParams): Promise<CodexThreadForkResponse>;
|
||||
readThread(threadId: string, includeTurns?: boolean): Promise<CodexThread>;
|
||||
archiveThread(threadId: string): Promise<void>;
|
||||
};
|
||||
|
||||
@@ -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<CodexThreadListResponse>;
|
||||
listThreadTurns(params: CodexThreadTurnsListParams): Promise<CodexThreadTurnsListResponse>;
|
||||
forkThread(params: CodexThreadForkParams): Promise<CodexThreadForkResponse>;
|
||||
readThread(threadId: string, includeTurns: boolean): Promise<CodexThread>;
|
||||
archiveThread(threadId: string): Promise<void>;
|
||||
};
|
||||
@@ -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<CodexThreadForkResponse>({
|
||||
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<typeof sessionBindingIdentity> | 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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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<AgentHarnessAttemptResult["agentHarnessResultClassification"]>;
|
||||
@@ -188,6 +224,13 @@ type AgentHarnessSessionLifecycleCapability = {
|
||||
dispose?(): Promise<void> | void;
|
||||
};
|
||||
|
||||
type AgentHarnessSessionForkCapability = {
|
||||
sessionFork?: {
|
||||
upstreamKinds: readonly import("../../plugins/session-catalog.js").SessionUpstreamKind[];
|
||||
fork(params: AgentHarnessSessionForkParams): Promise<AgentHarnessSessionForkResult>;
|
||||
};
|
||||
};
|
||||
|
||||
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 = {
|
||||
|
||||
@@ -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 },
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -51,6 +51,9 @@ export type {
|
||||
AgentHarnessSideQuestionParams,
|
||||
AgentHarnessSideQuestionResult,
|
||||
AgentHarnessResetParams,
|
||||
AgentHarnessSessionForkFailureCode,
|
||||
AgentHarnessSessionForkParams,
|
||||
AgentHarnessSessionForkResult,
|
||||
AgentHarnessSupport,
|
||||
AgentHarnessSupportContext,
|
||||
} from "../agents/harness/types.js";
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -18,7 +18,7 @@ type SessionUpstreamDatabase = Pick<
|
||||
>;
|
||||
type SessionUpstreamLinkRow = Selectable<OpenClawStateKyselyDatabase["session_upstream_links"]>;
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user