furyhawk/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-21 10:16:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: route auto-reply sessions through session seam (#89124)

Browse Source 
* clawdbot-6f0: route agent runtime session writes through seam

* clawdbot-6f0: route command entry persistence through seam

* test: ratchet auto-reply session accessor writes

* refactor: scope embedded attempt quota reads

* test: ratchet session store save writer
This commit is contained in:
Josh Lehman
2026-06-15 15:50:38 -07:00
committed by GitHub
parent 21e3cfa5e9
commit 127e174c9e
35 changed files with 1046 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check-session-accessor-boundary.mjs
+70 -13
View File
@@ -23,12 +23,24 @@ const legacyWholeStoreAccessNames = new Set([
"saveSessionStore",
"updateSessionStore",
]);
const legacyWriterNames = new Set([
"applySessionStoreEntryPatch",
"saveSessionStore",
"updateSessionStore",
"updateSessionStoreEntry",
]);
export const migratedSessionAccessorFiles = new Set([
"src/agents/embedded-agent-runner/compaction-successor-transcript.ts",
"src/agents/embedded-agent-runner/run/attempt.ts",
"src/agents/embedded-agent-runner/tool-result-truncation.ts",
"src/agents/embedded-agent-runner/transcript-rewrite.ts",
"src/agents/embedded-agent-runner/transcript-runtime-state.ts",
"src/auto-reply/reply/agent-runner-helpers.ts",
"src/auto-reply/reply/agent-runner.ts",
"src/auto-reply/reply/commands-subagents/action-info.ts",
"src/auto-reply/reply/followup-runner.ts",
"src/auto-reply/reply/queue/drain.ts",
"src/commands/export-trajectory.ts",
"src/commands/health.ts",
"src/commands/sandbox-explain.ts",
@@ -52,6 +64,30 @@ export const migratedBundledPluginSessionAccessorFiles = new Set([
"extensions/telegram/src/bot-handlers.runtime.ts",
]);
export const migratedSessionAccessorWriteFiles = new Set([
"src/agents/command/attempt-execution.shared.ts",
"src/agents/command/session-store.ts",
"src/agents/embedded-agent-runner/run.ts",
"src/agents/embedded-agent-runner/run/attempt.ts",
"src/auto-reply/reply/abort-cutoff.runtime.ts",
"src/auto-reply/reply/agent-runner-cli-dispatch.ts",
"src/auto-reply/reply/agent-runner-execution.ts",
"src/auto-reply/reply/agent-runner-memory.ts",
"src/auto-reply/reply/agent-runner.ts",
"src/auto-reply/reply/body.ts",
"src/auto-reply/reply/commands-acp/lifecycle.ts",
"src/auto-reply/reply/commands-reset.ts",
"src/auto-reply/reply/directive-handling.impl.ts",
"src/auto-reply/reply/directive-handling.persist.ts",
"src/auto-reply/reply/dispatch-from-config.runtime.ts",
"src/auto-reply/reply/followup-runner.ts",
"src/auto-reply/reply/get-reply.ts",
"src/auto-reply/reply/model-selection.ts",
"src/auto-reply/reply/session-reset-model.ts",
"src/auto-reply/reply/session-updates.ts",
"src/auto-reply/reply/session-usage.ts",
]);
function normalizeRelativePath(filePath) {
return filePath.replaceAll(path.sep, "/");
}
@@ -91,9 +127,8 @@ function bindingName(node) {
return null;
}
export function findSessionAccessorBoundaryViolations(content, fileName = "source.ts") {
function findNamedSessionStoreViolations(content, fileName, legacyNames, legacyKind) {
const sourceFile = ts.createSourceFile(fileName, content, ts.ScriptTarget.Latest, true);
const legacyNames = legacyNamesForFile(fileName);
const violations = [];
const visit = (node) => {
@@ -105,7 +140,7 @@ export function findSessionAccessorBoundaryViolations(content, fileName = "sourc
if (legacyNames.has(importedName)) {
violations.push({
line: toLine(sourceFile, specifier),
reason: `imports legacy session store access "${importedName}"`,
reason: `imports legacy session store ${legacyKind} "${importedName}"`,
});
}
}
@@ -117,7 +152,7 @@ export function findSessionAccessorBoundaryViolations(content, fileName = "sourc
if (name && legacyNames.has(name)) {
violations.push({
line: toLine(sourceFile, node),
reason: `aliases legacy session store access "${name}"`,
reason: `aliases legacy session store ${legacyKind} "${name}"`,
});
}
}
@@ -125,7 +160,7 @@ export function findSessionAccessorBoundaryViolations(content, fileName = "sourc
if (ts.isPropertyAccessExpression(node) && legacyNames.has(node.name.text)) {
violations.push({
line: toLine(sourceFile, node.name),
reason: `references legacy session store access "${node.name.text}"`,
reason: `references legacy session store ${legacyKind} "${node.name.text}"`,
});
}
@@ -136,7 +171,7 @@ export function findSessionAccessorBoundaryViolations(content, fileName = "sourc
) {
violations.push({
line: toLine(sourceFile, node.argumentExpression),
reason: `references legacy session store access "${node.argumentExpression.text}"`,
reason: `references legacy session store ${legacyKind} "${node.argumentExpression.text}"`,
});
}
@@ -149,7 +184,7 @@ export function findSessionAccessorBoundaryViolations(content, fileName = "sourc
) {
violations.push({
line: toLine(sourceFile, node.expression),
reason: `calls legacy session store access "${calleeName}"`,
reason: `calls legacy session store ${legacyKind} "${calleeName}"`,
});
}
}
@@ -161,21 +196,33 @@ export function findSessionAccessorBoundaryViolations(content, fileName = "sourc
return violations;
}
export function findSessionAccessorBoundaryViolations(content, fileName = "source.ts") {
const legacyNames = legacyNamesForFile(fileName);
const legacyKind = legacyNames === legacyWholeStoreAccessNames ? "access" : "reader";
return findNamedSessionStoreViolations(content, fileName, legacyNames, legacyKind);
}
export function findSessionAccessorWriteBoundaryViolations(content, fileName = "source.ts") {
return findNamedSessionStoreViolations(content, fileName, legacyWriterNames, "writer");
}
export async function main() {
const repoRoot = resolveRepoRoot(import.meta.url);
const sourceRoots = resolveSourceRoots(repoRoot, [
const readSourceRoots = resolveSourceRoots(repoRoot, [
"extensions/discord/src/monitor",
"extensions/telegram/src",
"src/agents/embedded-agent-runner",
"src/agents",
"src/auto-reply",
"src/commands",
"src/config/sessions",
"src/cron",
"src/gateway",
"src/infra",
]);
const violations = await collectFileViolations({
const writeSourceRoots = resolveSourceRoots(repoRoot, ["src/agents", "src/auto-reply"]);
const readViolations = await collectFileViolations({
repoRoot,
sourceRoots,
sourceRoots: readSourceRoots,
skipFile: (filePath) => {
const relativePath = normalizeRelativePath(path.relative(repoRoot, filePath));
return (
@@ -185,18 +232,28 @@ export async function main() {
},
findViolations: findSessionAccessorBoundaryViolations,
});
const writeViolations = await collectFileViolations({
repoRoot,
sourceRoots: writeSourceRoots,
skipFile: (filePath) =>
!migratedSessionAccessorWriteFiles.has(
normalizeRelativePath(path.relative(repoRoot, filePath)),
),
findViolations: findSessionAccessorWriteBoundaryViolations,
});
const violations = [...readViolations, ...writeViolations];
if (violations.length === 0) {
console.log("session accessor boundary guard passed.");
return;
}
console.error("Found legacy session store access usage in session-accessor migrated files:");
console.error("Found legacy session store usage in session-accessor migrated files:");
for (const violation of violations) {
console.error(`- ${violation.path}:${violation.line}: ${violation.reason}`);
}
console.error(
"Use src/config/sessions/session-accessor.ts helpers for migrated paths. Expand this ratchet only after a slice migrates more files.",
"Use src/config/sessions/session-accessor.ts helpers for migrated read/write paths. Expand this ratchet only after a slice migrates more files.",
);
process.exit(1);
}
src/agents/command/attempt-execution.shared.ts
+19 -14
View File
@@ -2,7 +2,7 @@
* Shared session persistence and prompt-body helpers for agent attempt
* execution paths.
*/
import { updateSessionStore } from "../../config/sessions/store.js";
import { patchSessionEntry } from "../../config/sessions/session-accessor.js";
import { mergeSessionEntry, type SessionEntry } from "../../config/sessions/types.js";
import {
formatAgentInternalEventsForPlainPrompt,
@@ -34,16 +34,19 @@ function normalizeTranscriptMarkerUpdatedAt(value: number | undefined): number |
export async function persistSessionEntry(
params: PersistSessionEntryParams,
): Promise<SessionEntry | undefined> {
const persisted = await updateSessionStore(
params.storePath,
(store) => {
const current = store[params.sessionKey];
if (params.shouldPersist && !params.shouldPersist(current)) {
return current;
let rejectedMissingEntry = false;
const persisted = await patchSessionEntry(
{ sessionKey: params.sessionKey, storePath: params.storePath },
(_entry, context) => {
if (params.shouldPersist && !params.shouldPersist(context.existingEntry)) {
rejectedMissingEntry = !context.existingEntry;
return null;
}
const merged = mergeSessionEntry(store[params.sessionKey], params.entry);
const merged = mergeSessionEntry(context.existingEntry, params.entry);
if (params.preserveTranscriptMarkerUpdatedAt) {
const currentUpdatedAt = normalizeTranscriptMarkerUpdatedAt(current?.updatedAt);
const currentUpdatedAt = normalizeTranscriptMarkerUpdatedAt(
context.existingEntry?.updatedAt,
);
const markerUpdatedAt = normalizeTranscriptMarkerUpdatedAt(params.entry.updatedAt);
if (markerUpdatedAt !== undefined) {
merged.updatedAt = Math.max(currentUpdatedAt ?? 0, markerUpdatedAt);
@@ -56,21 +59,23 @@ export async function persistSessionEntry(
Reflect.deleteProperty(merged, field);
}
}
store[params.sessionKey] = merged;
return merged;
},
{
resolveSingleEntryPersistence: (entry) =>
entry ? { sessionKey: params.sessionKey, entry } : null,
takeCacheOwnership: true,
fallbackEntry: params.sessionStore[params.sessionKey] ?? params.entry,
replaceEntry: true,
},
);
if (rejectedMissingEntry) {
delete params.sessionStore[params.sessionKey];
return undefined;
}
if (persisted) {
params.sessionStore[params.sessionKey] = persisted;
} else {
delete params.sessionStore[params.sessionKey];
}
return persisted;
return persisted ?? undefined;
}
/** Prepends hidden internal event context unless the body already carries it. */
src/agents/command/session-store.test.ts
+167 -24
View File
@@ -183,11 +183,21 @@ describe("updateSessionStoreAfterAgentRun", () => {
} as OpenClawConfig;
const sessionKey = "agent:main:explicit:test-maintenance-config";
const sessionId = "test-maintenance-config-session";
const now = Date.now();
const sessionStore: Record<string, SessionEntry> = {
[sessionKey]: {
sessionId,
updatedAt: 1,
updatedAt: now,
},
...Object.fromEntries(
Array.from({ length: 45 }, (_, index) => [
`agent:main:stale:${index}`,
{
sessionId: `stale-${index}`,
updatedAt: now - index - 1,
} satisfies SessionEntry,
]),
),
};
await fs.writeFile(storePath, JSON.stringify(sessionStore, null, 2));
const result: EmbeddedAgentRunResult = {
@@ -212,29 +222,10 @@ describe("updateSessionStoreAfterAgentRun", () => {
result,
});
// The gateway write takes cache ownership and supplies single-entry
// persistence so large stores are not rewritten unnecessarily.
const updateOptions = sessionStoreMocks.updateSessionStore.mock.calls.at(-1)?.[2];
expect(updateOptions).toMatchObject({
takeCacheOwnership: true,
maintenanceConfig: {
mode: "enforce",
maxEntries: 42,
},
});
expect(typeof updateOptions?.resolveSingleEntryPersistence).toBe("function");
expect(
updateOptions?.resolveSingleEntryPersistence?.({
sessionId,
updatedAt: 2,
} as SessionEntry),
).toEqual({
sessionKey,
entry: {
sessionId,
updatedAt: 2,
},
});
const persisted = loadSessionStore(storePath, { skipCache: true });
expect(Object.keys(persisted)).toHaveLength(42);
expect(persisted[sessionKey]?.sessionId).toBe(sessionId);
expect(persisted["agent:main:stale:44"]).toBeUndefined();
});
});
@@ -1858,6 +1849,52 @@ describe("updateSessionStoreAfterAgentRun", () => {
});
});
it("does not recreate a missing persisted row while preserving user-facing state", async () => {
await withTempSessionStore(async ({ storePath }) => {
const cfg = {} as OpenClawConfig;
const sessionKey = "agent:main:explicit:missing-visible-row";
const sessionId = "missing-visible-row-session";
const sessionStore: Record<string, SessionEntry> = {
[sessionKey]: {
sessionId,
updatedAt: 1,
modelProvider: "openai",
model: "gpt-5.5",
},
};
await fs.writeFile(storePath, JSON.stringify({}, null, 2), "utf8");
await updateSessionStoreAfterAgentRun({
cfg,
sessionId,
sessionKey,
storePath,
sessionStore,
defaultProvider: "claude-cli",
defaultModel: "claude-sonnet-4-6",
preserveUserFacingSessionModelState: true,
result: {
meta: {
durationMs: 1,
agentMeta: {
sessionId,
provider: "claude-cli",
model: "claude-sonnet-4-6",
},
},
},
});
expect(sessionStore[sessionKey]).toEqual({
sessionId,
updatedAt: 1,
modelProvider: "openai",
model: "gpt-5.5",
});
expect(loadSessionStore(storePath, { skipCache: true })[sessionKey]).toBeUndefined();
});
});
it("leaves contextTokens unset when entry has prior model but no contextTokens (heartbeat bleed guard)", async () => {
await withTempSessionStore(async ({ storePath }) => {
const cfg = {} as OpenClawConfig;
@@ -2212,6 +2249,58 @@ describe("recordCliCompactionInStore", () => {
expect(persisted[sessionKey]?.sessionFile).toBe(nextSessionFile);
});
});
it("recreates a complete persisted row when the caller snapshot survived a missing store row", async () => {
await withTempSessionStore(async ({ storePath }) => {
const sessionKey = "agent:main:explicit:test-record-cli-compaction-missing-row";
const sessionId = "test-record-cli-compaction-missing-row-session";
const sessionStore: Record<string, SessionEntry> = {
[sessionKey]: {
sessionId,
updatedAt: 1,
modelProvider: "openai",
model: "gpt-5.5",
totalTokens: 12_000,
totalTokensFresh: true,
inputTokens: 9_000,
outputTokens: 100,
cacheRead: 2_900,
cacheWrite: 0,
cliSessionBindings: {
codex: {
sessionId: "stale-cli-session",
},
},
cliSessionIds: {
codex: "stale-cli-session",
},
},
};
await fs.writeFile(storePath, JSON.stringify({}, null, 2), "utf8");
await recordCliCompactionInStore({
provider: "codex",
sessionKey,
sessionStore,
storePath,
tokensAfter: 42,
});
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
expect(sessionStore[sessionKey]?.sessionId).toBe(sessionId);
expect(sessionStore[sessionKey]?.modelProvider).toBe("openai");
expect(sessionStore[sessionKey]?.model).toBe("gpt-5.5");
expect(sessionStore[sessionKey]?.compactionCount).toBe(1);
expect(sessionStore[sessionKey]?.totalTokens).toBe(42);
expect(sessionStore[sessionKey]?.cliSessionBindings?.codex).toBeUndefined();
expect(persisted?.sessionId).toBe(sessionId);
expect(persisted?.modelProvider).toBe("openai");
expect(persisted?.model).toBe("gpt-5.5");
expect(persisted?.compactionCount).toBe(1);
expect(persisted?.totalTokens).toBe(42);
expect(persisted?.cliSessionBindings?.codex).toBeUndefined();
});
});
});
describe("clearCliSessionInStore", () => {
@@ -2292,4 +2381,58 @@ describe("clearCliSessionInStore", () => {
).toBe("claude-session-1");
});
});
it("clears the caller snapshot and recreates a complete persisted row when the store row is missing", async () => {
await withTempSessionStore(async ({ storePath }) => {
const sessionKey = "agent:main:explicit:test-clear-cli-missing-row";
const entry: SessionEntry = {
sessionId: "openclaw-session-1",
updatedAt: 1,
modelProvider: "anthropic",
model: "claude-opus-4-6",
cliSessionBindings: {
"claude-cli": {
sessionId: "claude-session-1",
authEpoch: "epoch-1",
},
"codex-cli": {
sessionId: "codex-session-1",
},
},
cliSessionIds: {
"claude-cli": "claude-session-1",
"codex-cli": "codex-session-1",
},
claudeCliSessionId: "claude-session-1",
};
const sessionStore: Record<string, SessionEntry> = { [sessionKey]: entry };
await fs.writeFile(storePath, JSON.stringify({}, null, 2), "utf8");
const cleared = await clearCliSessionInStore({
provider: "claude-cli",
sessionKey,
sessionStore,
storePath,
});
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
expect(cleared?.sessionId).toBe("openclaw-session-1");
expect(cleared?.modelProvider).toBe("anthropic");
expect(cleared?.model).toBe("claude-opus-4-6");
expect(cleared?.cliSessionBindings?.["claude-cli"]).toBeUndefined();
expect(cleared?.cliSessionBindings?.["codex-cli"]).toEqual({
sessionId: "codex-session-1",
});
expect(cleared?.claudeCliSessionId).toBeUndefined();
expect(sessionStore[sessionKey]).toEqual(cleared);
expect(persisted?.sessionId).toBe("openclaw-session-1");
expect(persisted?.modelProvider).toBe("anthropic");
expect(persisted?.model).toBe("claude-opus-4-6");
expect(persisted?.cliSessionBindings?.["claude-cli"]).toBeUndefined();
expect(persisted?.cliSessionBindings?.["codex-cli"]).toEqual({
sessionId: "codex-session-1",
});
expect(persisted?.claudeCliSessionId).toBeUndefined();
});
});
});
src/agents/command/session-store.ts
+35 -27
View File
@@ -5,14 +5,13 @@ import path from "node:path";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import {
canonicalizeAbsoluteSessionFilePath,
mergeSessionEntry,
resolveSessionFilePath,
resolveSessionFilePathOptions,
setSessionRuntimeModel,
type SessionEntry,
updateSessionStore,
rewriteSessionFileForNewSessionId,
} from "../../config/sessions.js";
import { patchSessionEntry } from "../../config/sessions/session-accessor.js";
import { resolveMaintenanceConfigFromInput } from "../../config/sessions/store-maintenance.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js";
@@ -294,21 +293,20 @@ export async function updateSessionStoreAfterAgentRun(params: {
}
: removeLifecycleStateFromMetadataPatch(next);
const maintenanceConfig = resolveMaintenanceConfigFromInput(cfg.session?.maintenance);
const persisted = await updateSessionStore(
storePath,
(store) => {
if (preserveUserFacingRunState && !store[sessionKey]) {
return undefined;
const persisted = await patchSessionEntry(
{
storePath,
sessionKey,
},
(_currentEntry, context) => {
if (preserveUserFacingRunState && !context.existingEntry) {
return null;
}
const merged = mergeSessionEntry(store[sessionKey], metadataPatch);
store[sessionKey] = merged;
return merged;
return metadataPatch;
},
{
takeCacheOwnership: true,
...(preserveUserFacingRunState ? {} : { fallbackEntry: entry }),
maintenanceConfig,
resolveSingleEntryPersistence: (entryLocal) =>
entryLocal ? { sessionKey, entry: entryLocal } : undefined,
},
);
if (persisted) {
@@ -333,13 +331,18 @@ export async function clearCliSessionInStore(params: {
clearCliSession(next, provider);
next.updatedAt = Date.now();
const persisted = await updateSessionStore(storePath, (store) => {
const merged = mergeSessionEntry(store[sessionKey], next);
store[sessionKey] = merged;
return merged;
});
sessionStore[sessionKey] = persisted;
return persisted;
const persisted = await patchSessionEntry(
{
storePath,
sessionKey,
},
() => next,
{ fallbackEntry: entry },
);
if (persisted) {
sessionStore[sessionKey] = persisted;
}
return persisted ?? undefined;
}
/** Records CLI compaction metadata on the persisted session entry. */
@@ -402,13 +405,18 @@ export async function recordCliCompactionInStore(params: {
next.cacheWrite = undefined;
}
const persisted = await updateSessionStore(storePath, (store) => {
const merged = mergeSessionEntry(store[sessionKey], next);
store[sessionKey] = merged;
return merged;
});
sessionStore[sessionKey] = persisted;
return persisted;
const persisted = await patchSessionEntry(
{
storePath,
sessionKey,
},
() => next,
{ fallbackEntry: entry },
);
if (persisted) {
sessionStore[sessionKey] = persisted;
}
return persisted ?? undefined;
}
function resolveCompactionSessionFile(params: {
src/agents/embedded-agent-runner/run.ts
+13 -8
View File
@@ -8,7 +8,8 @@ import { sanitizeForLog } from "../../../packages/terminal-core/src/ansi.js";
import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
import type { ThinkLevel } from "../../auto-reply/thinking.js";
import { SILENT_REPLY_TOKEN } from "../../auto-reply/tokens.js";
import { resolveStorePath, updateSessionStoreEntry } from "../../config/sessions.js";
import { resolveStorePath } from "../../config/sessions.js";
import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
import { ensureContextEnginesInitialized } from "../../context-engine/init.js";
import {
resolveContextEngine,
@@ -277,12 +278,12 @@ async function resetNoRealConversationTokenSnapshot(params: {
}
const storePath = resolveStorePath(params.config?.session?.store, { agentId: params.agentId });
try {
await updateSessionStoreEntry({
storePath,
sessionKey: params.sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
update: async () => ({
await updateSessionEntry(
{
storePath,
sessionKey: params.sessionKey,
},
async () => ({
totalTokens: 0,
totalTokensFresh: true,
inputTokens: undefined,
@@ -292,7 +293,11 @@ async function resetNoRealConversationTokenSnapshot(params: {
contextBudgetStatus: undefined,
updatedAt: Date.now(),
}),
});
{
skipMaintenance: true,
takeCacheOwnership: true,
},
);
} catch (err) {
log.warn(
`[context-overflow-precheck] failed to reset stale context snapshot for ` +
src/agents/embedded-agent-runner/run/attempt.ts
+59 -16
View File
@@ -13,15 +13,17 @@ import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../../../auto-reply/token
import { getRuntimeConfig } from "../../../config/config.js";
import { resolveStorePath } from "../../../config/sessions/paths.js";
import {
loadSessionStore,
runQuotaSuspensionMaintenance,
updateSessionStoreEntry,
} from "../../../config/sessions/store.js";
listSessionEntries,
loadSessionEntry,
updateSessionEntry,
} from "../../../config/sessions/session-accessor.js";
import { resolveQuotaSuspensionEntryMaintenance } from "../../../config/sessions/store-maintenance.js";
import {
bindOwnedSessionTranscriptWrites,
type OwnedSessionTranscriptCacheSnapshot,
withOwnedSessionTranscriptWrites,
} from "../../../config/sessions/transcript-write-context.js";
import type { SessionEntry } from "../../../config/sessions/types.js";
import {
assertContextEngineHostSupport,
OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST,
@@ -816,6 +818,41 @@ function collectAttemptExplicitToolAllowlistSources(params: {
]);
}
// Applies quota-resume TTL maintenance to only the active attempt session.
async function loadAttemptSessionEntryAfterQuotaMaintenance(params: {
storePath: string;
sessionKey: string;
}): Promise<SessionEntry | undefined> {
const entry = loadSessionEntry({
storePath: params.storePath,
sessionKey: params.sessionKey,
});
if (!entry?.quotaSuspension) {
return entry;
}
const now = Date.now();
const maintenance = resolveQuotaSuspensionEntryMaintenance({ entry, now });
if (!maintenance.patch) {
return entry;
}
const updated = await updateSessionEntry(
{
storePath: params.storePath,
sessionKey: params.sessionKey,
},
(currentEntry) =>
resolveQuotaSuspensionEntryMaintenance({
entry: currentEntry,
now,
}).patch,
{
skipMaintenance: true,
takeCacheOwnership: true,
},
);
return updated ?? entry;
}
export async function runEmbeddedAttempt(
params: EmbeddedRunAttemptParams,
): Promise<EmbeddedRunAttemptResult> {
@@ -3153,12 +3190,14 @@ export async function runEmbeddedAttempt(
const storePath = resolveStorePath(params.config?.session?.store, {
agentId: sessionAgentId,
});
await runQuotaSuspensionMaintenance({ storePath });
const store = loadSessionStore(storePath, { skipCache: true });
const sessionEntry = store[params.sessionKey];
const sessionEntry = await loadAttemptSessionEntryAfterQuotaMaintenance({
storePath,
sessionKey: params.sessionKey,
});
const suspension = sessionEntry?.quotaSuspension;
if (suspension?.state === "resuming") {
const subagents = Object.values(store)
if (sessionEntry && suspension?.state === "resuming") {
const subagents = listSessionEntries({ storePath, clone: false })
.map(({ entry }) => entry)
.filter((s) => s.spawnedBy === sessionEntry.sessionId)
.map((s) => ({
sessionId: s.sessionId,
@@ -3170,12 +3209,12 @@ export async function runEmbeddedAttempt(
activeSubagents: subagents,
});
validated.push(handoffMsg);
await updateSessionStoreEntry({
storePath,
sessionKey: params.sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
update: async (entry) => {
await updateSessionEntry(
{
storePath,
sessionKey: params.sessionKey,
},
async (entry) => {
if (entry.quotaSuspension?.state !== "resuming") {
return null;
}
@@ -3183,7 +3222,11 @@ export async function runEmbeddedAttempt(
quotaSuspension: { ...entry.quotaSuspension, state: "active" },
};
},
});
{
skipMaintenance: true,
takeCacheOwnership: true,
},
);
}
}
src/auto-reply/reply/abort-cutoff.runtime.test.ts
+54
View File
@@ -0,0 +1,54 @@
// Covers abort-cutoff clearing against missing persisted rows.
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { describe, expect, it } from "vitest";
import { loadSessionStore } from "../../config/sessions.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import { clearAbortCutoffInSessionRuntime } from "./abort-cutoff.runtime.js";
async function withTempStore<T>(run: (storePath: string) => Promise<T>): Promise<T> {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-abort-cutoff-"));
try {
return await run(path.join(dir, "sessions.json"));
} finally {
await fs.rm(dir, { recursive: true, force: true });
}
}
describe("clearAbortCutoffInSessionRuntime", () => {
it("recreates a complete persisted row when clearing abort cutoff state", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:explicit:cutoff-missing-row";
const entry: SessionEntry = {
sessionId: "cutoff-session",
updatedAt: 1,
modelProvider: "anthropic",
model: "claude-opus-4-6",
abortCutoffMessageSid: "msg-42",
abortCutoffTimestamp: 123,
};
const sessionStore: Record<string, SessionEntry> = { [sessionKey]: entry };
await fs.writeFile(storePath, JSON.stringify({}, null, 2), "utf8");
const cleared = await clearAbortCutoffInSessionRuntime({
sessionEntry: entry,
sessionStore,
sessionKey,
storePath,
});
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
expect(cleared).toBe(true);
expect(sessionStore[sessionKey]?.sessionId).toBe("cutoff-session");
expect(sessionStore[sessionKey]?.modelProvider).toBe("anthropic");
expect(sessionStore[sessionKey]?.abortCutoffMessageSid).toBeUndefined();
expect(sessionStore[sessionKey]?.abortCutoffTimestamp).toBeUndefined();
expect(persisted?.sessionId).toBe("cutoff-session");
expect(persisted?.modelProvider).toBe("anthropic");
expect(persisted?.model).toBe("claude-opus-4-6");
expect(persisted?.abortCutoffMessageSid).toBeUndefined();
expect(persisted?.abortCutoffTimestamp).toBeUndefined();
});
});
});
src/auto-reply/reply/abort-cutoff.runtime.ts
+12 -11
View File
@@ -1,5 +1,5 @@
/** Runtime persistence helper for clearing abort-cutoff state from sessions. */
import { updateSessionStore } from "../../config/sessions/store.js";
import { patchSessionEntry } from "../../config/sessions/session-accessor.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import { applyAbortCutoffToSessionEntry, hasAbortCutoff } from "./abort-cutoff.js";
@@ -16,19 +16,20 @@ export async function clearAbortCutoffInSessionRuntime(params: {
}
applyAbortCutoffToSessionEntry(sessionEntry, undefined);
sessionEntry.updatedAt = Date.now();
const updatedAt = Date.now();
sessionEntry.updatedAt = updatedAt;
sessionStore[sessionKey] = sessionEntry;
if (storePath) {
await updateSessionStore(storePath, (store) => {
const existing = store[sessionKey] ?? sessionEntry;
if (!existing) {
return;
}
applyAbortCutoffToSessionEntry(existing, undefined);
existing.updatedAt = Date.now();
store[sessionKey] = existing;
});
await patchSessionEntry(
{ storePath, sessionKey },
() => ({
abortCutoffMessageSid: undefined,
abortCutoffTimestamp: undefined,
updatedAt,
}),
{ fallbackEntry: sessionEntry },
);
}
return true;
src/auto-reply/reply/agent-runner-cli-dispatch.ts
+9 -4
View File
@@ -14,7 +14,8 @@ import {
isAgentRunRestartAbortReason,
resolveAgentRunAbortLifecycleFields,
} from "../../agents/run-termination.js";
import { updateSessionStore, type SessionEntry } from "../../config/sessions.js";
import type { SessionEntry } from "../../config/sessions.js";
import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
import type { AgentEventPayload } from "../../infra/agent-events.js";
import {
emitAgentEvent,
@@ -189,9 +190,13 @@ export async function clearDroppedCliSessionBinding(params: {
if (!params.storePath || !params.sessionKey) {
return;
}
await updateSessionStore(params.storePath, (store) => {
clearEntry(store[params.sessionKey!]);
});
await updateSessionEntry(
{ storePath: params.storePath, sessionKey: params.sessionKey },
(entry) => {
clearEntry(entry);
return entry;
},
);
}
function createToolEventBridge(params: {
src/auto-reply/reply/agent-runner-execution.ts
+57 -34
View File
@@ -65,11 +65,8 @@ import {
resolveAgentRunAbortLifecycleFields,
} from "../../agents/run-termination.js";
import { buildAgentRuntimeOutcomePlan } from "../../agents/runtime-plan/build.js";
import {
resolveGroupSessionKey,
type SessionEntry,
updateSessionStore,
} from "../../config/sessions.js";
import { resolveGroupSessionKey, type SessionEntry } from "../../config/sessions.js";
import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
import { resolveSilentReplyPolicy } from "../../config/silent-reply.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { logVerbose } from "../../globals.js";
@@ -505,6 +502,13 @@ function snapshotFallbackSelectionState(entry: SessionEntry): FallbackSelectionS
};
}
function buildFallbackSelectionStatePatch(entry: SessionEntry): Partial<SessionEntry> {
return {
...snapshotFallbackSelectionState(entry),
updatedAt: entry.updatedAt,
};
}
function buildFallbackSelectionState(params: {
provider: string;
model: string;
@@ -1846,14 +1850,13 @@ export async function runAgentTurnWithFallback(params: {
try {
if (params.storePath) {
await updateSessionStore(params.storePath, (store) => {
const persistedEntry = store[params.sessionKey!];
if (!persistedEntry) {
return;
}
applyFallbackSelectionState(persistedEntry, nextState);
store[params.sessionKey!] = persistedEntry;
});
await updateSessionEntry(
{ storePath: params.storePath, sessionKey: params.sessionKey },
(persistedEntry) => {
applyFallbackSelectionState(persistedEntry, nextState);
return buildFallbackSelectionStatePatch(persistedEntry);
},
);
}
} catch (error) {
rollbackFallbackSelectionStateIfUnchanged(activeSessionEntry, nextState, previousState);
@@ -1870,18 +1873,18 @@ export async function runAgentTurnWithFallback(params: {
if (rolledBackInMemory) {
params.activeSessionStore![params.sessionKey!] = activeSessionEntry;
}
if (!params.storePath) {
if (!params.storePath || !params.sessionKey) {
return;
}
await updateSessionStore(params.storePath, (store) => {
const persistedEntry = store[params.sessionKey!];
if (!persistedEntry) {
return;
}
if (rollbackFallbackSelectionStateIfUnchanged(persistedEntry, nextState, previousState)) {
store[params.sessionKey!] = persistedEntry;
}
});
await updateSessionEntry(
{ storePath: params.storePath, sessionKey: params.sessionKey },
(persistedEntry) => {
if (rollbackFallbackSelectionStateIfUnchanged(persistedEntry, nextState, previousState)) {
return buildFallbackSelectionStatePatch(persistedEntry);
}
return null;
},
);
};
};
const clearRecoveredAutoFallbackPrimaryProbe = async (paramsForClear: {
@@ -1914,17 +1917,37 @@ export async function runAgentTurnWithFallback(params: {
if (!params.storePath) {
return;
}
await updateSessionStore(params.storePath, (store) => {
const persistedEntry = store[params.sessionKey!];
if (!persistedEntry) {
return;
}
if (!entryMatchesAutoFallbackPrimaryProbe(persistedEntry, probe)) {
return;
}
clearAutoFallbackPrimaryProbeSelection(persistedEntry);
store[params.sessionKey!] = persistedEntry;
});
await updateSessionEntry(
{ storePath: params.storePath, sessionKey: params.sessionKey },
(persistedEntry) => {
if (!entryMatchesAutoFallbackPrimaryProbe(persistedEntry, probe)) {
return null;
}
const shouldClearAuthProfile =
persistedEntry.authProfileOverrideSource === "auto" ||
(persistedEntry.authProfileOverrideSource === undefined &&
persistedEntry.authProfileOverrideCompactionCount !== undefined);
clearAutoFallbackPrimaryProbeSelection(persistedEntry);
return {
providerOverride: undefined,
modelOverride: undefined,
modelOverrideSource: undefined,
modelOverrideFallbackOriginProvider: undefined,
modelOverrideFallbackOriginModel: undefined,
...(shouldClearAuthProfile
? {
authProfileOverride: undefined,
authProfileOverrideSource: undefined,
authProfileOverrideCompactionCount: undefined,
}
: {}),
fallbackNoticeSelectedModel: undefined,
fallbackNoticeActiveModel: undefined,
fallbackNoticeReason: undefined,
updatedAt: persistedEntry.updatedAt,
};
},
);
};
while (true) {
src/auto-reply/reply/agent-runner-helpers.test.ts
+18 -24
View File
@@ -4,17 +4,17 @@ import type { ReplyPayload } from "../types.js";
import type { TypingSignaler } from "./typing-mode.js";
const hoisted = vi.hoisted(() => {
const loadSessionStoreMock = vi.fn();
return { loadSessionStoreMock };
const loadSessionEntryMock = vi.fn();
return { loadSessionEntryMock };
});
vi.mock("../../config/sessions.js", async () => {
const actual = await vi.importActual<typeof import("../../config/sessions.js")>(
"../../config/sessions.js",
vi.mock("../../config/sessions/session-accessor.js", async () => {
const actual = await vi.importActual<typeof import("../../config/sessions/session-accessor.js")>(
"../../config/sessions/session-accessor.js",
);
return {
...actual,
loadSessionStore: (...args: unknown[]) => hoisted.loadSessionStoreMock(...args),
loadSessionEntry: (...args: unknown[]) => hoisted.loadSessionEntryMock(...args),
};
});
@@ -28,7 +28,7 @@ const {
describe("agent runner helpers", () => {
beforeEach(() => {
vi.useRealTimers();
hoisted.loadSessionStoreMock.mockReset();
hoisted.loadSessionEntryMock.mockReset();
});
it("detects audio payloads from mediaUrl/mediaUrls", () => {
@@ -45,9 +45,7 @@ describe("agent runner helpers", () => {
});
it("uses session verbose level when present", () => {
hoisted.loadSessionStoreMock.mockReturnValue({
"agent:main:main": { verboseLevel: "full" },
});
hoisted.loadSessionEntryMock.mockReturnValue({ verboseLevel: "full" });
const shouldEmitResult = createShouldEmitToolResult({
sessionKey: "agent:main:main",
storePath: "/tmp/store.json",
@@ -60,7 +58,9 @@ describe("agent runner helpers", () => {
});
expect(shouldEmitResult()).toBe(true);
expect(shouldEmitOutput()).toBe(true);
expect(hoisted.loadSessionStoreMock).toHaveBeenCalledWith("/tmp/store.json", {
expect(hoisted.loadSessionEntryMock).toHaveBeenCalledWith({
sessionKey: "agent:main:main",
storePath: "/tmp/store.json",
clone: false,
});
});
@@ -68,9 +68,7 @@ describe("agent runner helpers", () => {
it("caches session verbose reads briefly while still refreshing live changes", () => {
vi.useFakeTimers();
vi.setSystemTime(1_000);
hoisted.loadSessionStoreMock.mockReturnValue({
"agent:main:main": { verboseLevel: "full" },
});
hoisted.loadSessionEntryMock.mockReturnValue({ verboseLevel: "full" });
const shouldEmitOutput = createShouldEmitToolOutput({
sessionKey: "agent:main:main",
storePath: "/tmp/store.json",
@@ -78,19 +76,17 @@ describe("agent runner helpers", () => {
});
expect(shouldEmitOutput()).toBe(true);
hoisted.loadSessionStoreMock.mockReturnValue({
"agent:main:main": { verboseLevel: "off" },
});
hoisted.loadSessionEntryMock.mockReturnValue({ verboseLevel: "off" });
expect(shouldEmitOutput()).toBe(true);
expect(hoisted.loadSessionStoreMock).toHaveBeenCalledOnce();
expect(hoisted.loadSessionEntryMock).toHaveBeenCalledOnce();
vi.setSystemTime(1_251);
expect(shouldEmitOutput()).toBe(false);
expect(hoisted.loadSessionStoreMock).toHaveBeenCalledTimes(2);
expect(hoisted.loadSessionEntryMock).toHaveBeenCalledTimes(2);
});
it("falls back when store read fails or session value is invalid", () => {
hoisted.loadSessionStoreMock.mockImplementation(() => {
hoisted.loadSessionEntryMock.mockImplementation(() => {
throw new Error("boom");
});
const fallbackOn = createShouldEmitToolResult({
@@ -100,10 +96,8 @@ describe("agent runner helpers", () => {
});
expect(fallbackOn()).toBe(true);
hoisted.loadSessionStoreMock.mockClear();
hoisted.loadSessionStoreMock.mockReturnValue({
"agent:main:main": { verboseLevel: "weird" },
});
hoisted.loadSessionEntryMock.mockClear();
hoisted.loadSessionEntryMock.mockReturnValue({ verboseLevel: "weird" });
const fallbackFull = createShouldEmitToolOutput({
sessionKey: "agent:main:main",
storePath: "/tmp/store.json",
src/auto-reply/reply/agent-runner-helpers.ts
+6 -3
View File
@@ -4,7 +4,7 @@ import {
hasOutboundReplyContent,
resolveSendableOutboundReplyParts,
} from "openclaw/plugin-sdk/reply-payload";
import { loadSessionStore } from "../../config/sessions.js";
import { loadSessionEntry } from "../../config/sessions/session-accessor.js";
import { normalizeVerboseLevel, type VerboseLevel } from "../thinking.js";
import type { ReplyPayload } from "../types.js";
import type { TypingSignaler } from "./typing-mode.js";
@@ -29,8 +29,11 @@ function readCurrentVerboseLevel(params: VerboseGateParams): VerboseLevel | unde
return undefined;
}
try {
const store = loadSessionStore(params.storePath, { clone: false });
const entry = store[params.sessionKey];
const entry = loadSessionEntry({
storePath: params.storePath,
sessionKey: params.sessionKey,
clone: false,
});
return typeof entry?.verboseLevel === "string"
? normalizeVerboseLevel(entry.verboseLevel)
: undefined;
src/auto-reply/reply/agent-runner-memory.ts
+42 -14
View File
@@ -29,9 +29,8 @@ import {
resolveSessionFilePath,
resolveSessionFilePathOptions,
type SessionEntry,
applySessionStoreEntryPatch,
updateSessionStoreEntry,
} from "../../config/sessions.js";
import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { readSessionMessagesAsync } from "../../gateway/session-utils.fs.js";
import { logVerbose } from "../../globals.js";
@@ -64,6 +63,15 @@ import type { ReplyOperation } from "./reply-run-registry.js";
import { incrementCompactionCount } from "./session-updates.js";
type EmbeddedAgentRuntime = typeof import("../../agents/embedded-agent.js");
type UpdateSessionEntryParams = {
storePath: string;
sessionKey: string;
skipMaintenance?: boolean;
takeCacheOwnership?: boolean;
update: (
entry: SessionEntry,
) => Promise<Partial<SessionEntry> | null> | Partial<SessionEntry> | null;
};
const MAX_VISIBLE_MEMORY_FLUSH_ERROR_CHARS = 600;
const MAX_FLUSH_FAILURES = 3;
@@ -93,6 +101,22 @@ async function runEmbeddedAgentDefault(
return await runEmbeddedAgent(...args);
}
async function updateSessionEntryDefault(
params: UpdateSessionEntryParams,
): Promise<SessionEntry | null> {
return await updateSessionEntry(
{
storePath: params.storePath,
sessionKey: params.sessionKey,
},
params.update,
{
skipMaintenance: params.skipMaintenance,
takeCacheOwnership: params.takeCacheOwnership,
},
);
}
async function ensureMemoryFlushTargetFile(params: {
workspaceDir: string;
relativePath: string;
@@ -126,7 +150,7 @@ const memoryDeps = {
registerAgentRunContext,
refreshQueuedFollowupSession,
incrementCompactionCount,
updateSessionStoreEntry,
updateSessionEntry: updateSessionEntryDefault,
emitAgentEvent,
randomUUID: () => crypto.randomUUID(),
now: () => Date.now(),
@@ -143,7 +167,7 @@ export function setAgentRunnerMemoryTestDeps(overrides?: Partial<typeof memoryDe
registerAgentRunContext,
refreshQueuedFollowupSession,
incrementCompactionCount,
updateSessionStoreEntry,
updateSessionEntry: updateSessionEntryDefault,
emitAgentEvent,
randomUUID: () => crypto.randomUUID(),
now: () => Date.now(),
@@ -1164,13 +1188,17 @@ export async function runMemoryFlushIfNeeded(params: {
}
if (params.storePath && params.sessionKey) {
try {
const updatedEntry = await applySessionStoreEntryPatch({
storePath: params.storePath,
sessionKey: params.sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
patch: { totalTokens: transcriptPromptTokens, totalTokensFresh: true },
});
const updatedEntry = await updateSessionEntry(
{
storePath: params.storePath,
sessionKey: params.sessionKey,
},
() => ({ totalTokens: transcriptPromptTokens, totalTokensFresh: true }),
{
skipMaintenance: true,
takeCacheOwnership: true,
},
);
if (updatedEntry) {
entry = updatedEntry;
if (params.sessionStore) {
@@ -1395,7 +1423,7 @@ export async function runMemoryFlushIfNeeded(params: {
}
if (params.storePath && params.sessionKey) {
try {
const updatedEntry = await memoryDeps.updateSessionStoreEntry({
const updatedEntry = await memoryDeps.updateSessionEntry({
storePath: params.storePath,
sessionKey: params.sessionKey,
skipMaintenance: true,
@@ -1425,7 +1453,7 @@ export async function runMemoryFlushIfNeeded(params: {
if (!isAbortError(err) && params.storePath && params.sessionKey) {
try {
const failedAt = memoryDeps.now();
const failedEntry = await memoryDeps.updateSessionStoreEntry({
const failedEntry = await memoryDeps.updateSessionEntry({
storePath: params.storePath,
sessionKey: params.sessionKey,
skipMaintenance: true,
@@ -1473,7 +1501,7 @@ export async function runMemoryFlushIfNeeded(params: {
maxAttempts: MAX_FLUSH_FAILURES,
},
});
const exhaustedEntry = await memoryDeps.updateSessionStoreEntry({
const exhaustedEntry = await memoryDeps.updateSessionEntry({
storePath: params.storePath,
sessionKey: params.sessionKey,
skipMaintenance: true,
src/auto-reply/reply/agent-runner.ts
+44 -40
View File
@@ -23,13 +23,11 @@ import { deriveContextPromptTokens, hasNonzeroUsage, normalizeUsage } from "../.
import { enqueueCommitmentExtraction } from "../../commitments/runtime.js";
import type { OpenClawConfig } from "../../config/config.js";
import {
applySessionStoreEntryPatch,
loadSessionStore,
resolveSessionPluginStatusLines,
resolveSessionPluginTraceLines,
type SessionEntry,
updateSessionStoreEntry,
} from "../../config/sessions.js";
import { loadSessionEntry, updateSessionEntry } from "../../config/sessions/session-accessor.js";
import { parseSessionThreadInfoFast } from "../../config/sessions/thread-info.js";
import type { TypingMode } from "../../config/types.js";
import { resolveSessionTranscriptCandidates } from "../../gateway/session-utils.fs.js";
@@ -1109,8 +1107,10 @@ function refreshSessionEntryFromStore(params: {
return fallbackEntry;
}
try {
const latestStore = loadSessionStore(storePath, { skipCache: true, clone: false });
const latestEntry = latestStore?.[sessionKey];
const latestEntry = loadSessionEntry({
storePath,
sessionKey,
});
if (!latestEntry) {
return fallbackEntry;
}
@@ -1245,12 +1245,9 @@ export async function runReplyAgent(params: {
activeSessionEntry.updatedAt = updatedAt;
activeSessionStore[sessionKey] = activeSessionEntry;
if (storePath) {
await applySessionStoreEntryPatch({
storePath,
sessionKey,
await updateSessionEntry({ storePath, sessionKey }, () => ({ updatedAt }), {
skipMaintenance: true,
takeCacheOwnership: true,
patch: { updatedAt },
});
}
};
@@ -1490,14 +1487,16 @@ export async function runReplyAgent(params: {
restartRecoveryDeliveryRunId,
updatedAt,
};
const persisted = await updateSessionStoreEntry({
storePath,
sessionKey,
update: async (current) =>
const persisted = await updateSessionEntry(
{
storePath,
sessionKey,
},
async (current) =>
current.sessionId === replyOperation.sessionId && current.abortedLastRun !== true
? patch
: null,
});
);
if (persisted) {
activeSessionEntry = persisted;
if (activeSessionStore) {
@@ -1516,16 +1515,18 @@ export async function runReplyAgent(params: {
restartRecoveryDeliveryRunId: undefined,
updatedAt: Date.now(),
};
const persisted = await updateSessionStoreEntry({
storePath,
sessionKey,
update: async (current) =>
const persisted = await updateSessionEntry(
{
storePath,
sessionKey,
},
async (current) =>
current.sessionId === replyOperation.sessionId &&
current.abortedLastRun !== true &&
current.restartRecoveryDeliveryRunId === restartRecoveryDeliveryRunId
? patch
: null,
});
);
if (persisted) {
activeSessionEntry = persisted;
if (activeSessionStore) {
@@ -1742,16 +1743,17 @@ export async function runReplyAgent(params: {
activeSessionEntry.updatedAt = updatedAt;
activeSessionStore[sessionKey] = activeSessionEntry;
if (storePath) {
await applySessionStoreEntryPatch({
storePath,
sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
patch: {
await updateSessionEntry(
{ storePath, sessionKey },
() => ({
groupActivationNeedsSystemIntro: false,
updatedAt,
}),
{
skipMaintenance: true,
takeCacheOwnership: true,
},
});
);
}
}
@@ -1887,17 +1889,18 @@ export async function runReplyAgent(params: {
activeSessionStore[sessionKey] = fallbackStateEntry;
}
if (sessionKey && storePath) {
await applySessionStoreEntryPatch({
storePath,
sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
patch: {
await updateSessionEntry(
{ storePath, sessionKey },
() => ({
fallbackNoticeSelectedModel: fallbackTransition.nextState.selectedModel,
fallbackNoticeActiveModel: fallbackTransition.nextState.activeModel,
fallbackNoticeReason: fallbackTransition.nextState.reason,
}),
{
skipMaintenance: true,
takeCacheOwnership: true,
},
});
);
}
}
const usedCliProvider = isCliProvider(providerUsed, cfg);
@@ -2530,19 +2533,20 @@ export async function runReplyAgent(params: {
runtimePolicySessionKey,
opts,
});
await applySessionStoreEntryPatch({
storePath,
sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
patch: {
await updateSessionEntry(
{ storePath, sessionKey },
() => ({
pendingFinalDelivery: true,
pendingFinalDeliveryText: resolvedPendingText,
pendingFinalDeliveryContext,
pendingFinalDeliveryCreatedAt: Date.now(),
updatedAt: Date.now(),
}),
{
skipMaintenance: true,
takeCacheOwnership: true,
},
});
);
}
}
src/auto-reply/reply/body.test.ts
+53
View File
@@ -0,0 +1,53 @@
// Covers one-shot session hint persistence against missing persisted rows.
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { describe, expect, it } from "vitest";
import { loadSessionStore } from "../../config/sessions.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import { applySessionHints } from "./body.js";
async function withTempStore<T>(run: (storePath: string) => Promise<T>): Promise<T> {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-session-hints-"));
try {
return await run(path.join(dir, "sessions.json"));
} finally {
await fs.rm(dir, { recursive: true, force: true });
}
}
describe("applySessionHints", () => {
it("recreates a complete persisted row when clearing a consumed abort hint", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:explicit:hint-missing-row";
const entry: SessionEntry = {
sessionId: "hint-session",
updatedAt: 1,
modelProvider: "openai",
model: "gpt-5.5",
abortedLastRun: true,
};
const sessionStore: Record<string, SessionEntry> = { [sessionKey]: entry };
await fs.writeFile(storePath, JSON.stringify({}, null, 2), "utf8");
const body = await applySessionHints({
baseBody: "continue",
abortedLastRun: true,
sessionEntry: entry,
sessionStore,
sessionKey,
storePath,
});
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
expect(body).toContain("previous agent run was aborted");
expect(sessionStore[sessionKey]?.sessionId).toBe("hint-session");
expect(sessionStore[sessionKey]?.modelProvider).toBe("openai");
expect(sessionStore[sessionKey]?.abortedLastRun).toBe(false);
expect(persisted?.sessionId).toBe("hint-session");
expect(persisted?.modelProvider).toBe("openai");
expect(persisted?.model).toBe("gpt-5.5");
expect(persisted?.abortedLastRun).toBe(false);
});
});
});
src/auto-reply/reply/body.ts
+17 -16
View File
@@ -3,12 +3,12 @@ import type { SessionEntry } from "../../config/sessions/types.js";
import { createLazyImportLoader } from "../../shared/lazy-promise.js";
import { setAbortMemory } from "./abort-primitives.js";
const sessionStoreRuntimeLoader = createLazyImportLoader(
() => import("../../config/sessions/store.runtime.js"),
const sessionAccessorRuntimeLoader = createLazyImportLoader(
() => import("../../config/sessions/session-accessor.js"),
);
function loadSessionStoreRuntime() {
return sessionStoreRuntimeLoader.load();
function loadSessionAccessorRuntime() {
return sessionAccessorRuntimeLoader.load();
}
/** Applies one-shot session hints to the agent-visible body and clears consumed flags. */
@@ -29,23 +29,24 @@ export async function applySessionHints(params: {
prefixedBodyBase = `${abortedHint}\n\n${prefixedBodyBase}`;
// The abort hint is one-shot; clear durable state once it is added.
if (params.sessionEntry && params.sessionStore && params.sessionKey) {
const updatedAt = Date.now();
params.sessionEntry.abortedLastRun = false;
params.sessionEntry.updatedAt = Date.now();
params.sessionEntry.updatedAt = updatedAt;
params.sessionStore[params.sessionKey] = params.sessionEntry;
if (params.storePath) {
const sessionKey = params.sessionKey;
const { updateSessionStore } = await loadSessionStoreRuntime();
await updateSessionStore(params.storePath, (store) => {
const entry = store[sessionKey] ?? params.sessionEntry;
if (!entry) {
return;
}
store[sessionKey] = {
...entry,
const { patchSessionEntry } = await loadSessionAccessorRuntime();
await patchSessionEntry(
{
storePath: params.storePath,
sessionKey,
},
() => ({
abortedLastRun: false,
updatedAt: Date.now(),
};
});
updatedAt,
}),
{ fallbackEntry: params.sessionEntry },
);
}
} else if (params.abortKey) {
setAbortMemory(params.abortKey, false);
src/auto-reply/reply/commands-acp/lifecycle.ts
+9 -10
View File
@@ -36,7 +36,7 @@ import {
resolveThreadBindingPlacementForCurrentContext,
resolveThreadBindingSpawnPolicy,
} from "../../../channels/thread-bindings-policy.js";
import { updateSessionStore } from "../../../config/sessions.js";
import { updateSessionEntry } from "../../../config/sessions/session-accessor.js";
import type { SessionAcpMeta } from "../../../config/sessions/types.js";
import type { OpenClawConfig } from "../../../config/types.openclaw.js";
import { formatErrorMessage } from "../../../infra/errors.js";
@@ -474,17 +474,16 @@ async function persistSpawnedSessionLabel(params: {
if (!params.commandParams.storePath) {
return;
}
await updateSessionStore(params.commandParams.storePath, (store) => {
const existing = store[params.sessionKey];
if (!existing) {
return;
}
store[params.sessionKey] = {
...existing,
await updateSessionEntry(
{
storePath: params.commandParams.storePath,
sessionKey: params.sessionKey,
},
() => ({
label,
updatedAt: now,
};
});
}),
);
}
export async function handleAcpSpawnAction(
src/auto-reply/reply/commands-reset.ts
+8 -6
View File
@@ -2,7 +2,7 @@
import { clearBootstrapSnapshot } from "../../agents/bootstrap-cache.js";
import { clearAllCliSessions } from "../../agents/cli-session.js";
import { resetConfiguredBindingTargetInPlace } from "../../channels/plugins/binding-targets.js";
import { updateSessionStoreEntry } from "../../config/sessions/store.js";
import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
import { logVerbose } from "../../globals.js";
import { isAcpSessionKey } from "../../routing/session-key.js";
import { resolveBoundAcpThreadSessionKey } from "./commands-acp/targets.js";
@@ -81,10 +81,12 @@ export async function maybeHandleResetCommand(
params.sessionStore[params.sessionKey] = targetSessionEntry;
}
if (params.storePath && params.sessionKey) {
await updateSessionStoreEntry({
storePath: params.storePath,
sessionKey: params.sessionKey,
update: async (entry) => {
await updateSessionEntry(
{
storePath: params.storePath,
sessionKey: params.sessionKey,
},
async (entry) => {
const next = { ...entry };
clearAllCliSessions(next);
return {
@@ -95,7 +97,7 @@ export async function maybeHandleResetCommand(
lastInteractionAt: now,
};
},
});
);
}
}
src/auto-reply/reply/commands-subagents/action-info.ts
+8 -3
View File
@@ -4,7 +4,7 @@ import { subagentRuns } from "../../../agents/subagent-registry-memory.js";
import { countPendingDescendantRunsFromRuns } from "../../../agents/subagent-registry-queries.js";
import { getSubagentRunsSnapshotForRead } from "../../../agents/subagent-registry-state.js";
import { resolveStorePath } from "../../../config/sessions/paths.js";
import { loadSessionStore } from "../../../config/sessions/store-load.js";
import { loadSessionEntry } from "../../../config/sessions/session-accessor.js";
import { formatTimeAgo } from "../../../infra/format-time/format-relative.ts";
import { parseAgentSessionKey } from "../../../routing/session-key.js";
import { formatDurationCompact } from "../../../shared/subagents-format.js";
@@ -47,8 +47,13 @@ function loadSubagentSessionEntry(params: SubagentsCommandContext["params"], chi
const storePath = resolveStorePath(params.cfg.session?.store, {
agentId: parsed?.agentId,
});
const store = loadSessionStore(storePath);
return { entry: store[childKey] };
return {
entry: loadSessionEntry({
storePath,
sessionKey: childKey,
clone: false,
}),
};
}
export function handleSubagentsInfoAction(ctx: SubagentsCommandContext): CommandHandlerResult {
src/auto-reply/reply/directive-handling.impl.ts
+2 -4
View File
@@ -5,7 +5,7 @@ import { renderExecTargetLabel } from "../../agents/bash-tools.exec-runtime.js";
import { resolveExecDefaults } from "../../agents/exec-defaults.js";
import { resolveFastModeState } from "../../agents/fast-mode.js";
import { resolveSandboxRuntimeStatus } from "../../agents/sandbox.js";
import { updateSessionStore } from "../../config/sessions.js";
import { replaceSessionEntry } from "../../config/sessions/session-accessor.js";
import { triggerSessionPatchHook } from "../../gateway/session-patch-hooks.js";
import { enqueueSystemEvent } from "../../infra/system-events.js";
import { applyTraceOverride, applyVerboseOverride } from "../../sessions/level-overrides.js";
@@ -476,9 +476,7 @@ export async function handleDirectiveOnly(
sessionEntry.updatedAt = Date.now();
sessionStore[sessionKey] = sessionEntry;
if (storePath) {
await updateSessionStore(storePath, (store) => {
store[sessionKey] = sessionEntry;
});
await replaceSessionEntry({ storePath, sessionKey }, sessionEntry);
}
if (modelSelection && modelSelectionUpdated && sessionKey) {
triggerSessionPatchHook({
src/auto-reply/reply/directive-handling.persist.ts
+2 -4
View File
@@ -13,7 +13,7 @@ import {
type ModelAliasIndex,
} from "../../agents/model-selection.js";
import { resolveContextConfigProviderForRuntime } from "../../agents/openai-routing.js";
import { updateSessionStore } from "../../config/sessions/store.js";
import { replaceSessionEntry } from "../../config/sessions/session-accessor.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { triggerSessionPatchHook } from "../../gateway/session-patch-hooks.js";
@@ -364,9 +364,7 @@ export async function persistInlineDirectives(params: {
sessionEntry.updatedAt = Date.now();
sessionStore[sessionKey] = sessionEntry;
if (storePath) {
await updateSessionStore(storePath, (store) => {
store[sessionKey] = sessionEntry;
});
await replaceSessionEntry({ storePath, sessionKey }, sessionEntry);
}
if (modelDirective && modelUpdated) {
triggerSessionPatchHook({
src/auto-reply/reply/dispatch-from-config.runtime.ts
+25 -1
View File
@@ -1,9 +1,33 @@
/** Runtime-only dispatch dependencies shared by config-driven reply delivery. */
import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
import type { SessionEntry } from "../../config/sessions/types.js";
export { resolveStorePath } from "../../config/sessions/paths.js";
export {
loadSessionStore,
readSessionEntry,
resolveSessionStoreEntry,
updateSessionStoreEntry,
} from "../../config/sessions/store.js";
export { createInternalHookEvent, triggerInternalHook } from "../../hooks/internal-hooks.js";
export async function updateSessionStoreEntry(params: {
storePath: string;
sessionKey: string;
skipMaintenance?: boolean;
takeCacheOwnership?: boolean;
update: (
entry: SessionEntry,
) => Promise<Partial<SessionEntry> | null> | Partial<SessionEntry> | null;
}): Promise<SessionEntry | null> {
return await updateSessionEntry(
{
storePath: params.storePath,
sessionKey: params.sessionKey,
},
params.update,
{
skipMaintenance: params.skipMaintenance,
takeCacheOwnership: params.takeCacheOwnership,
},
);
}
src/auto-reply/reply/followup-runner.ts
+34 -11
View File
@@ -25,8 +25,8 @@ import {
buildAgentRuntimeDeliveryPlan,
buildAgentRuntimeOutcomePlan,
} from "../../agents/runtime-plan/build.js";
import { updateSessionStore, type SessionEntry } from "../../config/sessions.js";
import { readSessionEntry } from "../../config/sessions/store-load.js";
import type { SessionEntry } from "../../config/sessions.js";
import { loadSessionEntry, updateSessionEntry } from "../../config/sessions/session-accessor.js";
import type { TypingMode } from "../../config/types.js";
import { logVerbose } from "../../globals.js";
import {
@@ -529,7 +529,10 @@ export function createFollowupRunner(params: {
const resolveCurrentVerboseLevel = () => {
if (replySessionKey && storePath) {
try {
const level = readSessionEntry(storePath, replySessionKey)?.verboseLevel;
const level = loadSessionEntry({
storePath,
sessionKey: replySessionKey,
})?.verboseLevel;
if (typeof level === "string" && level.trim()) {
return level;
}
@@ -605,7 +608,10 @@ export function createFollowupRunner(params: {
const admittedSessionEntry = replySessionKey
? (sessionStore?.[replySessionKey] ??
(storePath
? (readSessionEntry(storePath, replySessionKey) as SessionEntry | undefined)
? loadSessionEntry({
storePath,
sessionKey: replySessionKey,
})
: undefined))
: undefined;
if (admittedSessionEntry?.sessionId === replyOperation.sessionId) {
@@ -790,16 +796,33 @@ export function createFollowupRunner(params: {
if (!storePath) {
return;
}
await updateSessionStore(storePath, (store) => {
const persistedEntry = store[replySessionKey];
if (!persistedEntry) {
return;
}
await updateSessionEntry({ storePath, sessionKey: replySessionKey }, (persistedEntry) => {
if (!entryMatchesAutoFallbackPrimaryProbe(persistedEntry, probe)) {
return;
return null;
}
const shouldClearAuthProfile =
persistedEntry.authProfileOverrideSource === "auto" ||
(persistedEntry.authProfileOverrideSource === undefined &&
persistedEntry.authProfileOverrideCompactionCount !== undefined);
clearAutoFallbackPrimaryProbeSelection(persistedEntry);
store[replySessionKey] = persistedEntry;
return {
providerOverride: undefined,
modelOverride: undefined,
modelOverrideSource: undefined,
modelOverrideFallbackOriginProvider: undefined,
modelOverrideFallbackOriginModel: undefined,
...(shouldClearAuthProfile
? {
authProfileOverride: undefined,
authProfileOverrideSource: undefined,
authProfileOverrideCompactionCount: undefined,
}
: {}),
fallbackNoticeSelectedModel: undefined,
fallbackNoticeActiveModel: undefined,
fallbackNoticeReason: undefined,
updatedAt: persistedEntry.updatedAt,
};
});
};
fallbackProvider = run.provider;
src/auto-reply/reply/get-reply.ts
+9 -8
View File
@@ -529,13 +529,10 @@ export async function getReplyFromConfig(
sessionStore[sessionKey] = sessionEntry;
}
if (sessionKey && storePath) {
const { applySessionStoreEntryPatch } = await import("../../config/sessions.js");
await applySessionStoreEntryPatch({
storePath,
sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
patch: {
const { updateSessionEntry } = await import("../../config/sessions/session-accessor.js");
await updateSessionEntry(
{ storePath, sessionKey },
() => ({
pendingFinalDelivery: undefined,
pendingFinalDeliveryText: undefined,
pendingFinalDeliveryCreatedAt: undefined,
@@ -543,8 +540,12 @@ export async function getReplyFromConfig(
pendingFinalDeliveryAttemptCount: undefined,
pendingFinalDeliveryLastError: undefined,
pendingFinalDeliveryContext: undefined,
}),
{
skipMaintenance: true,
takeCacheOwnership: true,
},
});
);
}
}
}
src/auto-reply/reply/model-selection.ts
+6 -9
View File
@@ -91,8 +91,8 @@ function shouldLogModelSelectionTiming(): boolean {
const modelCatalogRuntimeLoader = createLazyImportLoader(
() => import("../../agents/model-catalog.runtime.js"),
);
const sessionStoreRuntimeLoader = createLazyImportLoader(
() => import("../../config/sessions/store.runtime.js"),
const sessionAccessorRuntimeLoader = createLazyImportLoader(
() => import("../../config/sessions/session-accessor.js"),
);
function normalizeRuntimeModelRef(provider: string, model: string) {
return normalizeModelRef(provider, model, RUNTIME_MODEL_VISIBILITY_NORMALIZATION);
@@ -102,8 +102,8 @@ function loadModelCatalogRuntime() {
return modelCatalogRuntimeLoader.load();
}
function loadSessionStoreRuntime() {
return sessionStoreRuntimeLoader.load();
function loadSessionAccessorRuntime() {
return sessionAccessorRuntimeLoader.load();
}
function findSelectedCatalogEntry(params: {
@@ -295,11 +295,8 @@ export async function createModelSelectionState(params: {
if (updated) {
sessionStore[sessionKey] = sessionEntry;
if (storePath) {
await (
await loadSessionStoreRuntime()
).updateSessionStore(storePath, (store) => {
store[sessionKey] = sessionEntry;
});
const { replaceSessionEntry } = await loadSessionAccessorRuntime();
await replaceSessionEntry({ storePath, sessionKey }, sessionEntry);
}
}
resetModelOverride = updated;
src/auto-reply/reply/queue/drain.ts
+6 -2
View File
@@ -3,7 +3,7 @@ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coe
import { runAgentHarnessBeforeMessageWriteHook } from "../../../agents/harness/hook-helpers.js";
import { normalizeChatType } from "../../../channels/chat-type.js";
import { resolveStorePath } from "../../../config/sessions.js";
import { readSessionEntry } from "../../../config/sessions/store-load.js";
import { loadSessionEntry } from "../../../config/sessions/session-accessor.js";
// Drains queued follow-up runs while preserving route and session identity.
import {
channelRouteCompactKey,
@@ -548,7 +548,11 @@ async function runSyntheticOverflowSummary(params: {
config: params.source.run.config,
};
}
const sessionEntry = readSessionEntry(storePath, sessionKey);
const sessionEntry = loadSessionEntry({
storePath,
sessionKey,
clone: false,
});
return {
sessionId: sessionEntry?.sessionId ?? params.source.run.sessionId,
sessionKey,
src/auto-reply/reply/session-reset-model.ts
+3 -5
View File
@@ -126,11 +126,9 @@ function applySelectionToSession(params: {
}
sessionStore[sessionKey] = sessionEntry;
if (storePath) {
void import("../../config/sessions.js")
.then(({ updateSessionStore }) =>
updateSessionStore(storePath, (store) => {
store[sessionKey] = sessionEntry;
}),
void import("../../config/sessions/session-accessor.js")
.then(({ replaceSessionEntry }) =>
replaceSessionEntry({ storePath, sessionKey }, sessionEntry),
)
.catch(() => {
// Ignore persistence errors; session still proceeds.
src/auto-reply/reply/session-updates.lifecycle.test.ts
+32 -1
View File
@@ -4,7 +4,7 @@ import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import type { OpenClawConfig } from "../../config/config.js";
import type { SessionEntry } from "../../config/sessions.js";
import { loadSessionStore, type SessionEntry } from "../../config/sessions.js";
import type { HookRunner } from "../../plugins/hooks.js";
const hookRunnerMocks = vi.hoisted(() => ({
@@ -108,4 +108,35 @@ describe("session-updates lifecycle hooks", () => {
expect(startContext?.sessionKey).toBe(sessionKey);
expect(startContext?.agentId).toBe("main");
});
it("recreates a complete persisted row when compaction updates a missing store row", async () => {
const { storePath, sessionKey, sessionStore, entry } = await createFixture();
await fs.writeFile(storePath, JSON.stringify({}, null, 2), "utf-8");
await incrementCompactionCount({
sessionEntry: entry,
sessionStore,
sessionKey,
storePath,
newSessionId: "s2",
tokensAfter: 123,
now: 456,
});
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
expect(sessionStore[sessionKey]?.sessionId).toBe("s2");
expect(sessionStore[sessionKey]?.sessionFile).toContain("s2.jsonl");
expect(sessionStore[sessionKey]?.usageFamilyKey).toBe(sessionKey);
expect(sessionStore[sessionKey]?.usageFamilySessionIds).toEqual(["s1", "s2"]);
expect(sessionStore[sessionKey]?.compactionCount).toBe(1);
expect(sessionStore[sessionKey]?.totalTokens).toBe(123);
expect(sessionStore[sessionKey]?.updatedAt).toBeGreaterThanOrEqual(entry.updatedAt);
expect(persisted?.sessionId).toBe("s2");
expect(persisted?.sessionFile).toContain("s2.jsonl");
expect(persisted?.usageFamilyKey).toBe(sessionKey);
expect(persisted?.usageFamilySessionIds).toEqual(["s1", "s2"]);
expect(persisted?.compactionCount).toBe(1);
expect(persisted?.totalTokens).toBe(123);
expect(persisted?.updatedAt).toBeGreaterThanOrEqual(entry.updatedAt);
});
});
src/auto-reply/reply/session-updates.ts
+12 -16
View File
@@ -10,8 +10,8 @@ import {
resolveSessionFilePathOptions,
rewriteSessionFileForNewSessionId,
type SessionEntry,
updateSessionStore,
} from "../../config/sessions.js";
import { patchSessionEntry, upsertSessionEntry } from "../../config/sessions/session-accessor.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
forgetActiveSessionForShutdown,
@@ -43,17 +43,12 @@ async function persistSessionEntryUpdate(params: {
if (!params.storePath) {
return;
}
await updateSessionStore(
params.storePath,
(store) => {
const next = { ...store[params.sessionKey!], ...params.nextEntry };
store[params.sessionKey!] = next;
return next;
},
await upsertSessionEntry(
{
resolveSingleEntryPersistence: (entry) =>
entry && params.sessionKey ? { sessionKey: params.sessionKey, entry } : null,
storePath: params.storePath,
sessionKey: params.sessionKey,
},
params.nextEntry,
);
}
@@ -315,17 +310,18 @@ export async function incrementCompactionCount(params: {
} else if (incrementBy > 0) {
updates.totalTokensFresh = false;
}
sessionStore[sessionKey] = {
const nextEntry = {
...entry,
...updates,
};
sessionStore[sessionKey] = nextEntry;
if (storePath) {
await updateSessionStore(storePath, (store) => {
store[sessionKey] = {
...store[sessionKey],
...updates,
};
const persistedEntry = await patchSessionEntry({ storePath, sessionKey }, () => updates, {
fallbackEntry: nextEntry,
});
if (persistedEntry) {
sessionStore[sessionKey] = persistedEntry;
}
}
if ((sessionIdChanged || sessionFileChanged) && cfg) {
emitCompactionSessionLifecycleHooks({
src/auto-reply/reply/session-usage.ts
+23 -15
View File
@@ -14,8 +14,8 @@ import {
resolveSessionGoalDisplayState,
type SessionSystemPromptReport,
type SessionEntry,
updateSessionStoreEntry,
} from "../../config/sessions.js";
import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { logVerbose } from "../../globals.js";
import { estimateUsageCost, resolveModelCostConfig } from "../../utils/usage-format.js";
@@ -141,12 +141,12 @@ export async function persistSessionUsageUpdate(params: {
if (hasUsage || hasFreshContextSnapshot || hasCompactionSnapshot) {
try {
await updateSessionStoreEntry({
storePath,
sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
update: async (entry) => {
await updateSessionEntry(
{
storePath,
sessionKey,
},
async (entry) => {
const updatedAt = Date.now();
const preserveSessionModelState =
params.isHeartbeat === true ||
@@ -241,7 +241,11 @@ export async function persistSessionUsageUpdate(params: {
? patch
: applyCliSessionIdToSessionPatch(params, entry, patch);
},
});
{
skipMaintenance: true,
takeCacheOwnership: true,
},
);
} catch (err) {
logVerbose(`failed to persist ${label}usage update: ${String(err)}`);
}
@@ -250,12 +254,12 @@ export async function persistSessionUsageUpdate(params: {
if (params.modelUsed || params.contextTokensUsed) {
try {
await updateSessionStoreEntry({
storePath,
sessionKey,
skipMaintenance: true,
takeCacheOwnership: true,
update: async (entry) => {
await updateSessionEntry(
{
storePath,
sessionKey,
},
async (entry) => {
const preserveSessionModelState =
params.isHeartbeat === true ||
params.preserveRuntimeModel === true ||
@@ -279,7 +283,11 @@ export async function persistSessionUsageUpdate(params: {
? patch
: applyCliSessionIdToSessionPatch(params, entry, patch);
},
});
{
skipMaintenance: true,
takeCacheOwnership: true,
},
);
} catch (err) {
logVerbose(`failed to persist ${label}model/context update: ${String(err)}`);
}
src/config/sessions/session-accessor.ts
+4
View File
@@ -12,6 +12,7 @@ import {
resolveStorePath,
} from "./paths.js";
import { resolveAndPersistSessionFile } from "./session-file.js";
import type { ResolvedSessionMaintenanceConfig } from "./store-maintenance.js";
import {
getSessionEntry,
cleanupSessionLifecycleArtifacts as cleanupFileSessionLifecycleArtifacts,
@@ -157,6 +158,8 @@ export type SessionEntryUpdateOptions = {
export type SessionEntryPatchOptions = {
/** Entry to synthesize when a patch operation is allowed to create. */
fallbackEntry?: SessionEntry;
/** Fully resolved maintenance settings when the caller already has config loaded. */
maintenanceConfig?: ResolvedSessionMaintenanceConfig;
/** Keep the previous updatedAt value when the patch should not count as activity. */
preserveActivity?: boolean;
/** Replace the whole entry instead of merging the returned patch. */
@@ -267,6 +270,7 @@ export async function patchSessionEntry(
return await patchFileSessionEntry({
...scope,
fallbackEntry: options.fallbackEntry,
maintenanceConfig: options.maintenanceConfig,
preserveActivity: options.preserveActivity,
replaceEntry: options.replaceEntry,
update,
src/config/sessions/store-maintenance.ts
+51 -10
View File
@@ -213,6 +213,46 @@ export interface QuotaSuspensionMaintenanceResult {
cleared: number;
}
export type QuotaSuspensionEntryMaintenanceResult = {
/** Patch to apply to the entry, or null when no TTL transition is due. */
patch: Partial<SessionEntry> | null;
/** Present when the entry transitioned from suspended to resuming. */
resumed?: { laneId?: string };
/** True when the quota-suspension marker should be removed. */
cleared: boolean;
};
/**
* Resolves the TTL maintenance patch for one session entry without reading or
* mutating the whole store. Attempt hot paths use this before entry-scoped
* accessor writes so unrelated sessions stay out of the request path.
*/
export function resolveQuotaSuspensionEntryMaintenance(params: {
entry: SessionEntry;
now: number;
ttlMs?: number;
}): QuotaSuspensionEntryMaintenanceResult {
const suspension = params.entry.quotaSuspension;
if (!suspension) {
return { patch: null, cleared: false };
}
const ttlMs = params.ttlMs ?? DEFAULT_QUOTA_SUSPENSION_TTL_MS;
const cleanupAfterResumeMs = ttlMs * (QUOTA_SUSPENSION_CLEANUP_FACTOR - 1);
const resumeAtMs = suspension.expectedResumeBy ?? suspension.suspendedAt + ttlMs;
const cleanupAtMs = resumeAtMs + cleanupAfterResumeMs;
if (params.now >= cleanupAtMs) {
return { patch: { quotaSuspension: undefined }, cleared: true };
}
if (suspension.state === "suspended" && params.now >= resumeAtMs) {
return {
patch: { quotaSuspension: { ...suspension, state: "resuming" } },
resumed: { laneId: suspension.laneId },
cleared: false,
};
}
return { patch: null, cleared: false };
}
/**
* Two-stage TTL maintenance for `quotaSuspension` records:
* 1. After `ttlMs`, transition `state: "suspended" → "resuming"` so the next
@@ -230,24 +270,25 @@ export function pruneQuotaSuspensions(params: {
log?: boolean;
}): QuotaSuspensionMaintenanceResult {
const ttlMs = params.ttlMs ?? DEFAULT_QUOTA_SUSPENSION_TTL_MS;
const cleanupAfterResumeMs = ttlMs * (QUOTA_SUSPENSION_CLEANUP_FACTOR - 1);
const resumed: Array<{ sessionKey: string; laneId?: string }> = [];
let cleared = 0;
for (const [sessionKey, entry] of Object.entries(params.store)) {
const suspension = entry.quotaSuspension;
if (!suspension) {
const result = resolveQuotaSuspensionEntryMaintenance({
entry,
now: params.now,
ttlMs,
});
if (!result.patch) {
continue;
}
const resumeAtMs = suspension.expectedResumeBy ?? suspension.suspendedAt + ttlMs;
const cleanupAtMs = resumeAtMs + cleanupAfterResumeMs;
if (params.now >= cleanupAtMs) {
if (result.cleared) {
delete entry.quotaSuspension;
cleared++;
continue;
} else if (result.patch.quotaSuspension) {
entry.quotaSuspension = result.patch.quotaSuspension;
}
if (suspension.state === "suspended" && params.now >= resumeAtMs) {
entry.quotaSuspension = { ...suspension, state: "resuming" };
resumed.push({ sessionKey, laneId: suspension.laneId });
if (result.resumed) {
resumed.push({ sessionKey, laneId: result.resumed.laneId });
}
}
if ((resumed.length > 0 || cleared > 0) && params.log !== false) {
src/config/sessions/store.pruning.test.ts
+67
View File
@@ -9,6 +9,7 @@ import {
import {
isProtectedSessionMaintenanceEntry,
resolveMaintenanceConfigFromInput,
resolveQuotaSuspensionEntryMaintenance,
resolveSessionEntryMaintenanceHighWater,
} from "./store-maintenance.js";
import { capEntryCount, getActiveSessionMaintenanceWarning, pruneStaleEntries } from "./store.js";
@@ -77,6 +78,72 @@ describe("pruneStaleEntries", () => {
});
});
describe("resolveQuotaSuspensionEntryMaintenance", () => {
it("returns an entry-scoped patch when a suspended session should resume", () => {
const now = Date.now();
const result = resolveQuotaSuspensionEntryMaintenance({
entry: {
...makeEntry(now),
quotaSuspension: {
schemaVersion: 1,
suspendedAt: now - 30_000,
expectedResumeBy: now - 1,
state: "suspended",
reason: "quota_exhausted",
failedProvider: "anthropic",
failedModel: "claude-opus-4-6",
laneId: "main",
},
},
now,
ttlMs: 30_000,
});
expect(result).toEqual({
patch: {
quotaSuspension: {
schemaVersion: 1,
suspendedAt: now - 30_000,
expectedResumeBy: now - 1,
state: "resuming",
reason: "quota_exhausted",
failedProvider: "anthropic",
failedModel: "claude-opus-4-6",
laneId: "main",
},
},
resumed: { laneId: "main" },
cleared: false,
});
});
it("returns an entry-scoped cleanup patch after the resume window expires", () => {
const now = Date.now();
const result = resolveQuotaSuspensionEntryMaintenance({
entry: {
...makeEntry(now),
quotaSuspension: {
schemaVersion: 1,
suspendedAt: now - 61_000,
expectedResumeBy: now - 31_000,
state: "active",
reason: "circuit_open",
failedProvider: "anthropic",
failedModel: "claude-opus-4-6",
laneId: "main",
},
},
now,
ttlMs: 30_000,
});
expect(result).toEqual({
patch: { quotaSuspension: undefined },
cleared: true,
});
});
});
describe("capEntryCount", () => {
it("over limit: keeps N most recent by updatedAt, deletes rest", () => {
const now = Date.now();
src/config/sessions/store.ts
+4
View File
@@ -196,6 +196,7 @@ type SessionEntryWorkflowOptions = {
agentId?: string;
env?: NodeJS.ProcessEnv;
hydrateSkillPromptRefs?: boolean;
maintenanceConfig?: ResolvedSessionMaintenanceConfig;
storePath?: string;
};
@@ -1184,6 +1185,7 @@ async function persistResolvedSessionEntry(params: {
resolved: ReturnType<typeof resolveSessionStoreEntry>;
next: SessionEntry;
skipMaintenance?: boolean;
maintenanceConfig?: ResolvedSessionMaintenanceConfig;
takeCacheOwnership?: boolean;
returnDetached?: boolean;
requireWriteSuccess?: boolean;
@@ -1199,6 +1201,7 @@ async function persistResolvedSessionEntry(params: {
await saveSessionStoreUnlocked(params.storePath, params.store, {
activeSessionKey: params.resolved.normalizedKey,
skipMaintenance: params.skipMaintenance,
maintenanceConfig: params.maintenanceConfig,
skipSerializeForUnchangedStore: entryUnchanged,
singleEntryPersistence:
params.resolved.legacyKeys.length === 0 && params.resolved.existing
@@ -1310,6 +1313,7 @@ export async function patchSessionEntry(
store,
resolved,
next,
maintenanceConfig: params.maintenanceConfig,
takeCacheOwnership: true,
returnDetached: true,
});
test/scripts/check-session-accessor-boundary.test.ts
+66
View File
@@ -2,7 +2,9 @@ import { describe, expect, it } from "vitest";
import {
findSessionAccessorBoundaryViolations,
migratedBundledPluginSessionAccessorFiles,
findSessionAccessorWriteBoundaryViolations,
migratedSessionAccessorFiles,
migratedSessionAccessorWriteFiles,
} from "../../scripts/check-session-accessor-boundary.mjs";
describe("session accessor boundary guard", () => {
@@ -10,9 +12,15 @@ describe("session accessor boundary guard", () => {
expect(migratedSessionAccessorFiles).toEqual(
new Set([
"src/agents/embedded-agent-runner/compaction-successor-transcript.ts",
"src/agents/embedded-agent-runner/run/attempt.ts",
"src/agents/embedded-agent-runner/tool-result-truncation.ts",
"src/agents/embedded-agent-runner/transcript-rewrite.ts",
"src/agents/embedded-agent-runner/transcript-runtime-state.ts",
"src/auto-reply/reply/agent-runner-helpers.ts",
"src/auto-reply/reply/agent-runner.ts",
"src/auto-reply/reply/commands-subagents/action-info.ts",
"src/auto-reply/reply/followup-runner.ts",
"src/auto-reply/reply/queue/drain.ts",
"src/commands/export-trajectory.ts",
"src/commands/health.ts",
"src/commands/sandbox-explain.ts",
@@ -42,6 +50,34 @@ describe("session accessor boundary guard", () => {
);
});
it("ratchets only the auto-reply files migrated to session accessor writes", () => {
expect(migratedSessionAccessorWriteFiles).toEqual(
new Set([
"src/agents/command/attempt-execution.shared.ts",
"src/agents/command/session-store.ts",
"src/agents/embedded-agent-runner/run.ts",
"src/agents/embedded-agent-runner/run/attempt.ts",
"src/auto-reply/reply/abort-cutoff.runtime.ts",
"src/auto-reply/reply/agent-runner-cli-dispatch.ts",
"src/auto-reply/reply/agent-runner-execution.ts",
"src/auto-reply/reply/agent-runner-memory.ts",
"src/auto-reply/reply/agent-runner.ts",
"src/auto-reply/reply/body.ts",
"src/auto-reply/reply/commands-acp/lifecycle.ts",
"src/auto-reply/reply/commands-reset.ts",
"src/auto-reply/reply/directive-handling.impl.ts",
"src/auto-reply/reply/directive-handling.persist.ts",
"src/auto-reply/reply/dispatch-from-config.runtime.ts",
"src/auto-reply/reply/followup-runner.ts",
"src/auto-reply/reply/get-reply.ts",
"src/auto-reply/reply/model-selection.ts",
"src/auto-reply/reply/session-reset-model.ts",
"src/auto-reply/reply/session-updates.ts",
"src/auto-reply/reply/session-usage.ts",
]),
);
});
it("flags legacy reader imports", () => {
expect(
findSessionAccessorBoundaryViolations(`
@@ -112,6 +148,36 @@ describe("session accessor boundary guard", () => {
).toEqual([]);
});
it("flags legacy writer imports and calls", () => {
expect(
findSessionAccessorWriteBoundaryViolations(`
import { applySessionStoreEntryPatch, saveSessionStore, updateSessionStore, updateSessionStoreEntry as updateEntry } from "../config/sessions.js";
saveSessionStore(storePath, store);
updateSessionStore(storePath, () => undefined);
sessions.updateSessionStoreEntry({ storePath, sessionKey, update });
applySessionStoreEntryPatch({ storePath, sessionKey, patch });
`),
).toEqual([
{ line: 2, reason: 'imports legacy session store writer "applySessionStoreEntryPatch"' },
{ line: 2, reason: 'imports legacy session store writer "saveSessionStore"' },
{ line: 2, reason: 'imports legacy session store writer "updateSessionStore"' },
{ line: 2, reason: 'imports legacy session store writer "updateSessionStoreEntry"' },
{ line: 3, reason: 'calls legacy session store writer "saveSessionStore"' },
{ line: 4, reason: 'calls legacy session store writer "updateSessionStore"' },
{ line: 5, reason: 'references legacy session store writer "updateSessionStoreEntry"' },
{ line: 6, reason: 'calls legacy session store writer "applySessionStoreEntryPatch"' },
]);
});
it("allows migrated accessor writes", () => {
expect(
findSessionAccessorWriteBoundaryViolations(`
import { updateSessionEntry } from "../config/sessions/session-accessor.js";
updateSessionEntry({ storePath, sessionKey }, () => undefined);
`),
).toEqual([]);
});
it("ignores comments and strings that describe legacy readers", () => {
expect(
findSessionAccessorBoundaryViolations(`