From a68caa185bf15877a581a75d4bbb074b1fbd97c9 Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Tue, 7 Jul 2026 10:47:02 -0700 Subject: [PATCH] refactor(sessions): move parent-session forking behind the accessor boundary (#101699) * refactor(sessions): move parent-session forking behind the accessor boundary * fix(sessions): lazy-load fork transcript runtime to keep accessor imports lean * test(sessions): align parent fork mock shape --- .../lib/session-accessor-debt-baseline.json | 3 +- src/agents/subagent-spawn.context.test.ts | 5 - src/agents/subagent-spawn.ts | 3 - .../reply/session-fork.runtime.test.ts | 602 +---------------- src/auto-reply/reply/session-fork.runtime.ts | 231 +------ src/auto-reply/reply/session-fork.test.ts | 50 +- src/auto-reply/reply/session-fork.ts | 210 ++---- .../reply/session-parent-fork-prepare.ts | 5 +- src/auto-reply/reply/session.test.ts | 8 +- .../session-accessor.parent-fork.test.ts | 639 ++++++++++++++++++ src/config/sessions/session-accessor.ts | 283 +++++++- .../session-fork-transcript.runtime.ts | 186 +++++ src/gateway/session-create-service.ts | 7 +- 13 files changed, 1215 insertions(+), 1017 deletions(-) create mode 100644 src/config/sessions/session-accessor.parent-fork.test.ts create mode 100644 src/config/sessions/session-fork-transcript.runtime.ts diff --git a/scripts/lib/session-accessor-debt-baseline.json b/scripts/lib/session-accessor-debt-baseline.json index b63f9a0d0e28..a111c9a4fdfb 100644 --- a/scripts/lib/session-accessor-debt-baseline.json +++ b/scripts/lib/session-accessor-debt-baseline.json @@ -48,13 +48,12 @@ "src/agents/subagent-spawn.ts": 2, "src/auto-reply/reply/commands-name.ts": 2, "src/auto-reply/reply/dispatch-from-config.ts": 2, - "src/auto-reply/reply/session-fork.ts": 2, "src/commands/doctor-heartbeat-main-session-repair.ts": 2, "src/commands/doctor-session-state-providers.ts": 2, "src/commands/doctor-state-integrity.ts": 2, "src/commands/doctor/shared/codex-route-warnings.ts": 2, "src/config/sessions/plugin-host-cleanup.ts": 2, - "src/config/sessions/session-accessor.ts": 13, + "src/config/sessions/session-accessor.ts": 14, "src/config/sessions/session-file.ts": 2, "src/config/sessions/session-registry-maintenance.ts": 2, "src/gateway/server-methods/agent.ts": 2, diff --git a/src/agents/subagent-spawn.context.test.ts b/src/agents/subagent-spawn.context.test.ts index 02c6e99ef01e..07754f87b78d 100644 --- a/src/agents/subagent-spawn.context.test.ts +++ b/src/agents/subagent-spawn.context.test.ts @@ -1,4 +1,3 @@ -import path from "node:path"; import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { @@ -61,7 +60,6 @@ describe("sessions_spawn context modes", () => { fallbackEntry?: Record; parentStoreKeys?: string[]; sessionKey: string; - sessionsDir?: string; }) => { const parentEntry = params.parentStoreKeys ?.map((key) => store[key]) @@ -94,7 +92,6 @@ describe("sessions_spawn context modes", () => { const fork = await forkSessionFromParentMock({ parentEntry, agentId: params.agentId, - sessionsDir: params.sessionsDir, }); if (!fork) { return { status: "failed" }; @@ -191,7 +188,6 @@ describe("sessions_spawn context modes", () => { expect(forkSessionFromParentMock).toHaveBeenCalledWith({ parentEntry: store.main, agentId: "main", - sessionsDir: path.dirname(storePath), }); const childSessionKey = requireChildSessionKey(accepted); const childEntry = requireStoreEntry(store, childSessionKey); @@ -329,7 +325,6 @@ describe("sessions_spawn context modes", () => { expect(forkSessionFromParentMock).toHaveBeenCalledWith({ parentEntry: store.main, agentId: "main", - sessionsDir: path.dirname(storePath), }); const cleanupRequest = requireGatewayRequest("sessions.delete"); expect(cleanupRequest.params?.key).toBe(result.childSessionKey); diff --git a/src/agents/subagent-spawn.ts b/src/agents/subagent-spawn.ts index 8c5ace910277..2ce1c4cda0fa 100644 --- a/src/agents/subagent-spawn.ts +++ b/src/agents/subagent-spawn.ts @@ -5,7 +5,6 @@ */ import crypto from "node:crypto"; import { promises as fs } from "node:fs"; -import path from "node:path"; import { finiteSecondsToTimerSafeMilliseconds } from "@openclaw/normalization-core/number-coercion"; import { normalizeOptionalLowercaseString, @@ -510,7 +509,6 @@ async function prepareSubagentSessionContext(params: { let parentEntry: SessionEntry | undefined; let childEntry: SessionEntry | undefined; let forkFallbackNote: string | undefined; - const sessionsDir = path.dirname(parentTarget.storePath); try { if (params.targetAgentId !== params.requesterAgentId) { @@ -527,7 +525,6 @@ async function prepareSubagentSessionContext(params: { sessionStoreKeys: childTarget.storeKeys, fallbackEntry: { sessionId: "", updatedAt: Date.now() }, agentId: params.requesterAgentId, - sessionsDir, }); if (forkedResult.status === "missing-parent") { throw new Error( diff --git a/src/auto-reply/reply/session-fork.runtime.test.ts b/src/auto-reply/reply/session-fork.runtime.test.ts index c47362f9eb8a..e054b4f84f2e 100644 --- a/src/auto-reply/reply/session-fork.runtime.test.ts +++ b/src/auto-reply/reply/session-fork.runtime.test.ts @@ -2,14 +2,9 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import type { AssistantMessage } from "openclaw/plugin-sdk/llm"; import { afterEach, describe, expect, it } from "vitest"; -import { SessionManager } from "../../agents/sessions/session-manager.js"; import type { SessionEntry } from "../../config/sessions/types.js"; -import { - forkSessionFromParentRuntime, - resolveParentForkTokenCountRuntime, -} from "./session-fork.runtime.js"; +import { resolveParentForkTokenCountRuntime } from "./session-fork.runtime.js"; const roots: string[] = []; @@ -369,598 +364,3 @@ describe("resolveParentForkTokenCountRuntime", () => { expect(tokens).toBeLessThan(110_000); }); }); - -describe("forkSessionFromParentRuntime", () => { - it("forks the active branch without synchronously opening the session manager", async () => { - const root = await makeRoot("openclaw-parent-fork-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - const cwd = path.join(root, "workspace"); - await fs.mkdir(cwd); - const parentSessionId = "parent-session"; - const lines = [ - { - type: "session", - version: 3, - id: parentSessionId, - timestamp: "2026-05-01T00:00:00.000Z", - cwd, - }, - { - type: "message", - id: "user-1", - parentId: null, - timestamp: "2026-05-01T00:00:01.000Z", - message: { role: "user", content: "hello" }, - }, - { - type: "message", - id: "assistant-1", - parentId: "user-1", - timestamp: "2026-05-01T00:00:02.000Z", - message: { - role: "assistant", - content: [{ type: "text", text: "hi" }], - api: "openai-responses", - provider: "openai", - model: "gpt-5.4", - stopReason: "stop", - timestamp: 2, - }, - }, - { - type: "label", - id: "label-1", - parentId: "assistant-1", - timestamp: "2026-05-01T00:00:03.000Z", - targetId: "user-1", - label: "start", - }, - { - type: "message", - id: "delivery-side-branch", - parentId: "label-1", - timestamp: "2026-05-01T00:00:04.000Z", - message: { role: "assistant", content: "side delivery" }, - }, - { - type: "leaf", - id: "active-leaf", - parentId: "delivery-side-branch", - timestamp: "2026-05-01T00:00:05.000Z", - targetId: "label-1", - }, - ]; - await fs.writeFile( - parentSessionFile, - `${lines.map((entry) => JSON.stringify(entry)).join("\n")}\n`, - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: parentSessionId, - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (fork === null) { - throw new Error("Expected forked session"); - } - expect(fork.sessionFile).toContain(sessionsDir); - expect(fork.sessionId).not.toBe(parentSessionId); - const raw = await fs.readFile(fork.sessionFile, "utf-8"); - const forkedEntries = raw - .trim() - .split(/\r?\n/u) - .map((line) => JSON.parse(line) as Record); - const resolvedParentSessionFile = await fs.realpath(parentSessionFile); - const forkedHeader = forkedEntries[0]; - expect(forkedHeader?.type).toBe("session"); - expect(forkedHeader?.id).toBe(fork.sessionId); - expect(forkedHeader?.cwd).toBe(cwd); - expect(forkedHeader?.parentSession).toBe(resolvedParentSessionFile); - expect(forkedEntries.map((entry) => entry.type)).toEqual([ - "session", - "message", - "message", - "label", - "leaf", - ]); - const forkedLabel = forkedEntries.find((entry) => entry.type === "label"); - expect(forkedLabel?.type).toBe("label"); - expect(forkedLabel?.targetId).toBe("user-1"); - expect(forkedLabel?.label).toBe("start"); - expect(forkedEntries.at(-1)).toMatchObject({ - type: "leaf", - targetId: "label-1", - appendParentId: "label-1", - }); - expect(raw).not.toContain("side delivery"); - }); - - it("keeps opaque append-parent metadata on the active fork branch", async () => { - const root = await makeRoot("openclaw-parent-fork-opaque-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - const parentSessionId = "parent-opaque"; - const entries = [ - { - type: "session", - version: 3, - id: parentSessionId, - timestamp: "2026-06-15T00:00:00.000Z", - cwd: root, - }, - { - type: "message", - id: "active-root", - parentId: null, - timestamp: "2026-06-15T00:00:01.000Z", - message: { role: "assistant", content: "active root" }, - }, - { - type: "label", - id: "active-label", - parentId: "active-root", - timestamp: "2026-06-15T00:00:01.500Z", - targetId: "active-root", - label: "selected", - }, - { - type: "message", - id: "side-delivery", - parentId: "active-root", - timestamp: "2026-06-15T00:00:02.000Z", - message: { role: "assistant", content: "side delivery" }, - }, - { - type: "metadata", - id: "plugin-metadata", - parentId: "side-delivery", - payload: { source: "plugin" }, - }, - { - type: "leaf", - id: "active-leaf", - parentId: "side-delivery", - timestamp: "2026-06-15T00:00:03.000Z", - targetId: "active-root", - appendParentId: "plugin-metadata", - appendMode: "side", - }, - ]; - await fs.writeFile( - parentSessionFile, - `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`, - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: parentSessionId, - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (!fork) { - throw new Error("expected forked session"); - } - const raw = await fs.readFile(fork.sessionFile, "utf-8"); - expect(raw).toContain('"id":"active-root"'); - expect(raw).toContain('"id":"plugin-metadata"'); - expect(raw).not.toContain("side delivery"); - const forkedRecords = raw - .trim() - .split(/\r?\n/) - .map((line) => JSON.parse(line) as Record); - expect(forkedRecords.find((entry) => entry.id === "plugin-metadata")).toMatchObject({ - parentId: "active-root", - }); - expect(forkedRecords.find((entry) => entry.type === "label")).toMatchObject({ - targetId: "active-root", - label: "selected", - }); - expect(forkedRecords.at(-1)).toMatchObject({ - type: "leaf", - targetId: "active-root", - appendParentId: "plugin-metadata", - appendMode: "side", - }); - const reopened = SessionManager.open(fork.sessionFile, sessionsDir); - reopened.appendMessage({ role: "user", content: "continued", timestamp: Date.now() }); - const records = (await fs.readFile(fork.sessionFile, "utf-8")) - .trim() - .split(/\r?\n/) - .map((line) => JSON.parse(line) as Record); - expect(records.at(-1)).toMatchObject({ type: "message", parentId: "plugin-metadata" }); - expect(records.at(-1)).not.toHaveProperty("appendMode"); - expect(reopened.buildSessionContext().messages).toMatchObject([ - { role: "assistant", content: [{ type: "text", text: "active root" }] }, - { role: "user", content: "continued" }, - ]); - }); - - it("keeps parentless visible history with a disjoint append cursor", async () => { - const root = await makeRoot("openclaw-parent-fork-disjoint-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - await fs.writeFile( - parentSessionFile, - [ - { - type: "session", - version: 3, - id: "parent-disjoint", - timestamp: "2026-06-15T00:00:00.000Z", - cwd: root, - }, - { - type: "message", - id: "visible-user", - timestamp: "2026-06-15T00:00:01.000Z", - message: { role: "user", content: "visible question" }, - }, - { - type: "message", - id: "visible-assistant", - timestamp: "2026-06-15T00:00:02.000Z", - message: { role: "assistant", content: "visible answer" }, - }, - { - type: "metadata", - id: "append-root", - parentId: null, - payload: { source: "plugin" }, - }, - { - type: "leaf", - id: "active-leaf", - parentId: "append-root", - timestamp: "2026-06-15T00:00:03.000Z", - targetId: "visible-assistant", - appendParentId: "append-root", - }, - ] - .map((entry) => JSON.stringify(entry)) - .join("\n") + "\n", - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: "parent-disjoint", - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (!fork) { - throw new Error("expected forked session"); - } - const reopened = SessionManager.open(fork.sessionFile, sessionsDir); - expect(reopened.buildSessionContext().messages).toHaveLength(2); - reopened.appendMessage({ role: "user", content: "continued", timestamp: Date.now() }); - const raw = await fs.readFile(fork.sessionFile, "utf-8"); - expect(raw).toContain("visible question"); - expect(raw).toContain("visible answer"); - expect(raw).toContain('"id":"append-root"'); - const records = raw - .trim() - .split(/\r?\n/) - .map((line) => JSON.parse(line) as Record); - expect(records.at(-1)).toMatchObject({ type: "message", parentId: "append-root" }); - }); - - it("keeps an explicit empty visible branch separate from its opaque append parent", async () => { - const root = await makeRoot("openclaw-parent-fork-empty-opaque-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - await fs.writeFile( - parentSessionFile, - [ - { - type: "session", - version: 3, - id: "parent-empty-opaque", - timestamp: "2026-06-15T00:00:00.000Z", - cwd: root, - }, - { - type: "message", - id: "inactive-root", - parentId: null, - timestamp: "2026-06-15T00:00:01.000Z", - message: { role: "user", content: "inactive history" }, - }, - { - type: "leaf", - id: "empty-leaf", - parentId: "inactive-root", - timestamp: "2026-06-15T00:00:02.000Z", - targetId: null, - appendParentId: null, - }, - { - type: "metadata", - id: "plugin-metadata", - parentId: "inactive-root", - payload: { source: "plugin" }, - }, - ] - .map((entry) => JSON.stringify(entry)) - .join("\n") + "\n", - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: "parent-empty-opaque", - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (!fork) { - throw new Error("expected forked session"); - } - const reopened = SessionManager.open(fork.sessionFile, sessionsDir); - expect(reopened.buildSessionContext().messages).toEqual([]); - const continuedId = reopened.appendMessage({ - role: "user", - content: "continued", - timestamp: Date.now(), - }); - reopened.appendMessage({ - role: "assistant", - content: "done", - api: "responses", - provider: "openai", - model: "gpt-test", - timestamp: Date.now(), - } as unknown as AssistantMessage); - const records = (await fs.readFile(fork.sessionFile, "utf-8")) - .trim() - .split(/\r?\n/) - .map((line) => JSON.parse(line) as Record); - expect(records.some((record) => record.id === "inactive-root")).toBe(false); - expect(records.find((record) => record.id === continuedId)).toMatchObject({ - type: "message", - parentId: "plugin-metadata", - }); - }); - - it("keeps a reachable branch suffix when an older parent is missing", async () => { - const root = await makeRoot("openclaw-parent-fork-missing-ancestor-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - await fs.writeFile( - parentSessionFile, - [ - { - type: "session", - version: 3, - id: "parent-missing-ancestor", - timestamp: "2026-06-15T00:00:00.000Z", - cwd: root, - }, - { - type: "message", - id: "reachable-tail", - parentId: "missing-parent", - timestamp: "2026-06-15T00:00:01.000Z", - message: { role: "assistant", content: "reachable tail" }, - }, - ] - .map((entry) => JSON.stringify(entry)) - .join("\n") + "\n", - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: "parent-missing-ancestor", - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (!fork) { - throw new Error("expected forked session"); - } - const raw = await fs.readFile(fork.sessionFile, "utf-8"); - expect(raw).toContain("reachable tail"); - expect(raw).not.toContain("missing-parent"); - }); - - it("keeps visible history when the next append explicitly starts a root branch", async () => { - const root = await makeRoot("openclaw-parent-fork-root-append-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - await fs.writeFile( - parentSessionFile, - [ - { - type: "session", - version: 3, - id: "parent-root-append", - timestamp: "2026-06-15T00:00:00.000Z", - cwd: root, - }, - { - type: "message", - id: "visible-root", - parentId: null, - timestamp: "2026-06-15T00:00:01.000Z", - message: { role: "assistant", content: "visible history" }, - }, - { - type: "leaf", - id: "root-append-control", - parentId: "inactive-tail", - timestamp: "2026-06-15T00:00:02.000Z", - targetId: "visible-root", - appendParentId: null, - }, - ] - .map((entry) => JSON.stringify(entry)) - .join("\n") + "\n", - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: "parent-root-append", - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (!fork) { - throw new Error("expected forked session"); - } - const reopened = SessionManager.open(fork.sessionFile, sessionsDir); - expect(reopened.buildSessionContext().messages).toHaveLength(1); - reopened.appendMessage({ role: "user", content: "new root", timestamp: Date.now() }); - const records = (await fs.readFile(fork.sessionFile, "utf-8")) - .trim() - .split(/\r?\n/) - .map((line) => JSON.parse(line) as Record); - expect(records.at(-1)).toMatchObject({ type: "message", parentId: null }); - }); - - it("preserves supported current-version linear transcripts", async () => { - const root = await makeRoot("openclaw-parent-fork-linear-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - await fs.writeFile( - parentSessionFile, - [ - { - type: "session", - version: 3, - id: "parent-linear", - timestamp: "2026-06-15T00:00:00.000Z", - cwd: root, - }, - { - type: "message", - id: "linear-user", - timestamp: "2026-06-15T00:00:01.000Z", - message: { role: "user", content: "hello" }, - }, - { - type: "message", - id: "linear-assistant", - timestamp: "2026-06-15T00:00:02.000Z", - message: { role: "assistant", content: "hi" }, - }, - { - type: "metadata", - id: "linear-metadata", - parentId: "linear-assistant", - payload: { source: "plugin" }, - }, - ] - .map((entry) => JSON.stringify(entry)) - .join("\n") + "\n", - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: "parent-linear", - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (!fork) { - throw new Error("expected forked session"); - } - const records = (await fs.readFile(fork.sessionFile, "utf-8")) - .trim() - .split(/\r?\n/) - .map((line) => JSON.parse(line) as Record); - expect(records.slice(1)).toMatchObject([ - { id: "linear-user", parentId: null }, - { id: "linear-assistant", parentId: "linear-user" }, - { id: "linear-metadata", parentId: "linear-assistant" }, - ]); - const reopened = SessionManager.open(fork.sessionFile, sessionsDir); - expect(reopened.buildSessionContext().messages).toHaveLength(2); - reopened.appendMessage({ role: "user", content: "continued", timestamp: Date.now() }); - const continuedRecords = (await fs.readFile(fork.sessionFile, "utf-8")) - .trim() - .split(/\r?\n/) - .map((line) => JSON.parse(line) as Record); - expect(continuedRecords.at(-1)).toMatchObject({ - type: "message", - parentId: "linear-metadata", - }); - }); - - it("creates a header-only child when the parent has no entries", async () => { - const root = await makeRoot("openclaw-parent-fork-empty-"); - const sessionsDir = path.join(root, "sessions"); - await fs.mkdir(sessionsDir); - const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); - const parentSessionId = "parent-empty"; - await fs.writeFile( - parentSessionFile, - `${JSON.stringify({ - type: "session", - version: 3, - id: parentSessionId, - timestamp: "2026-05-01T00:00:00.000Z", - cwd: root, - })}\n`, - "utf-8", - ); - - const fork = await forkSessionFromParentRuntime({ - parentEntry: { - sessionId: parentSessionId, - sessionFile: parentSessionFile, - updatedAt: Date.now(), - }, - agentId: "main", - sessionsDir, - }); - - if (!fork) { - throw new Error("expected forked session entry"); - } - const raw = await fs.readFile(fork.sessionFile, "utf-8"); - const lines = raw.trim().split(/\r?\n/u); - expect(lines).toHaveLength(1); - const resolvedParentSessionFile = await fs.realpath(parentSessionFile); - const header = JSON.parse(lines[0] ?? "{}") as Record; - expect(header.type).toBe("session"); - expect(header.id).toBe(fork.sessionId); - expect(header.parentSession).toBe(resolvedParentSessionFile); - }); -}); diff --git a/src/auto-reply/reply/session-fork.runtime.ts b/src/auto-reply/reply/session-fork.runtime.ts index 6f70fdf93ed1..769625f6d324 100644 --- a/src/auto-reply/reply/session-fork.runtime.ts +++ b/src/auto-reply/reply/session-fork.runtime.ts @@ -1,43 +1,19 @@ -/** Runtime implementation for forking sessions from parent transcripts. */ -import crypto from "node:crypto"; +/** + * Lazy runtime seam for parent-fork token counting. File-era transcript tail + * reads flow through gateway fs helpers; the SQLite flip estimates parent + * tokens inside the storage boundary instead. + */ import fs from "node:fs/promises"; -import path from "node:path"; -import { isRecord } from "@openclaw/normalization-core/record-coerce"; -import { - migrateSessionEntries, - parseSessionEntries, - type SessionEntry as AgentSessionEntry, - type SessionHeader, -} from "../../agents/sessions/session-manager.js"; import { derivePromptTokens } from "../../agents/usage.js"; import { resolveSessionFilePath, resolveSessionFilePathOptions, } from "../../config/sessions/paths.js"; -import { createForkedSessionTranscript } from "../../config/sessions/session-accessor.js"; -import { - isSessionTranscriptLeafControl, - mergeSessionTranscriptVisiblePathWithOpaqueAppendPath, - scanSessionTranscriptTree, - selectSessionTranscriptTreePathNodes, -} from "../../config/sessions/transcript-tree.js"; import { resolveFreshSessionTotalTokens, type SessionEntry as StoreSessionEntry, } from "../../config/sessions/types.js"; import { readLatestRecentSessionUsageFromTranscriptAsync } from "../../gateway/session-utils.fs.js"; -import { readRegularFile } from "../../infra/fs-safe.js"; - -type ForkSourceTranscript = { - cwd: string; - sessionDir: string; - leafId: string | null; - appendParentId: string | null; - appendMode?: "side"; - preserveLeafControl: boolean; - branchEntries: unknown[]; - labelsToWrite: Array<{ targetId: string; label: string; timestamp: string }>; -}; const FALLBACK_TRANSCRIPT_BYTES_PER_TOKEN = 4; @@ -126,200 +102,3 @@ export async function resolveParentForkTokenCountRuntime(params: { return maxPositiveTokenCount(cachedTokens, byteEstimateTokens); } - -function generateEntryId(existingIds: Set): string { - for (let attempt = 0; attempt < 100; attempt += 1) { - const id = crypto.randomUUID().slice(0, 8); - if (!existingIds.has(id)) { - existingIds.add(id); - return id; - } - } - const id = crypto.randomUUID(); - existingIds.add(id); - return id; -} - -function hasAssistantEntry(entries: unknown[]): boolean { - return entries.some( - (entry) => - isRecord(entry) && - entry.type === "message" && - isRecord(entry.message) && - entry.message.role === "assistant", - ); -} - -function collectBranchLabels(params: { - allEntries: unknown[]; - pathEntryIds: Set; -}): Array<{ targetId: string; label: string; timestamp: string }> { - const labelsToWrite: Array<{ targetId: string; label: string; timestamp: string }> = []; - for (const entry of params.allEntries) { - if (!isRecord(entry)) { - continue; - } - if ( - entry.type === "label" && - typeof entry.label === "string" && - typeof entry.targetId === "string" && - typeof entry.id === "string" && - !params.pathEntryIds.has(entry.id) && - params.pathEntryIds.has(entry.targetId) && - typeof entry.timestamp === "string" - ) { - labelsToWrite.push({ - targetId: entry.targetId, - label: entry.label, - timestamp: entry.timestamp, - }); - } - } - return labelsToWrite; -} - -async function readForkSourceTranscript( - parentSessionFile: string, -): Promise { - const raw = (await readRegularFile({ filePath: parentSessionFile })).buffer.toString("utf-8"); - const fileEntries = parseSessionEntries(raw); - migrateSessionEntries(fileEntries); - const header = - fileEntries.find((entry): entry is SessionHeader => entry.type === "session") ?? null; - const entries = fileEntries.filter((entry) => entry.type !== "session"); - const tree = scanSessionTranscriptTree(entries); - const leafId = tree.leafId; - const appendParentId = tree.appendParentId; - const visiblePath = selectSessionTranscriptTreePathNodes(tree, leafId); - const appendPath = selectSessionTranscriptTreePathNodes(tree, appendParentId); - const mergedPath = mergeSessionTranscriptVisiblePathWithOpaqueAppendPath({ - visiblePath, - appendPath, - appendParentId, - }); - const branchEntries = mergedPath.nodes.flatMap((node) => { - if (!isRecord(node.entry)) { - return []; - } - const parentId = node.selectedParentId; - return [node.entry.parentId === parentId ? node.entry : { ...node.entry, parentId }]; - }); - const pathEntryIds = new Set( - branchEntries.flatMap((entry) => - isRecord(entry) && typeof entry.id === "string" ? [entry.id] : [], - ), - ); - const lastLeafUpdateNode = tree.nodes.findLast((node) => node.leafId !== undefined); - const lastLeafUpdateEntry = lastLeafUpdateNode?.entry; - return { - cwd: header?.cwd ?? process.cwd(), - sessionDir: path.dirname(parentSessionFile), - leafId, - appendParentId: mergedPath.appendParentId, - ...(lastLeafUpdateNode?.appendMode ? { appendMode: lastLeafUpdateNode.appendMode } : {}), - preserveLeafControl: isSessionTranscriptLeafControl(lastLeafUpdateEntry), - branchEntries, - labelsToWrite: collectBranchLabels({ allEntries: entries, pathEntryIds }), - }; -} - -function buildBranchLabelEntries(params: { - labelsToWrite: Array<{ targetId: string; label: string; timestamp: string }>; - pathEntryIds: Set; - lastEntryId: string | null; -}): AgentSessionEntry[] { - let parentId = params.lastEntryId; - const labelEntries: AgentSessionEntry[] = []; - for (const { targetId, label, timestamp } of params.labelsToWrite) { - const labelEntry = { - type: "label", - id: generateEntryId(params.pathEntryIds), - parentId, - timestamp, - targetId, - label, - } satisfies AgentSessionEntry; - params.pathEntryIds.add(labelEntry.id); - labelEntries.push(labelEntry); - parentId = labelEntry.id; - } - return labelEntries; -} - -async function writeBranchedSession(params: { - parentSessionFile: string; - source: ForkSourceTranscript; - sessionDir: string; -}): Promise<{ sessionId: string; sessionFile: string }> { - const pathEntries = params.source.branchEntries; - const pathEntryIds = new Set( - pathEntries.flatMap((entry) => - isRecord(entry) && typeof entry.id === "string" ? [entry.id] : [], - ), - ); - const lastPathEntry = pathEntries.at(-1); - const lastPathEntryId = - isRecord(lastPathEntry) && typeof lastPathEntry.id === "string" ? lastPathEntry.id : null; - const labelEntries = buildBranchLabelEntries({ - labelsToWrite: params.source.labelsToWrite, - pathEntryIds, - lastEntryId: lastPathEntryId, - }); - return await createForkedSessionTranscript({ - cwd: params.source.cwd, - parentSessionFile: params.parentSessionFile, - sessionsDir: params.sessionDir, - buildEntries: ({ timestamp }) => { - // The leaf-control record shares the fork header timestamp so the copied - // branch reopens on the same active leaf the parent displayed. - const leafEntry = params.source.preserveLeafControl - ? { - type: "leaf", - id: generateEntryId(pathEntryIds), - parentId: labelEntries.at(-1)?.id ?? lastPathEntryId, - timestamp, - targetId: params.source.leafId, - appendParentId: params.source.appendParentId, - ...(params.source.appendMode ? { appendMode: params.source.appendMode } : {}), - } - : null; - return [...pathEntries, ...labelEntries, ...(leafEntry ? [leafEntry] : [])]; - }, - }); -} - -/** Creates a child session transcript from a parent session branch. */ -export async function forkSessionFromParentRuntime(params: { - parentEntry: StoreSessionEntry; - agentId: string; - sessionsDir: string; - targetSessionsDir?: string; -}): Promise<{ sessionId: string; sessionFile: string } | null> { - const parentSessionFile = resolveSessionFilePath( - params.parentEntry.sessionId, - params.parentEntry, - { agentId: params.agentId, sessionsDir: params.sessionsDir }, - ); - if (!parentSessionFile) { - return null; - } - try { - const source = await readForkSourceTranscript(parentSessionFile); - if (!source) { - return null; - } - const shouldPersistBranch = - source.preserveLeafControl || hasAssistantEntry(source.branchEntries); - const targetSessionsDir = params.targetSessionsDir ?? source.sessionDir; - return shouldPersistBranch - ? await writeBranchedSession({ parentSessionFile, source, sessionDir: targetSessionsDir }) - : // Header-only fork: nothing on the active branch is worth copying. - await createForkedSessionTranscript({ - cwd: source.cwd, - parentSessionFile, - sessionsDir: targetSessionsDir, - }); - } catch { - return null; - } -} diff --git a/src/auto-reply/reply/session-fork.test.ts b/src/auto-reply/reply/session-fork.test.ts index 00c05e4aa959..af94a452be40 100644 --- a/src/auto-reply/reply/session-fork.test.ts +++ b/src/auto-reply/reply/session-fork.test.ts @@ -7,7 +7,6 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { forkSessionEntryFromParent } from "./session-fork.js"; const runtimeMocks = vi.hoisted(() => ({ - forkSessionFromParentRuntime: vi.fn(), resolveParentForkTokenCountRuntime: vi.fn(), })); @@ -16,13 +15,14 @@ vi.mock("./session-fork.runtime.js", () => runtimeMocks); const roots: string[] = []; async function makeRoot(prefix: string): Promise { - const root = await fs.mkdtemp(path.join(os.tmpdir(), prefix)); + // realpath first: macOS tmpdir is a /var -> /private/var symlink and the + // fork resolver returns canonical paths. + const root = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), prefix))); roots.push(root); return root; } afterEach(async () => { - runtimeMocks.forkSessionFromParentRuntime.mockReset(); runtimeMocks.resolveParentForkTokenCountRuntime.mockReset(); await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true }))); }); @@ -38,13 +38,34 @@ describe("forkSessionEntryFromParent", () => { const configStorePath = path.join(configStoreDir, "sessions.json"); const parentSessionKey = "agent:main:main"; const sessionKey = "agent:main:subagent:child"; + const parentSessionFile = path.join(activeStoreDir, "parent.jsonl"); + await fs.writeFile( + parentSessionFile, + [ + JSON.stringify({ + type: "session", + version: 3, + id: "parent-session", + timestamp: "2026-05-01T00:00:00.000Z", + cwd: root, + }), + JSON.stringify({ + type: "message", + id: "assistant-1", + parentId: null, + timestamp: "2026-05-01T00:00:01.000Z", + message: { role: "assistant", content: "hi" }, + }), + ].join("\n") + "\n", + "utf-8", + ); await fs.writeFile( storePath, JSON.stringify( { [parentSessionKey]: { sessionId: "parent-session", - sessionFile: path.join(activeStoreDir, "parent.jsonl"), + sessionFile: parentSessionFile, updatedAt: 1, }, [sessionKey]: { sessionId: "", updatedAt: 2 }, @@ -56,12 +77,6 @@ describe("forkSessionEntryFromParent", () => { ); runtimeMocks.resolveParentForkTokenCountRuntime.mockResolvedValue(10); - runtimeMocks.forkSessionFromParentRuntime.mockImplementation( - async ({ sessionsDir }: { sessionsDir: string }) => ({ - sessionId: "forked-session", - sessionFile: path.join(sessionsDir, "forked-session.jsonl"), - }), - ); const result = await forkSessionEntryFromParent({ agentId: "main", @@ -73,15 +88,16 @@ describe("forkSessionEntryFromParent", () => { }); expect(result.status).toBe("forked"); - expect(runtimeMocks.forkSessionFromParentRuntime).toHaveBeenCalledWith( - expect.objectContaining({ - sessionsDir: activeStoreDir, - }), - ); + if (result.status !== "forked") { + throw new Error("expected forked result"); + } + // The fork artifact lands beside the store being mutated, not the config store. + expect(path.dirname(result.fork.sessionFile)).toBe(activeStoreDir); const stored = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record< string, - { sessionFile?: string } + { sessionId?: string; sessionFile?: string } >; - expect(stored[sessionKey]?.sessionFile).toBe(path.join(activeStoreDir, "forked-session.jsonl")); + expect(stored[sessionKey]?.sessionId).toBe(result.fork.sessionId); + expect(stored[sessionKey]?.sessionFile).toBe(result.fork.sessionFile); }); }); diff --git a/src/auto-reply/reply/session-fork.ts b/src/auto-reply/reply/session-fork.ts index 2ab58378952b..334f14805085 100644 --- a/src/auto-reply/reply/session-fork.ts +++ b/src/auto-reply/reply/session-fork.ts @@ -1,7 +1,11 @@ -import path from "node:path"; import { resolveStorePath } from "../../config/sessions/paths.js"; -import { updateSessionStore } from "../../config/sessions/store.js"; -import { mergeSessionEntry, type SessionEntry } from "../../config/sessions/types.js"; +import { + forkSessionEntryFromParentTarget, + forkSessionFromParentTranscript, + type ParentForkedSessionTranscript, + type SessionParentForkDecision, +} from "../../config/sessions/session-accessor.js"; +import type { SessionEntry } from "../../config/sessions/types.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { createLazyImportLoader } from "../../shared/lazy-promise.js"; @@ -13,19 +17,7 @@ import { createLazyImportLoader } from "../../shared/lazy-promise.js"; const DEFAULT_PARENT_FORK_MAX_TOKENS = 100_000; const sessionForkRuntimeLoader = createLazyImportLoader(() => import("./session-fork.runtime.js")); -export type ParentForkDecision = - | { - status: "fork"; - maxTokens: number; - parentTokens?: number; - } - | { - status: "skip"; - reason: "parent-too-large"; - maxTokens: number; - parentTokens: number; - message: string; - }; +export type ParentForkDecision = SessionParentForkDecision; type ParentForkDecisionParams = { parentEntry: SessionEntry; @@ -35,17 +27,17 @@ type ParentForkDecisionParams = { }; type ForkSessionFromParentParams = { + parentSessionKey: string; parentEntry: SessionEntry; agentId: string; config?: OpenClawConfig; - sessionsDir?: string; - targetSessionsDir?: string; + sessionKey: string; + storePath?: string; + /** Cross-agent forks land the child transcript beside the child's store. */ + targetStorePath?: string; }; -export type ForkedParentSessionEntry = { - sessionId: string; - sessionFile: string; -}; +export type ForkedParentSessionEntry = ParentForkedSessionTranscript; export type ForkSessionEntryFromParentResult = | { @@ -112,14 +104,6 @@ function resolveParentForkStorePath(params: { ); } -function resolveParentForkSessionsDir(params: { - agentId: string; - config?: OpenClawConfig; - sessionsDir?: string; -}): string { - return params.sessionsDir ?? path.dirname(resolveParentForkStorePath(params)); -} - export async function resolveParentForkDecision( params: ParentForkDecisionParams, ): Promise { @@ -147,41 +131,34 @@ export async function resolveParentForkDecision( export async function forkSessionFromParent( params: ForkSessionFromParentParams, ): Promise<{ sessionId: string; sessionFile: string } | null> { - const runtime = await loadSessionForkRuntime(); - return runtime.forkSessionFromParentRuntime({ - ...params, - sessionsDir: resolveParentForkSessionsDir(params), + const storePath = resolveParentForkStorePath(params); + const fork = await forkSessionFromParentTranscript({ + agentId: params.agentId, + parentEntry: params.parentEntry, + parentSessionKey: params.parentSessionKey, + sessionKey: params.sessionKey, + storePath, + ...(params.targetStorePath ? { targetStorePath: params.targetStorePath } : {}), }); + return fork.status === "created" ? fork.transcript : null; } -function resolveEntryFromStoreKeys(params: { - store: Record; - keys: readonly string[]; -}): SessionEntry | undefined { - for (const key of params.keys) { - const entry = params.store[key]; - if (entry) { - return entry; +function normalizeForkTarget(params: { canonicalKey: string; storeKeys?: readonly string[] }): { + canonicalKey: string; + storeKeys: string[]; +} { + const keys = new Set(); + const remember = (value: string) => { + const trimmed = value.trim(); + if (trimmed) { + keys.add(trimmed); } + }; + remember(params.canonicalKey); + for (const key of params.storeKeys ?? []) { + remember(key); } - return undefined; -} - -function persistForkedSessionEntry(params: { - store: Record; - sessionKey: string; - sessionStoreKeys?: readonly string[]; - existing: SessionEntry; - patch: Partial; -}): SessionEntry { - const next = mergeSessionEntry(params.existing, params.patch); - params.store[params.sessionKey] = next; - for (const key of params.sessionStoreKeys ?? []) { - if (key !== params.sessionKey) { - delete params.store[key]; - } - } - return next; + return { canonicalKey: params.canonicalKey, storeKeys: [...keys] }; } /** @@ -192,103 +169,30 @@ export async function forkSessionEntryFromParent( params: ForkSessionEntryFromParentParams, ): Promise { const storePath = resolveParentForkStorePath(params); - return await updateSessionStore( - storePath, - async (store) => { - const parentEntry = resolveEntryFromStoreKeys({ - store, - keys: params.parentStoreKeys ?? [params.parentSessionKey], - }); - if (!parentEntry?.sessionId) { - return { status: "missing-parent" }; - } - - const entry = - resolveEntryFromStoreKeys({ - store, - keys: params.sessionStoreKeys ?? [params.sessionKey], - }) ?? params.fallbackEntry; - if (!entry) { - return { status: "missing-entry" }; - } - - if (params.skipForkWhen?.(entry)) { - const patch = params.skipPatch?.(entry); - const sessionEntry = patch - ? persistForkedSessionEntry({ - store, - sessionKey: params.sessionKey, - sessionStoreKeys: params.sessionStoreKeys, - existing: entry, - patch, - }) - : entry; - return { status: "skipped", reason: "existing-entry", parentEntry, sessionEntry }; - } - - const decision = await resolveParentForkDecision({ + return await forkSessionEntryFromParentTarget({ + agentId: params.agentId, + decisionSkipPatch: params.decisionSkipPatch, + fallbackEntry: params.fallbackEntry, + parentTarget: normalizeForkTarget({ + canonicalKey: params.parentSessionKey, + storeKeys: params.parentStoreKeys, + }), + patch: params.patch, + resolveDecision: (parentEntry) => + resolveParentForkDecision({ parentEntry, agentId: params.agentId, config: params.config, storePath, - }); - if (decision.status === "skip") { - const patch = params.decisionSkipPatch?.({ decision, entry, parentEntry }); - const sessionEntry = patch - ? persistForkedSessionEntry({ - store, - sessionKey: params.sessionKey, - sessionStoreKeys: params.sessionStoreKeys, - existing: entry, - patch, - }) - : entry; - return { - status: "skipped", - reason: "decision-skip", - parentEntry, - sessionEntry, - decision, - }; - } - - const fork = await forkSessionFromParent({ - parentEntry, - agentId: params.agentId, - config: params.config, - sessionsDir: params.sessionsDir ?? path.dirname(storePath), - }); - if (!fork) { - return { status: "failed" }; - } - const sessionEntry = persistForkedSessionEntry({ - store, - sessionKey: params.sessionKey, - sessionStoreKeys: params.sessionStoreKeys, - existing: entry, - patch: { - ...params.patch?.({ entry, parentEntry, fork, decision }), - sessionId: fork.sessionId, - sessionFile: fork.sessionFile, - forkedFromParent: true, - }, - }); - return { - status: "forked", - fork, - parentEntry, - sessionEntry, - decision, - }; - }, - { - skipSaveWhenResult: (result) => - result.status === "missing-entry" || - result.status === "missing-parent" || - result.status === "failed" || - (result.status === "skipped" && result.sessionEntry === params.fallbackEntry), - }, - ); + }), + sessionTarget: normalizeForkTarget({ + canonicalKey: params.sessionKey, + storeKeys: params.sessionStoreKeys, + }), + skipForkWhen: params.skipForkWhen, + skipPatch: params.skipPatch, + storePath, + }); } async function resolveParentForkTokenCount(params: { diff --git a/src/auto-reply/reply/session-parent-fork-prepare.ts b/src/auto-reply/reply/session-parent-fork-prepare.ts index ff06b5b8df11..3543f6639220 100644 --- a/src/auto-reply/reply/session-parent-fork-prepare.ts +++ b/src/auto-reply/reply/session-parent-fork-prepare.ts @@ -1,5 +1,4 @@ // Prepares parent-context fork metadata for guarded reply session initialization. -import path from "node:path"; import type { SessionEntry } from "../../config/sessions.js"; import { forkSessionFromParent, resolveParentForkDecision } from "./session-fork.js"; @@ -41,7 +40,9 @@ export async function prepareReplySessionParentFork(params: { const fork = await forkSessionFromParent({ parentEntry, agentId: params.agentId, - sessionsDir: path.dirname(params.storePath), + parentSessionKey: params.parentSessionKey, + sessionKey: params.sessionKey, + storePath: params.storePath, }); if (!fork) { return params.sessionEntry; diff --git a/src/auto-reply/reply/session.test.ts b/src/auto-reply/reply/session.test.ts index 131fe734e1f0..938fba154259 100644 --- a/src/auto-reply/reply/session.test.ts +++ b/src/auto-reply/reply/session.test.ts @@ -54,7 +54,7 @@ const browserMaintenanceMocks = vi.hoisted(() => ({ type ForkSessionParamsForTest = { parentEntry: SessionEntry; - sessionsDir: string; + storePath: string; }; vi.mock("./session-fork.js", () => ({ @@ -79,7 +79,6 @@ vi.mock("./session-fork.js", () => ({ entry: SessionEntry; parentEntry: SessionEntry; }) => Partial; - sessionsDir: string; }) => { const store = JSON.parse(await fs.readFile(params.storePath, "utf-8")) as Record< string, @@ -116,7 +115,7 @@ vi.mock("./session-fork.js", () => ({ } const fork = await sessionForkMocks.forkSessionFromParent({ parentEntry, - sessionsDir: params.sessionsDir, + storePath: params.storePath, }); if (!fork) { return { status: "failed" }; @@ -450,10 +449,11 @@ beforeEach(() => { }); sessionForkMocks.forkSessionFromParent .mockReset() - .mockImplementation(async ({ parentEntry, sessionsDir }: ForkSessionParamsForTest) => { + .mockImplementation(async ({ parentEntry, storePath }: ForkSessionParamsForTest) => { if (!parentEntry.sessionFile) { return null; } + const sessionsDir = path.dirname(storePath); await fs.mkdir(sessionsDir, { recursive: true }); const sessionId = `forked-session-${++sessionForkMocks.nextSessionId}`; const sessionFile = path.join(sessionsDir, `${sessionId}.jsonl`); diff --git a/src/config/sessions/session-accessor.parent-fork.test.ts b/src/config/sessions/session-accessor.parent-fork.test.ts new file mode 100644 index 000000000000..4da91d2e7411 --- /dev/null +++ b/src/config/sessions/session-accessor.parent-fork.test.ts @@ -0,0 +1,639 @@ +// Behavior tests for the accessor parent-fork transcript boundary. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { AssistantMessage } from "openclaw/plugin-sdk/llm"; +import { afterEach, describe, expect, it } from "vitest"; +import { SessionManager } from "../../agents/sessions/session-manager.js"; +import { forkSessionFromParentTranscript } from "./session-accessor.js"; + +const roots: string[] = []; + +async function makeRoot(prefix: string): Promise { + const root = await fs.mkdtemp(path.join(os.tmpdir(), prefix)); + roots.push(root); + return root; +} + +afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true }))); +}); + +describe("forkSessionFromParentTranscript", () => { + it("forks the active branch without synchronously opening the session manager", async () => { + const root = await makeRoot("openclaw-parent-fork-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + const cwd = path.join(root, "workspace"); + await fs.mkdir(cwd); + const parentSessionId = "parent-session"; + const lines = [ + { + type: "session", + version: 3, + id: parentSessionId, + timestamp: "2026-05-01T00:00:00.000Z", + cwd, + }, + { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-05-01T00:00:01.000Z", + message: { role: "user", content: "hello" }, + }, + { + type: "message", + id: "assistant-1", + parentId: "user-1", + timestamp: "2026-05-01T00:00:02.000Z", + message: { + role: "assistant", + content: [{ type: "text", text: "hi" }], + api: "openai-responses", + provider: "openai", + model: "gpt-5.4", + stopReason: "stop", + timestamp: 2, + }, + }, + { + type: "label", + id: "label-1", + parentId: "assistant-1", + timestamp: "2026-05-01T00:00:03.000Z", + targetId: "user-1", + label: "start", + }, + { + type: "message", + id: "delivery-side-branch", + parentId: "label-1", + timestamp: "2026-05-01T00:00:04.000Z", + message: { role: "assistant", content: "side delivery" }, + }, + { + type: "leaf", + id: "active-leaf", + parentId: "delivery-side-branch", + timestamp: "2026-05-01T00:00:05.000Z", + targetId: "label-1", + }, + ]; + await fs.writeFile( + parentSessionFile, + `${lines.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: parentSessionId, + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("Expected forked session"); + } + const fork = forked.transcript; + expect(fork.sessionFile).toContain(sessionsDir); + expect(fork.sessionId).not.toBe(parentSessionId); + const raw = await fs.readFile(fork.sessionFile, "utf-8"); + const forkedEntries = raw + .trim() + .split(/\r?\n/u) + .map((line) => JSON.parse(line) as Record); + const resolvedParentSessionFile = await fs.realpath(parentSessionFile); + const forkedHeader = forkedEntries[0]; + expect(forkedHeader?.type).toBe("session"); + expect(forkedHeader?.id).toBe(fork.sessionId); + expect(forkedHeader?.cwd).toBe(cwd); + expect(forkedHeader?.parentSession).toBe(resolvedParentSessionFile); + expect(forkedEntries.map((entry) => entry.type)).toEqual([ + "session", + "message", + "message", + "label", + "leaf", + ]); + const forkedLabel = forkedEntries.find((entry) => entry.type === "label"); + expect(forkedLabel?.type).toBe("label"); + expect(forkedLabel?.targetId).toBe("user-1"); + expect(forkedLabel?.label).toBe("start"); + expect(forkedEntries.at(-1)).toMatchObject({ + type: "leaf", + targetId: "label-1", + appendParentId: "label-1", + }); + expect(raw).not.toContain("side delivery"); + }); + + it("keeps opaque append-parent metadata on the active fork branch", async () => { + const root = await makeRoot("openclaw-parent-fork-opaque-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + const parentSessionId = "parent-opaque"; + const entries = [ + { + type: "session", + version: 3, + id: parentSessionId, + timestamp: "2026-06-15T00:00:00.000Z", + cwd: root, + }, + { + type: "message", + id: "active-root", + parentId: null, + timestamp: "2026-06-15T00:00:01.000Z", + message: { role: "assistant", content: "active root" }, + }, + { + type: "label", + id: "active-label", + parentId: "active-root", + timestamp: "2026-06-15T00:00:01.500Z", + targetId: "active-root", + label: "selected", + }, + { + type: "message", + id: "side-delivery", + parentId: "active-root", + timestamp: "2026-06-15T00:00:02.000Z", + message: { role: "assistant", content: "side delivery" }, + }, + { + type: "metadata", + id: "plugin-metadata", + parentId: "side-delivery", + payload: { source: "plugin" }, + }, + { + type: "leaf", + id: "active-leaf", + parentId: "side-delivery", + timestamp: "2026-06-15T00:00:03.000Z", + targetId: "active-root", + appendParentId: "plugin-metadata", + appendMode: "side", + }, + ]; + await fs.writeFile( + parentSessionFile, + `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: parentSessionId, + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("expected forked session"); + } + const fork = forked.transcript; + const raw = await fs.readFile(fork.sessionFile, "utf-8"); + expect(raw).toContain('"id":"active-root"'); + expect(raw).toContain('"id":"plugin-metadata"'); + expect(raw).not.toContain("side delivery"); + const forkedRecords = raw + .trim() + .split(/\r?\n/) + .map((line) => JSON.parse(line) as Record); + expect(forkedRecords.find((entry) => entry.id === "plugin-metadata")).toMatchObject({ + parentId: "active-root", + }); + expect(forkedRecords.find((entry) => entry.type === "label")).toMatchObject({ + targetId: "active-root", + label: "selected", + }); + expect(forkedRecords.at(-1)).toMatchObject({ + type: "leaf", + targetId: "active-root", + appendParentId: "plugin-metadata", + appendMode: "side", + }); + const reopened = SessionManager.open(fork.sessionFile, sessionsDir); + reopened.appendMessage({ role: "user", content: "continued", timestamp: Date.now() }); + const records = (await fs.readFile(fork.sessionFile, "utf-8")) + .trim() + .split(/\r?\n/) + .map((line) => JSON.parse(line) as Record); + expect(records.at(-1)).toMatchObject({ type: "message", parentId: "plugin-metadata" }); + expect(records.at(-1)).not.toHaveProperty("appendMode"); + expect(reopened.buildSessionContext().messages).toMatchObject([ + { role: "assistant", content: [{ type: "text", text: "active root" }] }, + { role: "user", content: "continued" }, + ]); + }); + + it("keeps parentless visible history with a disjoint append cursor", async () => { + const root = await makeRoot("openclaw-parent-fork-disjoint-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + await fs.writeFile( + parentSessionFile, + [ + { + type: "session", + version: 3, + id: "parent-disjoint", + timestamp: "2026-06-15T00:00:00.000Z", + cwd: root, + }, + { + type: "message", + id: "visible-user", + timestamp: "2026-06-15T00:00:01.000Z", + message: { role: "user", content: "visible question" }, + }, + { + type: "message", + id: "visible-assistant", + timestamp: "2026-06-15T00:00:02.000Z", + message: { role: "assistant", content: "visible answer" }, + }, + { + type: "metadata", + id: "append-root", + parentId: null, + payload: { source: "plugin" }, + }, + { + type: "leaf", + id: "active-leaf", + parentId: "append-root", + timestamp: "2026-06-15T00:00:03.000Z", + targetId: "visible-assistant", + appendParentId: "append-root", + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n") + "\n", + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: "parent-disjoint", + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("expected forked session"); + } + const fork = forked.transcript; + const reopened = SessionManager.open(fork.sessionFile, sessionsDir); + expect(reopened.buildSessionContext().messages).toHaveLength(2); + reopened.appendMessage({ role: "user", content: "continued", timestamp: Date.now() }); + const raw = await fs.readFile(fork.sessionFile, "utf-8"); + expect(raw).toContain("visible question"); + expect(raw).toContain("visible answer"); + expect(raw).toContain('"id":"append-root"'); + const records = raw + .trim() + .split(/\r?\n/) + .map((line) => JSON.parse(line) as Record); + expect(records.at(-1)).toMatchObject({ type: "message", parentId: "append-root" }); + }); + + it("keeps an explicit empty visible branch separate from its opaque append parent", async () => { + const root = await makeRoot("openclaw-parent-fork-empty-opaque-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + await fs.writeFile( + parentSessionFile, + [ + { + type: "session", + version: 3, + id: "parent-empty-opaque", + timestamp: "2026-06-15T00:00:00.000Z", + cwd: root, + }, + { + type: "message", + id: "inactive-root", + parentId: null, + timestamp: "2026-06-15T00:00:01.000Z", + message: { role: "user", content: "inactive history" }, + }, + { + type: "leaf", + id: "empty-leaf", + parentId: "inactive-root", + timestamp: "2026-06-15T00:00:02.000Z", + targetId: null, + appendParentId: null, + }, + { + type: "metadata", + id: "plugin-metadata", + parentId: "inactive-root", + payload: { source: "plugin" }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n") + "\n", + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: "parent-empty-opaque", + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("expected forked session"); + } + const fork = forked.transcript; + const reopened = SessionManager.open(fork.sessionFile, sessionsDir); + expect(reopened.buildSessionContext().messages).toEqual([]); + const continuedId = reopened.appendMessage({ + role: "user", + content: "continued", + timestamp: Date.now(), + }); + reopened.appendMessage({ + role: "assistant", + content: "done", + api: "responses", + provider: "openai", + model: "gpt-test", + timestamp: Date.now(), + } as unknown as AssistantMessage); + const records = (await fs.readFile(fork.sessionFile, "utf-8")) + .trim() + .split(/\r?\n/) + .map((line) => JSON.parse(line) as Record); + expect(records.some((record) => record.id === "inactive-root")).toBe(false); + expect(records.find((record) => record.id === continuedId)).toMatchObject({ + type: "message", + parentId: "plugin-metadata", + }); + }); + + it("keeps a reachable branch suffix when an older parent is missing", async () => { + const root = await makeRoot("openclaw-parent-fork-missing-ancestor-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + await fs.writeFile( + parentSessionFile, + [ + { + type: "session", + version: 3, + id: "parent-missing-ancestor", + timestamp: "2026-06-15T00:00:00.000Z", + cwd: root, + }, + { + type: "message", + id: "reachable-tail", + parentId: "missing-parent", + timestamp: "2026-06-15T00:00:01.000Z", + message: { role: "assistant", content: "reachable tail" }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n") + "\n", + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: "parent-missing-ancestor", + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("expected forked session"); + } + const fork = forked.transcript; + const raw = await fs.readFile(fork.sessionFile, "utf-8"); + expect(raw).toContain("reachable tail"); + expect(raw).not.toContain("missing-parent"); + }); + + it("keeps visible history when the next append explicitly starts a root branch", async () => { + const root = await makeRoot("openclaw-parent-fork-root-append-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + await fs.writeFile( + parentSessionFile, + [ + { + type: "session", + version: 3, + id: "parent-root-append", + timestamp: "2026-06-15T00:00:00.000Z", + cwd: root, + }, + { + type: "message", + id: "visible-root", + parentId: null, + timestamp: "2026-06-15T00:00:01.000Z", + message: { role: "assistant", content: "visible history" }, + }, + { + type: "leaf", + id: "root-append-control", + parentId: "inactive-tail", + timestamp: "2026-06-15T00:00:02.000Z", + targetId: "visible-root", + appendParentId: null, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n") + "\n", + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: "parent-root-append", + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("expected forked session"); + } + const fork = forked.transcript; + const reopened = SessionManager.open(fork.sessionFile, sessionsDir); + expect(reopened.buildSessionContext().messages).toHaveLength(1); + reopened.appendMessage({ role: "user", content: "new root", timestamp: Date.now() }); + const records = (await fs.readFile(fork.sessionFile, "utf-8")) + .trim() + .split(/\r?\n/) + .map((line) => JSON.parse(line) as Record); + expect(records.at(-1)).toMatchObject({ type: "message", parentId: null }); + }); + + it("preserves supported current-version linear transcripts", async () => { + const root = await makeRoot("openclaw-parent-fork-linear-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + await fs.writeFile( + parentSessionFile, + [ + { + type: "session", + version: 3, + id: "parent-linear", + timestamp: "2026-06-15T00:00:00.000Z", + cwd: root, + }, + { + type: "message", + id: "linear-user", + timestamp: "2026-06-15T00:00:01.000Z", + message: { role: "user", content: "hello" }, + }, + { + type: "message", + id: "linear-assistant", + timestamp: "2026-06-15T00:00:02.000Z", + message: { role: "assistant", content: "hi" }, + }, + { + type: "metadata", + id: "linear-metadata", + parentId: "linear-assistant", + payload: { source: "plugin" }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n") + "\n", + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: "parent-linear", + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("expected forked session"); + } + const fork = forked.transcript; + const records = (await fs.readFile(fork.sessionFile, "utf-8")) + .trim() + .split(/\r?\n/) + .map((line) => JSON.parse(line) as Record); + expect(records.slice(1)).toMatchObject([ + { id: "linear-user", parentId: null }, + { id: "linear-assistant", parentId: "linear-user" }, + { id: "linear-metadata", parentId: "linear-assistant" }, + ]); + const reopened = SessionManager.open(fork.sessionFile, sessionsDir); + expect(reopened.buildSessionContext().messages).toHaveLength(2); + reopened.appendMessage({ role: "user", content: "continued", timestamp: Date.now() }); + const continuedRecords = (await fs.readFile(fork.sessionFile, "utf-8")) + .trim() + .split(/\r?\n/) + .map((line) => JSON.parse(line) as Record); + expect(continuedRecords.at(-1)).toMatchObject({ + type: "message", + parentId: "linear-metadata", + }); + }); + + it("creates a header-only child when the parent has no entries", async () => { + const root = await makeRoot("openclaw-parent-fork-empty-"); + const sessionsDir = path.join(root, "sessions"); + await fs.mkdir(sessionsDir); + const parentSessionFile = path.join(sessionsDir, "parent.jsonl"); + const parentSessionId = "parent-empty"; + await fs.writeFile( + parentSessionFile, + `${JSON.stringify({ + type: "session", + version: 3, + id: parentSessionId, + timestamp: "2026-05-01T00:00:00.000Z", + cwd: root, + })}\n`, + "utf-8", + ); + + const forked = await forkSessionFromParentTranscript({ + parentEntry: { + sessionId: parentSessionId, + sessionFile: parentSessionFile, + updatedAt: Date.now(), + }, + agentId: "main", + parentSessionKey: "agent:main:main", + sessionKey: "agent:main:child", + storePath: path.join(sessionsDir, "sessions.json"), + }); + + if (forked.status !== "created") { + throw new Error("expected forked session entry"); + } + const fork = forked.transcript; + const raw = await fs.readFile(fork.sessionFile, "utf-8"); + const lines = raw.trim().split(/\r?\n/u); + expect(lines).toHaveLength(1); + const resolvedParentSessionFile = await fs.realpath(parentSessionFile); + const header = JSON.parse(lines[0] ?? "{}") as Record; + expect(header.type).toBe("session"); + expect(header.id).toBe(fork.sessionId); + expect(header.parentSession).toBe(resolvedParentSessionFile); + }); +}); diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index 0ba4af2928cf..cec12f8f937c 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -109,7 +109,12 @@ import { resolveOwnedSessionTranscriptWriteLockRunner, withOwnedSessionTranscriptWrites, } from "./transcript-write-context.js"; -import type { GroupKeyResolution, SessionCompactionCheckpoint, SessionEntry } from "./types.js"; +import { + mergeSessionEntry, + type GroupKeyResolution, + type SessionCompactionCheckpoint, + type SessionEntry, +} from "./types.js"; /** * Session access API for callers that need entries or transcripts without @@ -457,6 +462,13 @@ const loadSessionArchiveRuntime = createLazyRuntimeModule( () => import("../../gateway/session-archive.runtime.js"), ); +// Fork-source reading parses legacy transcript versions through the agents +// session-manager; load it lazily so accessor consumers do not pull that +// runtime (and its module-init package metadata reads) at import time. +const loadSessionForkTranscriptRuntime = createLazyRuntimeModule( + () => import("./session-fork-transcript.runtime.js"), +); + export type SessionEntryPatchOptions = { /** Entry to synthesize when a patch operation is allowed to create. */ fallbackEntry?: SessionEntry; @@ -1369,6 +1381,91 @@ export type ForkedSessionTranscriptTarget = { sessionId: string; }; +/** Decision made before inheriting parent context into a child session. */ +export type SessionParentForkDecision = + | { + status: "fork"; + maxTokens: number; + parentTokens?: number; + } + | { + status: "skip"; + reason: "parent-too-large"; + maxTokens: number; + parentTokens: number; + message: string; + }; + +/** Transcript identity created for a child fork. */ +export type ParentForkedSessionTranscript = { + sessionFile: string; + sessionId: string; +}; + +export type ForkSessionFromParentTranscriptResult = + | { + status: "created"; + transcript: ParentForkedSessionTranscript; + } + | { status: "missing-parent" } + | { status: "failed" }; + +export type ForkSessionFromParentTranscriptParams = { + agentId?: string; + parentEntry: SessionEntry; + parentSessionKey: string; + sessionKey: string; + storePath: string; + /** Cross-agent forks land the child transcript beside the child's store. */ + targetStorePath?: string; +}; + +export type ForkSessionEntryFromParentTargetResult = + | { + status: "forked"; + fork: ParentForkedSessionTranscript; + parentEntry: SessionEntry; + sessionEntry: SessionEntry; + decision: Extract; + } + | { + status: "skipped"; + reason: "existing-entry" | "decision-skip"; + parentEntry?: SessionEntry; + sessionEntry: SessionEntry; + decision?: SessionParentForkDecision; + } + | { status: "missing-entry" } + | { status: "missing-parent" } + | { status: "failed" }; + +export type ForkSessionEntryFromParentTargetParams = { + agentId?: string; + decisionSkipPatch?: (params: { + decision: Extract; + entry: SessionEntry; + parentEntry: SessionEntry; + }) => Partial | null; + fallbackEntry?: SessionEntry; + parentTarget: SessionLifecycleStoreTarget; + patch?: (params: { + entry: SessionEntry; + parentEntry: SessionEntry; + fork: ParentForkedSessionTranscript; + decision: Extract; + }) => Partial; + /** + * File-era seam: token counting for the fork decision still reads transcript + * tails through gateway helpers, so the caller supplies the decision. The + * SQLite flip resolves the decision inside the storage boundary instead. + */ + resolveDecision: (parentEntry: SessionEntry) => Promise; + sessionTarget: SessionLifecycleStoreTarget; + skipForkWhen?: (entry: SessionEntry) => boolean; + skipPatch?: (entry: SessionEntry) => Partial | null; + storePath: string; +}; + export type CreateForkedSessionTranscriptParams = { /** Working directory recorded in the forked transcript header. */ cwd: string; @@ -1414,6 +1511,190 @@ export async function createForkedSessionTranscript( return { sessionFile, sessionId }; } +/** + * Forks the active branch of a parent transcript into a fresh child transcript. + * This is for guarded callers that already own the eventual entry commit. + */ +export async function forkSessionFromParentTranscript( + params: ForkSessionFromParentTranscriptParams, +): Promise { + let parentSessionFile: string; + try { + parentSessionFile = resolveSessionFilePath( + params.parentEntry.sessionId, + params.parentEntry, + resolveSessionFilePathOptions({ + ...(params.agentId ? { agentId: params.agentId } : {}), + storePath: params.storePath, + }), + ); + } catch { + return { status: "missing-parent" }; + } + if (!parentSessionFile) { + return { status: "missing-parent" }; + } + try { + const { buildForkedBranchEntries, forkSourceHasAssistantEntry, readForkSourceTranscript } = + await loadSessionForkTranscriptRuntime(); + const source = await readForkSourceTranscript(parentSessionFile); + if (!source) { + return { status: "failed" }; + } + const shouldPersistBranch = + source.preserveLeafControl || forkSourceHasAssistantEntry(source.branchEntries); + // Cross-agent forks land beside the child's store; same-store forks stay + // beside the parent transcript so sibling artifacts sort together. + const targetSessionsDir = params.targetStorePath + ? path.dirname(params.targetStorePath) + : source.sessionDir; + const transcript = shouldPersistBranch + ? await createForkedSessionTranscript({ + cwd: source.cwd, + parentSessionFile, + sessionsDir: targetSessionsDir, + buildEntries: ({ timestamp }) => buildForkedBranchEntries({ source, timestamp }), + }) + : // Header-only fork: nothing on the active branch is worth copying. + await createForkedSessionTranscript({ + cwd: source.cwd, + parentSessionFile, + sessionsDir: targetSessionsDir, + }); + return { status: "created", transcript }; + } catch { + return { status: "failed" }; + } +} + +function resolveEntryFromStoreKeys(params: { + store: Record; + keys: readonly string[]; +}): SessionEntry | undefined { + for (const key of params.keys) { + const entry = params.store[key]; + if (entry) { + return entry; + } + } + return undefined; +} + +function persistForkedSessionEntry(params: { + store: Record; + target: SessionLifecycleStoreTarget; + existing: SessionEntry; + patch: Partial; +}): SessionEntry { + const next = mergeSessionEntry(params.existing, params.patch); + params.store[params.target.canonicalKey] = next; + // Alias rows collapse onto the canonical key so the forked identity has one owner row. + for (const key of params.target.storeKeys) { + if (key !== params.target.canonicalKey) { + delete params.store[key]; + } + } + return next; +} + +/** + * Forks parent transcript content and persists the child entry/alias cleanup in + * one storage-owned operation. + */ +export async function forkSessionEntryFromParentTarget( + params: ForkSessionEntryFromParentTargetParams, +): Promise { + return await updateSessionStore( + params.storePath, + async (store) => { + const parentEntry = resolveEntryFromStoreKeys({ + store, + keys: params.parentTarget.storeKeys, + }); + if (!parentEntry?.sessionId) { + return { status: "missing-parent" }; + } + + const entry = + resolveEntryFromStoreKeys({ store, keys: params.sessionTarget.storeKeys }) ?? + params.fallbackEntry; + if (!entry) { + return { status: "missing-entry" }; + } + + if (params.skipForkWhen?.(entry)) { + const patch = params.skipPatch?.(entry); + const sessionEntry = patch + ? persistForkedSessionEntry({ + store, + target: params.sessionTarget, + existing: entry, + patch, + }) + : entry; + return { status: "skipped", reason: "existing-entry", parentEntry, sessionEntry }; + } + + const decision = await params.resolveDecision(parentEntry); + if (decision.status === "skip") { + const patch = params.decisionSkipPatch?.({ decision, entry, parentEntry }); + const sessionEntry = patch + ? persistForkedSessionEntry({ + store, + target: params.sessionTarget, + existing: entry, + patch, + }) + : entry; + return { + status: "skipped", + reason: "decision-skip", + parentEntry, + sessionEntry, + decision, + }; + } + + const forked = await forkSessionFromParentTranscript({ + ...(params.agentId ? { agentId: params.agentId } : {}), + parentEntry, + parentSessionKey: params.parentTarget.canonicalKey, + sessionKey: params.sessionTarget.canonicalKey, + storePath: params.storePath, + }); + if (forked.status !== "created") { + return { status: "failed" }; + } + const fork = forked.transcript; + const sessionEntry = persistForkedSessionEntry({ + store, + target: params.sessionTarget, + existing: entry, + patch: { + ...params.patch?.({ entry, parentEntry, fork, decision }), + sessionId: fork.sessionId, + sessionFile: fork.sessionFile, + forkedFromParent: true, + }, + }); + return { + status: "forked", + fork, + parentEntry, + sessionEntry, + decision, + }; + }, + { + skipSaveWhenResult: (result) => + result.status === "missing-entry" || + result.status === "missing-parent" || + result.status === "failed" || + (result.status === "skipped" && result.sessionEntry === params.fallbackEntry), + }, + ); +} + /** Updates an existing entry only; returns null when the session is absent. */ export async function updateSessionEntry( scope: SessionAccessScope, diff --git a/src/config/sessions/session-fork-transcript.runtime.ts b/src/config/sessions/session-fork-transcript.runtime.ts new file mode 100644 index 000000000000..7408400f7643 --- /dev/null +++ b/src/config/sessions/session-fork-transcript.runtime.ts @@ -0,0 +1,186 @@ +// Parent-fork transcript source reading and branch-entry construction. +// The session accessor composes these helpers with createForkedSessionTranscript +// so parent forks stay one storage-owned operation (#88838). +import crypto from "node:crypto"; +import path from "node:path"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { + migrateSessionEntries, + parseSessionEntries, + type SessionEntry as AgentSessionEntry, + type SessionHeader, +} from "../../agents/sessions/session-manager.js"; +import { readRegularFile } from "../../infra/fs-safe.js"; +import { + isSessionTranscriptLeafControl, + mergeSessionTranscriptVisiblePathWithOpaqueAppendPath, + scanSessionTranscriptTree, + selectSessionTranscriptTreePathNodes, +} from "./transcript-tree.js"; + +/** Active-branch snapshot of a parent transcript selected for forking. */ +export type ForkSourceTranscript = { + cwd: string; + sessionDir: string; + leafId: string | null; + appendParentId: string | null; + appendMode?: "side"; + preserveLeafControl: boolean; + branchEntries: unknown[]; + labelsToWrite: Array<{ targetId: string; label: string; timestamp: string }>; +}; + +function generateEntryId(existingIds: Set): string { + for (let attempt = 0; attempt < 100; attempt += 1) { + const id = crypto.randomUUID().slice(0, 8); + if (!existingIds.has(id)) { + existingIds.add(id); + return id; + } + } + const id = crypto.randomUUID(); + existingIds.add(id); + return id; +} + +/** True when the selected branch carries at least one assistant message. */ +export function forkSourceHasAssistantEntry(entries: unknown[]): boolean { + return entries.some( + (entry) => + isRecord(entry) && + entry.type === "message" && + isRecord(entry.message) && + entry.message.role === "assistant", + ); +} + +function collectBranchLabels(params: { + allEntries: unknown[]; + pathEntryIds: Set; +}): Array<{ targetId: string; label: string; timestamp: string }> { + const labelsToWrite: Array<{ targetId: string; label: string; timestamp: string }> = []; + for (const entry of params.allEntries) { + if (!isRecord(entry)) { + continue; + } + if ( + entry.type === "label" && + typeof entry.label === "string" && + typeof entry.targetId === "string" && + typeof entry.id === "string" && + !params.pathEntryIds.has(entry.id) && + params.pathEntryIds.has(entry.targetId) && + typeof entry.timestamp === "string" + ) { + labelsToWrite.push({ + targetId: entry.targetId, + label: entry.label, + timestamp: entry.timestamp, + }); + } + } + return labelsToWrite; +} + +/** Reads the parent transcript and selects the active branch to copy. */ +export async function readForkSourceTranscript( + parentSessionFile: string, +): Promise { + const raw = (await readRegularFile({ filePath: parentSessionFile })).buffer.toString("utf-8"); + const fileEntries = parseSessionEntries(raw); + migrateSessionEntries(fileEntries); + const header = + fileEntries.find((entry): entry is SessionHeader => entry.type === "session") ?? null; + const entries = fileEntries.filter((entry) => entry.type !== "session"); + const tree = scanSessionTranscriptTree(entries); + const leafId = tree.leafId; + const appendParentId = tree.appendParentId; + const visiblePath = selectSessionTranscriptTreePathNodes(tree, leafId); + const appendPath = selectSessionTranscriptTreePathNodes(tree, appendParentId); + const mergedPath = mergeSessionTranscriptVisiblePathWithOpaqueAppendPath({ + visiblePath, + appendPath, + appendParentId, + }); + const branchEntries = mergedPath.nodes.flatMap((node) => { + if (!isRecord(node.entry)) { + return []; + } + const parentId = node.selectedParentId; + return [node.entry.parentId === parentId ? node.entry : { ...node.entry, parentId }]; + }); + const pathEntryIds = new Set( + branchEntries.flatMap((entry) => + isRecord(entry) && typeof entry.id === "string" ? [entry.id] : [], + ), + ); + const lastLeafUpdateNode = tree.nodes.findLast((node) => node.leafId !== undefined); + const lastLeafUpdateEntry = lastLeafUpdateNode?.entry; + return { + cwd: header?.cwd ?? process.cwd(), + sessionDir: path.dirname(parentSessionFile), + leafId, + appendParentId: mergedPath.appendParentId, + ...(lastLeafUpdateNode?.appendMode ? { appendMode: lastLeafUpdateNode.appendMode } : {}), + preserveLeafControl: isSessionTranscriptLeafControl(lastLeafUpdateEntry), + branchEntries, + labelsToWrite: collectBranchLabels({ allEntries: entries, pathEntryIds }), + }; +} + +function buildBranchLabelEntries(params: { + labelsToWrite: Array<{ targetId: string; label: string; timestamp: string }>; + pathEntryIds: Set; + lastEntryId: string | null; +}): AgentSessionEntry[] { + let parentId = params.lastEntryId; + const labelEntries: AgentSessionEntry[] = []; + for (const { targetId, label, timestamp } of params.labelsToWrite) { + const labelEntry = { + type: "label", + id: generateEntryId(params.pathEntryIds), + parentId, + timestamp, + targetId, + label, + } satisfies AgentSessionEntry; + params.pathEntryIds.add(labelEntry.id); + labelEntries.push(labelEntry); + parentId = labelEntry.id; + } + return labelEntries; +} + +/** Builds the copied branch, re-targeted labels, and optional leaf control for a fork. */ +export function buildForkedBranchEntries(params: { + source: ForkSourceTranscript; + /** Fork header timestamp; the leaf control shares it so the copied branch reopens on the parent's active leaf. */ + timestamp: string; +}): unknown[] { + const pathEntries = params.source.branchEntries; + const pathEntryIds = new Set( + pathEntries.flatMap((entry) => + isRecord(entry) && typeof entry.id === "string" ? [entry.id] : [], + ), + ); + const lastPathEntry = pathEntries.at(-1); + const lastPathEntryId = + isRecord(lastPathEntry) && typeof lastPathEntry.id === "string" ? lastPathEntry.id : null; + const labelEntries = buildBranchLabelEntries({ + labelsToWrite: params.source.labelsToWrite, + pathEntryIds, + lastEntryId: lastPathEntryId, + }); + const leafEntry = params.source.preserveLeafControl + ? { + type: "leaf", + id: generateEntryId(pathEntryIds), + parentId: labelEntries.at(-1)?.id ?? lastPathEntryId, + timestamp: params.timestamp, + targetId: params.source.leafId, + appendParentId: params.source.appendParentId, + ...(params.source.appendMode ? { appendMode: params.source.appendMode } : {}), + } + : null; + return [...pathEntries, ...labelEntries, ...(leafEntry ? [leafEntry] : [])]; +} diff --git a/src/gateway/session-create-service.ts b/src/gateway/session-create-service.ts index 0667df0bd53a..fdc1b107eb20 100644 --- a/src/gateway/session-create-service.ts +++ b/src/gateway/session-create-service.ts @@ -1,5 +1,4 @@ import { randomUUID } from "node:crypto"; -import path from "node:path"; import { normalizeOptionalLowercaseString, normalizeOptionalString, @@ -449,9 +448,11 @@ export async function createGatewaySession(params: { const fork = await forkSessionFromParent({ parentEntry: currentParentSessionEntry, agentId: parentSessionTarget.agentId, - sessionsDir: path.dirname(parentSessionTarget.storePath), + parentSessionKey: canonicalParentSessionKey, + sessionKey: target.canonicalKey, + storePath: parentSessionTarget.storePath, // Keep the fork transcript owned by the child store across agent boundaries. - targetSessionsDir: path.dirname(target.storePath), + targetStorePath: target.storePath, }); if (!fork) { return {