mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-09 15:59:30 +00:00
refactor(sessions): route new session store bypasses through the accessor (#101179)
This commit is contained in:
parent
97777a7026
commit
4e2a80cac3
11 changed files with 296 additions and 109 deletions
|
|
@ -186,6 +186,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
|
|||
"src/commands/tasks.ts",
|
||||
"src/config/sessions/cleanup-service.ts",
|
||||
"src/gateway/boot.ts",
|
||||
"src/gateway/server-methods/sessions.ts",
|
||||
"src/gateway/server-node-events.ts",
|
||||
"src/gateway/session-compaction-checkpoints.ts",
|
||||
"src/plugins/host-hook-cleanup.ts",
|
||||
|
|
@ -196,6 +197,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
|
|||
export const migratedTranscriptWriterFiles = new Set([
|
||||
"src/agents/command/attempt-execution.ts",
|
||||
"src/agents/embedded-agent-runner/context-engine-maintenance.ts",
|
||||
"src/auto-reply/reply/session-fork.runtime.ts",
|
||||
"src/config/sessions/transcript.ts",
|
||||
"src/gateway/server-methods/chat.ts",
|
||||
"src/gateway/server-methods/chat-transcript-inject.ts",
|
||||
|
|
@ -569,6 +571,7 @@ const writeSourceRootPaths = [
|
|||
const transcriptWriterSourceRootPaths = [
|
||||
"src/agents/command",
|
||||
"src/agents/embedded-agent-runner",
|
||||
"src/auto-reply/reply",
|
||||
"src/config/sessions",
|
||||
"src/gateway/server-methods",
|
||||
"src/sessions",
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@
|
|||
"src/config/sessions/store.ts": 15,
|
||||
"src/config/sessions/test-helpers.ts": 2,
|
||||
"src/config/sessions/transcript.ts": 6,
|
||||
"src/cron/isolated-agent/session.ts": 3,
|
||||
"src/cron/isolated-agent/session.ts": 2,
|
||||
"src/infra/approval-request-account-binding.ts": 2,
|
||||
"src/infra/heartbeat-runner.ts": 4,
|
||||
"src/plugins/runtime/runtime-agent.ts": 1
|
||||
|
|
@ -60,7 +60,6 @@
|
|||
"src/config/sessions/session-registry-maintenance.ts": 2,
|
||||
"src/gateway/server-methods/agent.ts": 2,
|
||||
"src/gateway/server-methods/chat.ts": 2,
|
||||
"src/gateway/server-methods/sessions.ts": 3,
|
||||
"src/gateway/session-lifecycle-state.ts": 2,
|
||||
"src/gateway/test-helpers.mocks.ts": 1,
|
||||
"src/plugins/runtime/runtime-agent.ts": 2
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import {
|
|||
resolveSessionFilePath,
|
||||
resolveSessionFilePathOptions,
|
||||
} from "../../config/sessions/paths.js";
|
||||
import { createForkedSessionTranscript } from "../../config/sessions/session-accessor.js";
|
||||
import {
|
||||
isSessionTranscriptLeafControl,
|
||||
mergeSessionTranscriptVisiblePathWithOpaqueAppendPath,
|
||||
|
|
@ -24,7 +25,6 @@ import {
|
|||
resolveFreshSessionTotalTokens,
|
||||
type SessionEntry as StoreSessionEntry,
|
||||
} from "../../config/sessions/types.js";
|
||||
import { CURRENT_SESSION_VERSION } from "../../config/sessions/version.js";
|
||||
import { readLatestRecentSessionUsageFromTranscriptAsync } from "../../gateway/session-utils.fs.js";
|
||||
import { readRegularFile } from "../../infra/fs-safe.js";
|
||||
|
||||
|
|
@ -246,41 +246,11 @@ function buildBranchLabelEntries(params: {
|
|||
return labelEntries;
|
||||
}
|
||||
|
||||
async function writeForkHeaderOnly(params: {
|
||||
parentSessionFile: string;
|
||||
sessionDir: string;
|
||||
cwd: string;
|
||||
}): Promise<{ sessionId: string; sessionFile: string }> {
|
||||
const sessionId = crypto.randomUUID();
|
||||
const timestamp = new Date().toISOString();
|
||||
const fileTimestamp = timestamp.replace(/[:.]/g, "-");
|
||||
const sessionFile = path.join(params.sessionDir, `${fileTimestamp}_${sessionId}.jsonl`);
|
||||
const header = {
|
||||
type: "session",
|
||||
version: CURRENT_SESSION_VERSION,
|
||||
id: sessionId,
|
||||
timestamp,
|
||||
cwd: params.cwd,
|
||||
parentSession: params.parentSessionFile,
|
||||
} satisfies SessionHeader;
|
||||
await fs.mkdir(path.dirname(sessionFile), { recursive: true });
|
||||
await fs.writeFile(sessionFile, `${JSON.stringify(header)}\n`, {
|
||||
encoding: "utf-8",
|
||||
mode: 0o600,
|
||||
flag: "wx",
|
||||
});
|
||||
return { sessionId, sessionFile };
|
||||
}
|
||||
|
||||
async function writeBranchedSession(params: {
|
||||
parentSessionFile: string;
|
||||
source: ForkSourceTranscript;
|
||||
sessionDir: string;
|
||||
}): Promise<{ sessionId: string; sessionFile: string }> {
|
||||
const sessionId = crypto.randomUUID();
|
||||
const timestamp = new Date().toISOString();
|
||||
const fileTimestamp = timestamp.replace(/[:.]/g, "-");
|
||||
const sessionFile = path.join(params.sessionDir, `${fileTimestamp}_${sessionId}.jsonl`);
|
||||
const pathEntries = params.source.branchEntries;
|
||||
const pathEntryIds = new Set(
|
||||
pathEntries.flatMap((entry) =>
|
||||
|
|
@ -295,33 +265,27 @@ async function writeBranchedSession(params: {
|
|||
pathEntryIds,
|
||||
lastEntryId: lastPathEntryId,
|
||||
});
|
||||
const header = {
|
||||
type: "session",
|
||||
version: CURRENT_SESSION_VERSION,
|
||||
id: sessionId,
|
||||
timestamp,
|
||||
return await createForkedSessionTranscript({
|
||||
cwd: params.source.cwd,
|
||||
parentSession: params.parentSessionFile,
|
||||
} satisfies SessionHeader;
|
||||
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;
|
||||
const entries = [header, ...pathEntries, ...labelEntries, ...(leafEntry ? [leafEntry] : [])];
|
||||
await fs.mkdir(path.dirname(sessionFile), { recursive: true });
|
||||
await fs.writeFile(sessionFile, `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`, {
|
||||
encoding: "utf-8",
|
||||
mode: 0o600,
|
||||
flag: "wx",
|
||||
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] : [])];
|
||||
},
|
||||
});
|
||||
return { sessionId, sessionFile };
|
||||
}
|
||||
|
||||
/** Creates a child session transcript from a parent session branch. */
|
||||
|
|
@ -349,10 +313,11 @@ export async function forkSessionFromParentRuntime(params: {
|
|||
const targetSessionsDir = params.targetSessionsDir ?? source.sessionDir;
|
||||
return shouldPersistBranch
|
||||
? await writeBranchedSession({ parentSessionFile, source, sessionDir: targetSessionsDir })
|
||||
: await writeForkHeaderOnly({
|
||||
parentSessionFile,
|
||||
sessionDir: targetSessionsDir,
|
||||
: // Header-only fork: nothing on the active branch is worth copying.
|
||||
await createForkedSessionTranscript({
|
||||
cwd: source.cwd,
|
||||
parentSessionFile,
|
||||
sessionsDir: targetSessionsDir,
|
||||
});
|
||||
} catch {
|
||||
return null;
|
||||
|
|
|
|||
68
src/config/sessions/session-accessor.fork-transcript.test.ts
Normal file
68
src/config/sessions/session-accessor.fork-transcript.test.ts
Normal file
|
|
@ -0,0 +1,68 @@
|
|||
// Behavior tests for the accessor fork-transcript creation boundary.
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, expect, it } from "vitest";
|
||||
import { createForkedSessionTranscript } from "./session-accessor.js";
|
||||
import { CURRENT_SESSION_VERSION } from "./version.js";
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
function makeSessionsDir(): string {
|
||||
const dir = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-fork-transcript-")));
|
||||
tempDirs.push(dir);
|
||||
return dir;
|
||||
}
|
||||
|
||||
function readJsonlRecords(filePath: string): Record<string, unknown>[] {
|
||||
return fs
|
||||
.readFileSync(filePath, "utf-8")
|
||||
.trim()
|
||||
.split(/\r?\n/u)
|
||||
.map((line) => JSON.parse(line) as Record<string, unknown>);
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
for (const dir of tempDirs.splice(0)) {
|
||||
fs.rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("writes a header-only fork artifact with parent lineage", async () => {
|
||||
const sessionsDir = makeSessionsDir();
|
||||
const fork = await createForkedSessionTranscript({
|
||||
cwd: "/workspace/project",
|
||||
parentSessionFile: "/sessions/parent.jsonl",
|
||||
sessionsDir,
|
||||
});
|
||||
|
||||
expect(path.dirname(fork.sessionFile)).toBe(sessionsDir);
|
||||
expect(fork.sessionFile.endsWith(`_${fork.sessionId}.jsonl`)).toBe(true);
|
||||
const records = readJsonlRecords(fork.sessionFile);
|
||||
expect(records).toHaveLength(1);
|
||||
expect(records[0]).toMatchObject({
|
||||
type: "session",
|
||||
version: CURRENT_SESSION_VERSION,
|
||||
id: fork.sessionId,
|
||||
cwd: "/workspace/project",
|
||||
parentSession: "/sessions/parent.jsonl",
|
||||
});
|
||||
});
|
||||
|
||||
it("copies built records in order sharing the fork header timestamp", async () => {
|
||||
const sessionsDir = makeSessionsDir();
|
||||
const fork = await createForkedSessionTranscript({
|
||||
cwd: "/workspace/project",
|
||||
parentSessionFile: "/sessions/parent.jsonl",
|
||||
sessionsDir,
|
||||
buildEntries: ({ sessionId, timestamp }) => [
|
||||
{ type: "message", id: "m-1", parentId: null },
|
||||
{ type: "leaf", id: "leaf-1", parentId: "m-1", timestamp, forkSessionId: sessionId },
|
||||
],
|
||||
});
|
||||
|
||||
const records = readJsonlRecords(fork.sessionFile);
|
||||
expect(records.map((record) => record.type)).toEqual(["session", "message", "leaf"]);
|
||||
expect(records[2]?.timestamp).toBe(records[0]?.timestamp);
|
||||
expect(records[2]?.forkSessionId).toBe(fork.sessionId);
|
||||
});
|
||||
|
|
@ -90,7 +90,7 @@ import {
|
|||
} from "./transcript-append.js";
|
||||
import { resolveSessionTranscriptFile } from "./transcript-file-resolve.js";
|
||||
import { createSessionTranscriptHeader } from "./transcript-header.js";
|
||||
import { writeJsonlLines } from "./transcript-jsonl.js";
|
||||
import { serializeJsonlLine, writeJsonlLines } from "./transcript-jsonl.js";
|
||||
import { replayRecentUserAssistantMessages } from "./transcript-replay.js";
|
||||
import { streamSessionTranscriptLines } from "./transcript-stream.js";
|
||||
import {
|
||||
|
|
@ -1334,6 +1334,57 @@ function ensureCreatedSessionTranscript(params: {
|
|||
}
|
||||
}
|
||||
|
||||
/** Fork target identity generated by createForkedSessionTranscript. */
|
||||
export type ForkedSessionTranscriptTarget = {
|
||||
sessionFile: string;
|
||||
sessionId: string;
|
||||
};
|
||||
|
||||
export type CreateForkedSessionTranscriptParams = {
|
||||
/** Working directory recorded in the forked transcript header. */
|
||||
cwd: string;
|
||||
/** Source transcript recorded as the fork's parentSession lineage. */
|
||||
parentSessionFile: string;
|
||||
/** Directory that owns the forked transcript artifact. */
|
||||
sessionsDir: string;
|
||||
/**
|
||||
* Builds the non-header records copied into the fork, in transcript order.
|
||||
* Receives the generated fork identity so records can share its timestamp.
|
||||
*/
|
||||
buildEntries?: (fork: { sessionId: string; timestamp: string }) => readonly unknown[];
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates a forked transcript artifact under a fresh session id as one storage
|
||||
* operation. File-backed storage writes header plus copied records as a new
|
||||
* JSONL artifact; SQLite implements the same fork as transcript row copies
|
||||
* inside one write transaction (#88838).
|
||||
*/
|
||||
export async function createForkedSessionTranscript(
|
||||
params: CreateForkedSessionTranscriptParams,
|
||||
): Promise<ForkedSessionTranscriptTarget> {
|
||||
const sessionId = randomUUID();
|
||||
const timestamp = new Date().toISOString();
|
||||
// Fork artifacts keep the timestamp-prefixed file name so sibling transcripts
|
||||
// in one sessions dir sort by creation time.
|
||||
const fileTimestamp = timestamp.replace(/[:.]/g, "-");
|
||||
const sessionFile = path.join(params.sessionsDir, `${fileTimestamp}_${sessionId}.jsonl`);
|
||||
const header = createSessionTranscriptHeader({
|
||||
sessionId,
|
||||
cwd: params.cwd,
|
||||
parentSession: params.parentSessionFile,
|
||||
timestamp,
|
||||
});
|
||||
const entries = params.buildEntries?.({ sessionId, timestamp }) ?? [];
|
||||
fs.mkdirSync(path.dirname(sessionFile), { recursive: true });
|
||||
// "wx" keeps create-only semantics: a fork must never clobber an existing transcript.
|
||||
await writeJsonlLines(sessionFile, [header, ...entries].map(serializeJsonlLine), {
|
||||
flag: "wx",
|
||||
mode: 0o600,
|
||||
});
|
||||
return { sessionFile, sessionId };
|
||||
}
|
||||
|
||||
/** Updates an existing entry only; returns null when the session is absent. */
|
||||
export async function updateSessionEntry(
|
||||
scope: SessionAccessScope,
|
||||
|
|
|
|||
|
|
@ -6,6 +6,10 @@ import { CURRENT_SESSION_VERSION } from "./version.js";
|
|||
type SessionTranscriptHeaderParams = {
|
||||
sessionId?: string;
|
||||
cwd?: string;
|
||||
/** Source transcript lineage recorded on forked transcript headers. */
|
||||
parentSession?: string;
|
||||
/** Stable timestamp shared with sibling records written in the same operation. */
|
||||
timestamp?: string;
|
||||
};
|
||||
|
||||
/** Creates a session transcript header entry with current version metadata. */
|
||||
|
|
@ -14,7 +18,8 @@ export function createSessionTranscriptHeader(params: SessionTranscriptHeaderPar
|
|||
type: "session",
|
||||
version: CURRENT_SESSION_VERSION,
|
||||
id: params.sessionId ?? randomUUID(),
|
||||
timestamp: new Date().toISOString(),
|
||||
timestamp: params.timestamp ?? new Date().toISOString(),
|
||||
cwd: params.cwd ?? process.cwd(),
|
||||
...(params.parentSession ? { parentSession: params.parentSession } : {}),
|
||||
};
|
||||
}
|
||||
|
|
|
|||
59
src/cron/isolated-agent/session.latest-read.test.ts
Normal file
59
src/cron/isolated-agent/session.latest-read.test.ts
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
// Behavior tests for the cron lifecycle-guard session read.
|
||||
// loadCronSessionEntryLatest must observe the latest persisted row even when a
|
||||
// cached in-process store snapshot is still considered current, because cron
|
||||
// admission guards fence on it (see run.ts assertAllowed).
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { clearSessionStoreCacheForTest, loadSessionStore } from "../../config/sessions/store.js";
|
||||
import { loadCronSessionEntryLatest } from "./session.js";
|
||||
|
||||
const SESSION_KEY = "agent:main:cron:job-1";
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
function createStoreFile(sessionId: string): string {
|
||||
const dir = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cron-latest-")));
|
||||
tempDirs.push(dir);
|
||||
const storePath = path.join(dir, "sessions.json");
|
||||
fs.writeFileSync(storePath, serializeStore(sessionId), "utf-8");
|
||||
return storePath;
|
||||
}
|
||||
|
||||
// Both serializations must have identical byte length so an out-of-band rewrite
|
||||
// with a restored mtime looks unchanged to the mtime/size-validated cache.
|
||||
function serializeStore(sessionId: string): string {
|
||||
return JSON.stringify({ [SESSION_KEY]: { sessionId, updatedAt: 1000 } });
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllEnvs();
|
||||
clearSessionStoreCacheForTest();
|
||||
for (const dir of tempDirs.splice(0)) {
|
||||
fs.rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("reads the latest persisted row past a still-current cached store snapshot", () => {
|
||||
vi.stubEnv("OPENCLAW_SESSION_CACHE_TTL_MS", "45000");
|
||||
clearSessionStoreCacheForTest();
|
||||
const storePath = createStoreFile("sess-one");
|
||||
// A whole-second mtime round-trips exactly through utimes/stat, so the
|
||||
// rewritten file below validates as unchanged against the cached snapshot.
|
||||
const pinnedTime = new Date(1_700_000_000_000);
|
||||
fs.utimesSync(storePath, pinnedTime, pinnedTime);
|
||||
|
||||
// Warm the in-process cache, then rewrite the row out-of-band with the same
|
||||
// byte length and mtime so the cached snapshot still validates as current.
|
||||
expect(loadSessionStore(storePath)[SESSION_KEY]?.sessionId).toBe("sess-one");
|
||||
fs.writeFileSync(storePath, serializeStore("sess-two"), "utf-8");
|
||||
fs.utimesSync(storePath, pinnedTime, pinnedTime);
|
||||
|
||||
expect(loadSessionStore(storePath)[SESSION_KEY]?.sessionId).toBe("sess-one");
|
||||
expect(loadCronSessionEntryLatest(storePath, SESSION_KEY)?.sessionId).toBe("sess-two");
|
||||
});
|
||||
|
||||
it("returns undefined for a session key without a persisted row", () => {
|
||||
const storePath = createStoreFile("sess-one");
|
||||
expect(loadCronSessionEntryLatest(storePath, "agent:main:cron:missing")).toBeUndefined();
|
||||
});
|
||||
|
|
@ -13,6 +13,7 @@ import {
|
|||
resolveSessionResetPolicy,
|
||||
type SessionFreshness,
|
||||
} from "../../config/sessions/reset-policy.js";
|
||||
import { loadSessionEntry } from "../../config/sessions/session-accessor.js";
|
||||
import { loadSessionStore } from "../../config/sessions/store-load.js";
|
||||
import type { SessionEntry } from "../../config/sessions/types.js";
|
||||
import type { OpenClawConfig } from "../../config/types.openclaw.js";
|
||||
|
|
@ -110,12 +111,18 @@ function sanitizeFreshCronSessionEntry(
|
|||
return next;
|
||||
}
|
||||
|
||||
/** Reads the exact cron session row without an in-process cache snapshot. */
|
||||
/**
|
||||
* Reads the current cron session row without an in-process cache snapshot.
|
||||
* Lifecycle admission guards compare this against the run's initial entry, so
|
||||
* the read must bypass cached store snapshots (accessor readConsistency
|
||||
* "latest"). Cron keys are canonicalized before use, so accessor key
|
||||
* resolution selects the same row the cron persist path writes.
|
||||
*/
|
||||
export function loadCronSessionEntryLatest(
|
||||
storePath: string,
|
||||
sessionKey: string,
|
||||
): SessionEntry | undefined {
|
||||
return loadSessionStore(storePath, { skipCache: true })[sessionKey];
|
||||
return loadSessionEntry({ sessionKey, storePath, readConsistency: "latest" });
|
||||
}
|
||||
|
||||
/** Resolves or rolls over the cron session entry for one isolated-agent run. */
|
||||
|
|
|
|||
|
|
@ -55,7 +55,6 @@ import {
|
|||
listConfiguredSessionStoreAgentIds,
|
||||
deleteSessionEntryLifecycle,
|
||||
type SessionEntry,
|
||||
updateSessionStore,
|
||||
} from "../../config/sessions.js";
|
||||
import { resolveAgentMainSessionKey } from "../../config/sessions/main-session.js";
|
||||
import {
|
||||
|
|
@ -2552,15 +2551,33 @@ export const sessionsHandlers: GatewayRequestHandlers = {
|
|||
agentId: requestedAgentId,
|
||||
});
|
||||
// Lock + read in a short critical section; transcript work happens outside.
|
||||
const compactTarget = await updateSessionStore(storePath, (store) => {
|
||||
const { entry, primaryKey } = migrateAndPruneGatewaySessionStoreKey({
|
||||
cfg,
|
||||
key,
|
||||
store,
|
||||
agentId: requestedAgentId,
|
||||
});
|
||||
return { entry, primaryKey };
|
||||
// The projection resolver re-runs gateway key migration on the writer
|
||||
// snapshot so alias promotion/pruning persists through the accessor.
|
||||
let compactPrimaryKey = target.canonicalKey;
|
||||
const compactRead = await applySessionPatchProjection({
|
||||
storePath,
|
||||
resolveTarget: ({ entries }) => {
|
||||
const snapshot = Object.fromEntries(
|
||||
entries.map(({ sessionKey, entry }) => [sessionKey, entry]),
|
||||
);
|
||||
const { target: migratedTarget, primaryKey } = migrateAndPruneGatewaySessionStoreKey({
|
||||
cfg,
|
||||
key,
|
||||
store: snapshot,
|
||||
agentId: requestedAgentId,
|
||||
});
|
||||
compactPrimaryKey = primaryKey;
|
||||
return { primaryKey, candidateKeys: migratedTarget.storeKeys };
|
||||
},
|
||||
// Read-only projection: persist the resolved row unchanged so the alias
|
||||
// migration above is saved even when compaction bails out below.
|
||||
project: ({ existingEntry }) =>
|
||||
existingEntry ? { ok: true, entry: existingEntry } : { ok: false },
|
||||
});
|
||||
const compactTarget = {
|
||||
entry: compactRead.ok ? compactRead.entry : undefined,
|
||||
primaryKey: compactPrimaryKey,
|
||||
};
|
||||
const entry = compactTarget.entry;
|
||||
const sessionId = entry?.sessionId;
|
||||
if (!sessionId) {
|
||||
|
|
@ -2829,42 +2846,50 @@ export const sessionsHandlers: GatewayRequestHandlers = {
|
|||
if (result.ok && result.compacted) {
|
||||
let persisted: boolean;
|
||||
try {
|
||||
persisted = await updateSessionStore(storePath, (store) => {
|
||||
const entryToUpdate = store[compactTarget.primaryKey];
|
||||
if (
|
||||
!entryToUpdate ||
|
||||
entryToUpdate.sessionId !== sessionId ||
|
||||
entryToUpdate.lifecycleRevision !== lifecycleRevision ||
|
||||
resolveSessionWorkStartError(target.canonicalKey, entryToUpdate)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
entryToUpdate.updatedAt = Date.now();
|
||||
entryToUpdate.compactionCount = Math.max(0, entryToUpdate.compactionCount ?? 0) + 1;
|
||||
if (
|
||||
result.result?.sessionId &&
|
||||
result.result.sessionId !== entryToUpdate.sessionId
|
||||
) {
|
||||
entryToUpdate.sessionId = result.result.sessionId;
|
||||
}
|
||||
if (result.result?.sessionFile) {
|
||||
entryToUpdate.sessionFile = result.result.sessionFile;
|
||||
}
|
||||
delete entryToUpdate.inputTokens;
|
||||
delete entryToUpdate.outputTokens;
|
||||
delete entryToUpdate.contextBudgetStatus;
|
||||
if (
|
||||
typeof result.result?.tokensAfter === "number" &&
|
||||
Number.isFinite(result.result.tokensAfter)
|
||||
) {
|
||||
entryToUpdate.totalTokens = result.result.tokensAfter;
|
||||
entryToUpdate.totalTokensFresh = true;
|
||||
} else {
|
||||
delete entryToUpdate.totalTokens;
|
||||
delete entryToUpdate.totalTokensFresh;
|
||||
}
|
||||
return true;
|
||||
// Guarded terminal persist: skip when session ownership rotated
|
||||
// while compaction ran (sessionId/lifecycleRevision/work-start).
|
||||
const persistProjection = await applySessionPatchProjection({
|
||||
storePath,
|
||||
resolveTarget: () => ({ primaryKey: compactTarget.primaryKey }),
|
||||
project: ({ existingEntry }) => {
|
||||
if (
|
||||
!existingEntry ||
|
||||
existingEntry.sessionId !== sessionId ||
|
||||
existingEntry.lifecycleRevision !== lifecycleRevision ||
|
||||
resolveSessionWorkStartError(target.canonicalKey, existingEntry)
|
||||
) {
|
||||
return { ok: false };
|
||||
}
|
||||
const entryToUpdate = existingEntry;
|
||||
entryToUpdate.updatedAt = Date.now();
|
||||
entryToUpdate.compactionCount =
|
||||
Math.max(0, entryToUpdate.compactionCount ?? 0) + 1;
|
||||
if (
|
||||
result.result?.sessionId &&
|
||||
result.result.sessionId !== entryToUpdate.sessionId
|
||||
) {
|
||||
entryToUpdate.sessionId = result.result.sessionId;
|
||||
}
|
||||
if (result.result?.sessionFile) {
|
||||
entryToUpdate.sessionFile = result.result.sessionFile;
|
||||
}
|
||||
delete entryToUpdate.inputTokens;
|
||||
delete entryToUpdate.outputTokens;
|
||||
delete entryToUpdate.contextBudgetStatus;
|
||||
if (
|
||||
typeof result.result?.tokensAfter === "number" &&
|
||||
Number.isFinite(result.result.tokensAfter)
|
||||
) {
|
||||
entryToUpdate.totalTokens = result.result.tokensAfter;
|
||||
entryToUpdate.totalTokensFresh = true;
|
||||
} else {
|
||||
delete entryToUpdate.totalTokens;
|
||||
delete entryToUpdate.totalTokensFresh;
|
||||
}
|
||||
return { ok: true, entry: entryToUpdate };
|
||||
},
|
||||
});
|
||||
persisted = persistProjection.ok;
|
||||
} catch (err) {
|
||||
emitCompactionEnd(false, formatErrorMessage(err));
|
||||
throw err;
|
||||
|
|
|
|||
|
|
@ -1706,6 +1706,9 @@ function hasInvalidLeafControl(lines: Iterable<string>): boolean {
|
|||
return tree.hasInvalidLeafControl;
|
||||
}
|
||||
|
||||
// File-tier only (#88838): this module is the file backend behind the
|
||||
// session-transcript-readers seam, and the index read operates on an already
|
||||
// resolved transcript artifact path, never on live session identity.
|
||||
async function extractLatestUsageFromTranscriptIndex(
|
||||
filePath: string,
|
||||
): Promise<SessionTranscriptUsageSnapshot | null> {
|
||||
|
|
|
|||
|
|
@ -149,6 +149,7 @@ describe("session accessor boundary guard", () => {
|
|||
"src/commands/tasks.ts",
|
||||
"src/config/sessions/cleanup-service.ts",
|
||||
"src/gateway/boot.ts",
|
||||
"src/gateway/server-methods/sessions.ts",
|
||||
"src/gateway/server-node-events.ts",
|
||||
"src/gateway/session-compaction-checkpoints.ts",
|
||||
"src/plugins/host-hook-cleanup.ts",
|
||||
|
|
@ -163,6 +164,7 @@ describe("session accessor boundary guard", () => {
|
|||
new Set([
|
||||
"src/agents/command/attempt-execution.ts",
|
||||
"src/agents/embedded-agent-runner/context-engine-maintenance.ts",
|
||||
"src/auto-reply/reply/session-fork.runtime.ts",
|
||||
"src/config/sessions/transcript.ts",
|
||||
"src/gateway/server-methods/chat.ts",
|
||||
"src/gateway/server-methods/chat-transcript-inject.ts",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue