vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-09 15:59:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 22

refactor(sessions): move inbound meta, goals, and delivery reads behind the session accessor (#101180)

Browse source
This commit is contained in:
Josh Lehman 2026-07-06 18:07:26 -07:00 • committed by GitHub
parent bb658d0a6d
commit cb0d8a1294
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 377 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

4
docs/.generated/plugin-sdk-api-baseline.sha256
View file

@ -1,2 +1,2 @@
1805b1ddbd9036dac247b4eb01144c791e54e3a4826298bca4f85f67e936d8e2 plugin-sdk-api-baseline.json
e1a4138ec4962ecb09472bf97cabebb98477d82af75f701142323706dcaa95f2 plugin-sdk-api-baseline.jsonl
84f32cec516711bd2b96bcb0bf07f086ee63867a21b4d00de3fcca3def17070d plugin-sdk-api-baseline.json
bd4505b690bbe4cad38712d2c6de15cfde905c3ed76bafad86b41f2028292f2b plugin-sdk-api-baseline.jsonl

10
scripts/check-session-accessor-boundary.mjs
View file

@ -17,6 +17,7 @@ const legacyReaderNames = new Set([
"readSessionEntries",
"readSessionEntry",
"readSessionStoreReadOnly",
"readSessionStoreSnapshot",
"resolveSessionStoreEntry",
]);
const legacyWholeStoreAccessNames = new Set([
@ -26,7 +27,9 @@ const legacyWholeStoreAccessNames = new Set([
]);
const legacyWriterNames = new Set([
"applySessionStoreEntryPatch",
"recordSessionMetaFromInbound",
"saveSessionStore",
"updateLastRoute",
"updateSessionStore",
"updateSessionStoreEntry",
]);
@ -101,6 +104,8 @@ export const migratedSessionAccessorFiles = new Set([
"src/commands/status.summary.ts",
"src/commands/tasks.ts",
"src/config/sessions/combined-store-gateway.ts",
"src/config/sessions/delivery-info.ts",
"src/config/sessions/goals.ts",
"src/cron/isolated-agent/delivery-target.ts",
"src/cron/service/timer.ts",
"src/gateway/session-compaction-checkpoints.ts",
@ -156,6 +161,7 @@ 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-subscribe.handlers.compaction.runtime.ts",
"src/agents/embedded-agent-runner/run/attempt.ts",
"src/agents/live-model-switch.ts",
"src/agents/main-session-restart-recovery.ts",
@ -185,12 +191,15 @@ export const migratedSessionAccessorWriteFiles = new Set([
"src/auto-reply/reply/session-usage.ts",
"src/commands/tasks.ts",
"src/config/sessions/cleanup-service.ts",
"src/config/sessions/goals.ts",
"src/gateway/boot.ts",
"src/gateway/server-methods/sessions.ts",
"src/gateway/server-node-events.ts",
"src/gateway/session-compaction-checkpoints.ts",
"src/infra/outbound/outbound-session.ts",
"src/plugins/host-hook-cleanup.ts",
"src/plugins/host-hook-state.ts",
"src/plugins/runtime/runtime-channel.ts",
"src/tui/embedded-backend.ts",
]);
@ -565,6 +574,7 @@ const writeSourceRootPaths = [
"src/commands",
"src/config/sessions",
"src/gateway",
"src/infra",
"src/plugins",
"src/tui",
];

7
scripts/lib/session-accessor-debt-baseline.json
View file

@ -33,7 +33,7 @@
"src/config/sessions/cleanup-service.ts": 4,
"src/config/sessions/session-accessor.ts": 25,
"src/config/sessions/session-registry-maintenance.ts": 2,
"src/config/sessions/store-load.ts": 4,
"src/config/sessions/store-load.ts": 5,
"src/config/sessions/store.ts": 15,
"src/config/sessions/test-helpers.ts": 2,
"src/config/sessions/transcript.ts": 6,
@ -43,7 +43,6 @@
"src/plugins/runtime/runtime-agent.ts": 1
},
"sessionAccessorWrite": {
"src/agents/embedded-agent-subscribe.handlers.compaction.runtime.ts": 2,
"src/agents/session-suspension.ts": 2,
"src/agents/subagent-orphan-recovery.ts": 3,
"src/agents/subagent-spawn.ts": 2,
@ -55,13 +54,15 @@
"src/commands/doctor-state-integrity.ts": 2,
"src/commands/doctor/shared/codex-route-warnings.ts": 2,
"src/config/sessions/plugin-host-cleanup.ts": 2,
"src/config/sessions/session-accessor.ts": 11,
"src/config/sessions/session-accessor.ts": 13,
"src/config/sessions/session-file.ts": 2,
"src/config/sessions/session-registry-maintenance.ts": 2,
"src/gateway/server-methods/agent.ts": 2,
"src/gateway/server-methods/chat.ts": 2,
"src/gateway/session-lifecycle-state.ts": 2,
"src/gateway/test-helpers.mocks.ts": 1,
"src/infra/heartbeat-runner.ts": 5,
"src/infra/state-migrations.ts": 2,
"src/plugins/runtime/runtime-agent.ts": 2
},
"sessionCompactManualTrim": {},

31
src/agents/embedded-agent-subscribe.handlers.compaction.runtime.ts
View file

@ -1,7 +1,8 @@
/**
* Runtime helpers for reconciling compaction counts after subscribe events.
*/
import { resolveStorePath, updateSessionStoreEntry } from "../config/sessions.js";
import { resolveStorePath } from "../config/sessions/paths.js";
import { updateSessionEntry } from "../config/sessions/session-accessor.js";
/** Persist the highest observed compaction count after a successful subscribed run. */
export default async function reconcileSessionStoreCompactionCountAfterSuccess(params: {
@ -16,22 +17,18 @@ export default async function reconcileSessionStoreCompactionCountAfterSuccess(p
return undefined;
}
const storePath = resolveStorePath(configStore, { agentId });
const nextEntry = await updateSessionStoreEntry({
storePath,
sessionKey,
update: async (entry) => {
// The live stream and store can both observe compactions. Keep the max so
// late lower-count updates cannot make future resume labels regress.
const currentCount = Math.max(0, entry.compactionCount ?? 0);
const nextCount = Math.max(currentCount, observedCompactionCount);
if (nextCount === currentCount) {
return null;
}
return {
compactionCount: nextCount,
updatedAt: Math.max(entry.updatedAt ?? 0, now),
};
},
const nextEntry = await updateSessionEntry({ sessionKey, storePath }, async (entry) => {
// The live stream and store can both observe compactions. Keep the max so
// late lower-count updates cannot make future resume labels regress.
const currentCount = Math.max(0, entry.compactionCount ?? 0);
const nextCount = Math.max(currentCount, observedCompactionCount);
if (nextCount === currentCount) {
return null;
}
return {
compactionCount: nextCount,
updatedAt: Math.max(entry.updatedAt ?? 0, now),
};
});
return nextEntry?.compactionCount;
}

2
src/channels/session-meta.ts
View file

@ -23,7 +23,7 @@ export async function recordInboundSessionMetaSafe(params: {
agentId: params.agentId,
});
try {
await runtime.recordSessionMetaFromInbound({
await runtime.recordInboundSessionMeta({
storePath,
sessionKey: params.sessionKey,
ctx: params.ctx,

4
src/channels/session.test.ts
View file

@ -6,8 +6,8 @@ const recordSessionMetaFromInboundMock = vi.fn((_args?: unknown) => Promise.reso
const updateLastRouteMock = vi.fn((_args?: unknown) => Promise.resolve(undefined));
vi.mock("../config/sessions/inbound.runtime.js", () => ({
recordSessionMetaFromInbound: (args: unknown) => recordSessionMetaFromInboundMock(args),
updateLastRoute: (args: unknown) => updateLastRouteMock(args),
recordInboundSessionMeta: (args: unknown) => recordSessionMetaFromInboundMock(args),
updateSessionLastRoute: (args: unknown) => updateLastRouteMock(args),
}));
type SessionModule = typeof import("./session.js");

4
src/channels/session.ts
View file

@ -43,7 +43,7 @@ export async function recordInboundSession(params: {
const canonicalSessionKey = normalizeSessionKeyPreservingOpaquePeerIds(sessionKey);
const runtime = await loadInboundSessionRuntime();
const metaTask = runtime
.recordSessionMetaFromInbound({
.recordInboundSessionMeta({
storePath,
sessionKey: canonicalSessionKey,
ctx,
@ -62,7 +62,7 @@ export async function recordInboundSession(params: {
return;
}
const targetSessionKey = normalizeSessionKeyPreservingOpaquePeerIds(update.sessionKey);
await runtime.updateLastRoute({
await runtime.updateSessionLastRoute({
storePath,
sessionKey: targetSessionKey,
route: update.route,

69
src/config/sessions/delivery-info.test.ts
View file

@ -9,8 +9,15 @@ const storeState = vi.hoisted(() => {
const state = {
store: {} as Record<string, SessionEntry>,
stores: {} as Record<string, Record<string, SessionEntry>>,
loadSessionStore: vi.fn((storePath: string) => state.stores[storePath] ?? state.store),
readSessionStoreSnapshot: vi.fn((storePath: string) => state.stores[storePath] ?? state.store),
// Mirrors the accessor view contract: raw exact-key get, enumeration only via entries().
openSessionEntryReadView: vi.fn((scope: { storePath?: string }) => {
const store = state.stores[scope.storePath ?? ""] ?? state.store;
return {
get: (sessionKey: string) =>
Object.hasOwn(store, sessionKey) ? store[sessionKey] : undefined,
entries: () => Object.entries(store).map(([sessionKey, entry]) => ({ sessionKey, entry })),
};
}),
};
return state;
});
@ -24,9 +31,8 @@ vi.mock("./paths.js", () => ({
opts?.agentId === "worker" ? "/tmp/worker-sessions.json" : "/tmp/sessions.json",
}));
vi.mock("./store.js", () => ({
loadSessionStore: storeState.loadSessionStore,
readSessionStoreSnapshot: storeState.readSessionStoreSnapshot,
vi.mock("./session-accessor.js", () => ({
openSessionEntryReadView: storeState.openSessionEntryReadView,
}));
vi.mock("./targets.js", () => ({
@ -53,8 +59,7 @@ beforeEach(() => {
setActivePluginRegistry(createSessionConversationTestRegistry());
storeState.store = {};
storeState.stores = {};
storeState.loadSessionStore.mockClear();
storeState.readSessionStoreSnapshot.mockClear();
storeState.openSessionEntryReadView.mockClear();
});
describe("extractDeliveryInfo", () => {
@ -94,7 +99,7 @@ describe("extractDeliveryInfo", () => {
});
});
it("uses session-store snapshots for direct session keys", () => {
it("reads borrowed accessor views for direct session keys", () => {
const sessionKey = "agent:main:webchat:dm:user-123";
storeState.store[sessionKey] = buildEntry({
channel: "webchat",
@ -105,32 +110,16 @@ describe("extractDeliveryInfo", () => {
const result = extractDeliveryInfo(sessionKey);
expect(result.deliveryContext?.to).toBe("webchat:user-123");
expect(storeState.readSessionStoreSnapshot).toHaveBeenCalledWith("/tmp/sessions.json");
expect(storeState.loadSessionStore).not.toHaveBeenCalled();
});
it("returns deliveryContext for direct session keys", () => {
const sessionKey = "agent:main:webchat:dm:user-123";
storeState.store[sessionKey] = buildEntry({
channel: "webchat",
to: "webchat:user-123",
accountId: "default",
});
const result = extractDeliveryInfo(sessionKey);
expect(result).toEqual({
deliveryContext: {
channel: "webchat",
to: "webchat:user-123",
accountId: "default",
},
threadId: undefined,
expect(storeState.openSessionEntryReadView).toHaveBeenCalledWith({
storePath: "/tmp/sessions.json",
});
});
it("does not build the normalized index when an exact routable key is present", () => {
it("does not enumerate the store when an exact routable key is present", () => {
const sessionKey = "agent:main:webchat:dm:user-123";
// Enumeration trap: the accessor-view mock lists rows via Object.entries, so
// building the normalized fallback index on this cheap path throws here and
// extractDeliveryInfo would return no delivery context.
storeState.store = new Proxy(
{
[sessionKey]: buildEntry({
@ -158,6 +147,26 @@ describe("extractDeliveryInfo", () => {
});
});
it("returns deliveryContext for direct session keys", () => {
const sessionKey = "agent:main:webchat:dm:user-123";
storeState.store[sessionKey] = buildEntry({
channel: "webchat",
to: "webchat:user-123",
accountId: "default",
});
const result = extractDeliveryInfo(sessionKey);
expect(result).toEqual({
deliveryContext: {
channel: "webchat",
to: "webchat:user-123",
accountId: "default",
},
threadId: undefined,
});
});
it("falls back to base sessions for :thread: keys", () => {
const baseKey = "agent:main:slack:channel:C0123ABC";
const threadKey = `${baseKey}:thread:1234567890.123456`;

58
src/config/sessions/delivery-info.ts
View file

@ -9,13 +9,13 @@ import { deliveryContextFromSession } from "../../utils/delivery-context.shared.
import { getRuntimeConfig } from "../io.js";
import type { OpenClawConfig } from "../types.openclaw.js";
import { resolveStorePath } from "./paths.js";
import { openSessionEntryReadView, type SessionEntryReadView } from "./session-accessor.js";
import {
foldedSessionKeyAliasCandidates,
hasMismatchedCaseSensitiveDeliveryProof,
isConfirmedLowercasedLegacyAlias,
normalizeStoreSessionKey,
} from "./store-entry.js";
import { readSessionStoreSnapshot } from "./store.js";
import { resolveAllAgentSessionStoreTargetsSync } from "./targets.js";
import { parseSessionThreadInfo } from "./thread-info.js";
import type { SessionEntry } from "./types.js";
@ -93,14 +93,7 @@ function resolveDeliveryStorePaths(cfg: OpenClawConfig, agentId: string): string
return [...paths];
}
function asSessionEntry(entry: unknown): SessionEntry | undefined {
return entry as SessionEntry | undefined;
}
function findSessionEntryInStore(
store: ReturnType<typeof readSessionStoreSnapshot>,
keys: readonly string[],
) {
function findSessionEntryInStore(store: SessionEntryReadView, keys: readonly string[]) {
let normalizedIndex: Map<string, SessionEntry> | undefined;
let bestEntry: SessionEntry | undefined;
let bestUpdatedAt = 0;
@ -109,11 +102,10 @@ function findSessionEntryInStore(
// Preference order: routable delivery context first; then Matrix/tail-preserved
// exact keys over folded aliases; then freshness. Ordinary lowercase-canonical
// channels keep the previous freshest-routable alias behavior.
const acceptCandidate = (candidate: unknown, isExact = false) => {
if (!candidate) {
const acceptCandidate = (entry: SessionEntry | undefined, isExact = false) => {
if (!entry) {
return;
}
const entry = candidate as SessionEntry;
const candidateRoutable = hasRoutableDeliveryContext(deliveryContextFromSession(entry));
const candidateUpdatedAt = entry.updatedAt ?? 0;
if (
@ -136,37 +128,28 @@ function findSessionEntryInStore(
const foldedLegacyKeys = foldedSessionKeyAliasCandidates(normalized);
const exactKeyWins = requiresFoldedSessionKeyAliasProof(normalized);
let foundRoutableCandidate = false;
if (
Object.hasOwn(store, normalized) &&
!hasMismatchedCaseSensitiveDeliveryProof(asSessionEntry(store[normalized]), normalized)
) {
foundRoutableCandidate ||= hasRoutableDeliveryContext(
deliveryContextFromSession(asSessionEntry(store[normalized])),
);
acceptCandidate(store[normalized], exactKeyWins);
// Exact and alias probes are raw keyed reads; the store is never enumerated here.
const exactEntry = store.get(normalized);
if (exactEntry && !hasMismatchedCaseSensitiveDeliveryProof(exactEntry, normalized)) {
foundRoutableCandidate ||= hasRoutableDeliveryContext(deliveryContextFromSession(exactEntry));
acceptCandidate(exactEntry, exactKeyWins);
}
for (const foldedLegacyKey of foldedLegacyKeys) {
if (
!Object.hasOwn(store, foldedLegacyKey) ||
!isConfirmedLowercasedLegacyAlias(asSessionEntry(store[foldedLegacyKey]), normalized)
) {
const foldedLegacyEntry = store.get(foldedLegacyKey);
if (!foldedLegacyEntry || !isConfirmedLowercasedLegacyAlias(foldedLegacyEntry, normalized)) {
continue;
}
const foldedLegacyEntry = asSessionEntry(store[foldedLegacyKey]);
foundRoutableCandidate ||= hasRoutableDeliveryContext(
deliveryContextFromSession(foldedLegacyEntry),
);
acceptCandidate(foldedLegacyEntry);
}
if (
trimmed !== normalized &&
Object.hasOwn(store, trimmed) &&
!hasMismatchedCaseSensitiveDeliveryProof(asSessionEntry(store[trimmed]), normalized)
) {
const trimmedEntry = trimmed !== normalized ? store.get(trimmed) : undefined;
if (trimmedEntry && !hasMismatchedCaseSensitiveDeliveryProof(trimmedEntry, normalized)) {
foundRoutableCandidate ||= hasRoutableDeliveryContext(
deliveryContextFromSession(asSessionEntry(store[trimmed])),
deliveryContextFromSession(trimmedEntry),
);
acceptCandidate(store[trimmed]);
acceptCandidate(trimmedEntry);
}
if (trimmed !== normalized || !foundRoutableCandidate) {
// Build the normalized index only after direct/exact probes fail; large session stores can
@ -187,12 +170,9 @@ function findSessionEntryInStore(
return bestEntry;
}
function buildFreshestSessionEntryIndex(
store: Readonly<Record<string, unknown>>,
): Map<string, SessionEntry> {
function buildFreshestSessionEntryIndex(store: SessionEntryReadView): Map<string, SessionEntry> {
const index = new Map<string, SessionEntry>();
for (const [key, candidate] of Object.entries(store)) {
const entry = asSessionEntry(candidate);
for (const { sessionKey: key, entry } of store.entries()) {
if (!entry) {
continue;
}
@ -252,7 +232,9 @@ function loadDeliverySessionEntry(params: {
}
| undefined;
for (const storePath of resolveDeliveryStorePaths(params.cfg, agentId)) {
const store = readSessionStoreSnapshot(storePath);
// Borrowed keyed view over this store's rows; exact probes stay cheap keyed reads and the
// borrowed rows are dropped before any await (this lookup is fully synchronous).
const store = openSessionEntryReadView({ storePath });
const entry = findSessionEntryInStore(store, sessionKeys);
const baseEntry = findSessionEntryInStore(store, baseKeys);
if (!entry && !baseEntry) {

53
src/config/sessions/goals.ts
View file

@ -1,7 +1,7 @@
// Session goal state tracks objective progress and token budgets in the session store.
import crypto from "node:crypto";
import { formatTokenCount } from "../../utils/token-format.js";
import { getSessionEntry, patchSessionEntry } from "./store.js";
import { loadSessionEntry, patchSessionEntry } from "./session-accessor.js";
import { resolveFreshSessionTotalTokens } from "./types.js";
import type { SessionEntry, SessionGoal, SessionGoalStatus } from "./types.js";
@ -162,7 +162,7 @@ export async function getSessionGoal(
if (options.persist === false) {
// Status rendering should not write incidental budget/baseline adoption unless callers opt in.
const entry =
getSessionEntry({ sessionKey: options.sessionKey, storePath: options.storePath }) ??
loadSessionEntry({ sessionKey: options.sessionKey, storePath: options.storePath }) ??
options.fallbackEntry;
const projected = entry
? resolveSessionGoalDisplayState(entry, now, { adoptFreshBaseline: false })
@ -170,11 +170,9 @@ export async function getSessionGoal(
return projected ? { status: "found", goal: projected } : { status: "missing" };
}
let goal: SessionGoal | undefined;
const result = await patchSessionEntry({
sessionKey: options.sessionKey,
storePath: options.storePath,
fallbackEntry: options.fallbackEntry,
update: (entry) => {
const result = await patchSessionEntry(
{ sessionKey: options.sessionKey, storePath: options.storePath },
(entry) => {
const accounted = accountGoalUsage(entry, now);
goal = accounted ? cloneGoal(accounted) : undefined;
if (!accounted || goalsEqual(accounted, entry.goal)) {
@ -182,7 +180,8 @@ export async function getSessionGoal(
}
return { goal: accounted };
},
});
{ fallbackEntry: options.fallbackEntry },
);
if (!result || !goal) {
return { status: "missing" };
}
@ -196,11 +195,9 @@ export async function createSessionGoal(options: CreateSessionGoalOptions): Prom
}
const now = nowMs(options.now);
let created: SessionGoal | undefined;
const result = await patchSessionEntry({
sessionKey: options.sessionKey,
storePath: options.storePath,
fallbackEntry: options.fallbackEntry,
update: (entry) => {
const result = await patchSessionEntry(
{ sessionKey: options.sessionKey, storePath: options.storePath },
(entry) => {
if (entry.goal) {
throw new Error("goal already exists");
}
@ -221,7 +218,8 @@ export async function createSessionGoal(options: CreateSessionGoalOptions): Prom
};
return { goal: created };
},
});
{ fallbackEntry: options.fallbackEntry },
);
if (!result || !created) {
throw new Error("session not found");
}
@ -234,10 +232,9 @@ export async function updateSessionGoalStatus(
const now = nowMs(options.now);
let updated: SessionGoal | undefined;
let foundSession = false;
const result = await patchSessionEntry({
sessionKey: options.sessionKey,
storePath: options.storePath,
update: (entry) => {
const result = await patchSessionEntry(
{ sessionKey: options.sessionKey, storePath: options.storePath },
(entry) => {
foundSession = true;
const accounted = accountGoalUsage(entry, now);
if (!accounted) {
@ -280,7 +277,7 @@ export async function updateSessionGoalStatus(
updated = next;
return { goal: updated };
},
});
);
if (!result || !updated) {
throw new Error(foundSession ? "goal not found" : "session not found");
}
@ -297,10 +294,9 @@ export async function updateSessionGoalObjective(
const now = nowMs(options.now);
let updated: SessionGoal | undefined;
let foundSession = false;
const result = await patchSessionEntry({
sessionKey: options.sessionKey,
storePath: options.storePath,
update: (entry) => {
const result = await patchSessionEntry(
{ sessionKey: options.sessionKey, storePath: options.storePath },
(entry) => {
foundSession = true;
const accounted = accountGoalUsage(entry, now);
if (!accounted) {
@ -313,7 +309,7 @@ export async function updateSessionGoalObjective(
updated = { ...accounted, objective, updatedAt: now };
return { goal: updated };
},
});
);
if (!result || !updated) {
throw new Error(foundSession ? "goal not found" : "session not found");
}
@ -322,16 +318,15 @@ export async function updateSessionGoalObjective(
export async function clearSessionGoal(options: SessionGoalStoreOptions): Promise<boolean> {
let removed = false;
const result = await patchSessionEntry({
sessionKey: options.sessionKey,
storePath: options.storePath,
update: (entry) => {
const result = await patchSessionEntry(
{ sessionKey: options.sessionKey, storePath: options.storePath },
(entry) => {
if (!entry.goal) {
return null;
}
removed = true;
return { goal: undefined };
},
});
);
return Boolean(result && removed);
}

4
src/config/sessions/inbound.runtime.ts
View file

@ -1,3 +1,3 @@
// Runtime facade for inbound session store updates.
// Runtime facade keeping inbound session persistence lazy behind the session accessor.
export { resolveStorePath } from "./paths.js";
export { recordSessionMetaFromInbound, updateLastRoute } from "./store.js";
export { recordInboundSessionMeta, updateSessionLastRoute } from "./session-accessor.js";

114
src/config/sessions/session-accessor.test.ts
View file

@ -3,6 +3,7 @@ import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { SessionManager } from "../../agents/sessions/session-manager.js";
import type { MsgContext } from "../../auto-reply/templating.js";
import { onSessionTranscriptUpdate } from "../../sessions/transcript-events.js";
import { createCanonicalFixtureSkill } from "../../skills/test-support/test-helpers.js";
import type { OpenClawConfig } from "../types.openclaw.js";
@ -21,12 +22,14 @@ import {
loadReplySessionInitializationSnapshot,
loadSessionEntry,
markSessionAbortTarget,
openSessionEntryReadView,
patchSessionEntry,
persistSessionResetLifecycle,
persistSessionTranscriptTurn,
purgeDeletedAgentSessionEntries,
publishTranscriptUpdate,
readSessionUpdatedAt,
recordInboundSessionMeta,
replaceSessionEntry,
resolveSessionEntryCandidateTarget,
resolveSessionEntryAccessTarget,
@ -37,6 +40,7 @@ import {
trimSessionTranscriptForManualCompact,
updateResolvedSessionEntry,
updateSessionEntry,
updateSessionLastRoute,
upsertSessionEntry,
} from "./session-accessor.js";
import * as sessionStore from "./store.js";
@ -97,6 +101,27 @@ describe("session accessor file-backed seam", () => {
});
});
it("opens a borrowed read view with raw exact-key probes and deferred enumeration", async () => {
const mixedKey = "agent:main:matrix:channel:!RoomAbC:example.org";
await upsertSessionEntry(
{ sessionKey: mixedKey, storePath },
{ sessionId: "mixed-session", updatedAt: 10 },
);
const view = openSessionEntryReadView({ storePath });
expect(view.get(mixedKey)?.sessionId).toBe("mixed-session");
// Raw probe contract: unlike loadSessionEntry, no folded-alias or
// canonical-key resolution happens on get.
expect(view.get(mixedKey.toLowerCase())).toBeUndefined();
expect(view.entries()).toEqual([
{
sessionKey: mixedKey,
entry: expect.objectContaining({ sessionId: "mixed-session" }),
},
]);
});
it("keeps case-distinct Matrix sessions separate under nested agent ownership", async () => {
const mixedKey = "agent:voice:agent:other:matrix:channel:!RoomAbC:example.org";
const lowerKey = "agent:voice:agent:other:matrix:channel:!Roomabc:example.org";
@ -118,6 +143,95 @@ describe("session accessor file-backed seam", () => {
]);
});
it("records inbound session meta as a createIfMissing upsert returning a detached entry", async () => {
const sessionKey = "agent:main:webchat:dm:user-1";
const ctx: MsgContext = {
Provider: "webchat",
Surface: "webchat",
ChatType: "direct",
From: "webchat:user-1",
To: "webchat:agent",
SessionKey: sessionKey,
OriginatingTo: "webchat:user-1",
};
const recorded = await recordInboundSessionMeta({ storePath, sessionKey, ctx });
expect(recorded?.origin?.provider).toBe("webchat");
// Detached result: caller mutations must never leak into cached store state.
if (recorded) {
recorded.origin = { provider: "mutated" };
}
expect(loadSessionEntry({ sessionKey, storePath })?.origin?.provider).toBe("webchat");
});
it("does not create sessions when inbound meta recording opts out of upsert", async () => {
const sessionKey = "agent:main:webchat:dm:absent";
const recorded = await recordInboundSessionMeta({
storePath,
sessionKey,
ctx: { Provider: "webchat", From: "webchat:absent", OriginatingTo: "webchat:absent" },
createIfMissing: false,
});
expect(recorded).toBeNull();
expect(loadSessionEntry({ sessionKey, storePath })).toBeUndefined();
});
it("preserves activity timestamps across inbound meta and last-route updates", async () => {
const sessionKey = "agent:main:webchat:dm:user-2";
const anchorUpdatedAt = Date.now() - 60_000;
await replaceSessionEntry(
{ sessionKey, storePath },
{ sessionId: "session-2", updatedAt: anchorUpdatedAt },
);
await recordInboundSessionMeta({
storePath,
sessionKey,
ctx: {
Provider: "webchat",
Surface: "webchat",
ChatType: "direct",
From: "webchat:user-2",
To: "webchat:agent",
SessionKey: sessionKey,
OriginatingTo: "webchat:user-2",
},
});
const afterMeta = loadSessionEntry({ sessionKey, storePath });
expect(afterMeta?.origin?.provider).toBe("webchat");
// Inbound metadata must not count as activity; idle reset relies on
// updatedAt moving only for real session turns.
expect(afterMeta?.updatedAt).toBe(anchorUpdatedAt);
const routed = await updateSessionLastRoute({
storePath,
sessionKey,
channel: "webchat",
to: "webchat:user-2",
});
expect(routed?.lastChannel).toBe("webchat");
const afterRoute = loadSessionEntry({ sessionKey, storePath });
expect(afterRoute?.lastTo).toBe("webchat:user-2");
expect(afterRoute?.route).toEqual({ channel: "webchat", target: { to: "webchat:user-2" } });
expect(afterRoute?.updatedAt).toBe(anchorUpdatedAt);
});
it("returns null from last-route updates for missing sessions when createIfMissing is false", async () => {
const sessionKey = "agent:main:webchat:dm:ghost";
const routed = await updateSessionLastRoute({
storePath,
sessionKey,
channel: "webchat",
to: "webchat:ghost",
createIfMissing: false,
});
expect(routed).toBeNull();
expect(loadSessionEntry({ sessionKey, storePath })).toBeUndefined();
});
it("marks abort targets while canonicalizing legacy session keys", async () => {
fs.writeFileSync(
storePath,

103
src/config/sessions/session-accessor.ts
View file

@ -8,6 +8,7 @@ import {
acquireSessionWriteLock,
resolveSessionWriteLockOptions,
} from "../../agents/session-write-lock.js";
import type { MsgContext } from "../../auto-reply/templating.js";
import {
resolveSessionStoreAgentId,
resolveSessionStoreKey,
@ -21,6 +22,7 @@ import type {
SessionTranscriptUpdateTarget,
} from "../../sessions/transcript-events.js";
import { createLazyRuntimeModule } from "../../shared/lazy-runtime.js";
import type { DeliveryContext } from "../../utils/delivery-context.types.js";
import type { OpenClawConfig } from "../types.openclaw.js";
import { formatSessionArchiveTimestamp } from "./artifacts.js";
import { extractGeneratedTranscriptSessionId } from "./generated-transcript-session-id.js";
@ -56,8 +58,10 @@ import {
purgeDeletedAgentSessionEntries as purgeFileDeletedAgentSessionEntries,
projectSessionEntryForPersistenceRevision,
readSessionUpdatedAt as readFileSessionUpdatedAt,
recordSessionMetaFromInbound as recordFileSessionMetaFromInbound,
resolveSessionStoreEntry,
resetSessionEntryLifecycle as resetFileSessionEntryLifecycle,
updateLastRoute as updateFileSessionLastRoute,
updateSessionStore,
updateSessionStoreEntry as updateFileSessionStoreEntry,
type DeleteSessionEntryLifecycleResult,
@ -102,7 +106,7 @@ import {
resolveOwnedSessionTranscriptWriteLockRunner,
withOwnedSessionTranscriptWrites,
} from "./transcript-write-context.js";
import type { SessionCompactionCheckpoint, SessionEntry } from "./types.js";
import type { GroupKeyResolution, SessionCompactionCheckpoint, SessionEntry } from "./types.js";
/**
* Session access API for callers that need entries or transcripts without
@ -257,6 +261,13 @@ export type SessionEntrySummary = {
entry: SessionEntry;
};
export type SessionEntryReadView = {
/** Row stored under the exact persisted key; no alias or canonical-key resolution. */
get(sessionKey: string): SessionEntry | undefined;
/** Every persisted row; call only when exact-key probes cannot settle the lookup. */
entries(): SessionEntrySummary[];
};
/** Raw transcript record for non-message events; message records use appendTranscriptMessage. */
export type TranscriptEvent = unknown;
@ -924,16 +935,31 @@ export function loadSessionEntry(scope: SessionAccessScope): SessionEntry | unde
/** Lists entries from the resolved store, preserving the persisted key for each row. */
export function listSessionEntries(scope: SessionEntryListScope = {}): SessionEntrySummary[] {
if (scope.clone === false) {
return Object.entries(
loadSessionStore(resolveSessionStorePathForScope({ ...scope, sessionKey: "" }), {
clone: false,
...(scope.hydrateSkillPromptRefs === false ? { hydrateSkillPromptRefs: false } : {}),
}),
).map(([sessionKey, entry]) => ({ sessionKey, entry }));
return openSessionEntryReadView(scope).entries();
}
return listFileSessionEntries(scope);
}
/**
* Borrowed keyed view over one resolved store for synchronous read-only hot paths.
* Unlike loadSessionEntry, `get` is a raw exact persisted-key probe with no alias
* or canonical-key resolution and no row scans, so large stores stay cheap until
* `entries` is called. Rows are borrowed, not cloned: callers must not mutate them
* and must drop the view before any await.
*/
export function openSessionEntryReadView(
scope: Omit<SessionEntryListScope, "clone" | "readConsistency"> = {},
): SessionEntryReadView {
const store = loadSessionStore(resolveSessionStorePathForScope({ ...scope, sessionKey: "" }), {
clone: false,
...(scope.hydrateSkillPromptRefs === false ? { hydrateSkillPromptRefs: false } : {}),
});
return {
get: (sessionKey) => (Object.hasOwn(store, sessionKey) ? store[sessionKey] : undefined),
entries: () => Object.entries(store).map(([sessionKey, entry]) => ({ sessionKey, entry })),
};
}
/** Reads the last activity timestamp for one session entry, or undefined when absent. */
export function readSessionUpdatedAt(scope: SessionAccessScope): number | undefined {
if (scope.storePath) {
@ -1403,6 +1429,69 @@ export async function updateSessionEntry(
});
}
export type RecordInboundSessionMetaParams = {
/** Set false to only patch existing entries; missing sessions stay absent. */
createIfMissing?: boolean;
/** Inbound message context whose stable metadata is derived and persisted. */
ctx: MsgContext;
/** Group routing resolution for group-owned session keys. */
groupResolution?: GroupKeyResolution | null;
/** Canonical or alias session key for the inbound conversation. */
sessionKey: string;
/** Explicit store target for file-backed stores and SQLite migration adapters. */
storePath: string;
};
export type UpdateSessionLastRouteParams = {
/** Account owning the delivery route when the channel is multi-account. */
accountId?: string;
/** Delivery channel id persisted as the last route channel. */
channel?: SessionEntry["lastChannel"];
/** Set false to only patch existing entries; missing sessions stay absent. */
createIfMissing?: boolean;
/** Optional inbound context whose session metadata is derived alongside the route. */
ctx?: MsgContext;
/** Explicit delivery context merged over the persisted session fallback. */
deliveryContext?: DeliveryContext;
/** Group routing resolution for group-owned session keys. */
groupResolution?: GroupKeyResolution | null;
/** Canonical channel route persisted as the session route slot. */
route?: SessionEntry["route"];
/** Canonical or alias session key for the routed conversation. */
sessionKey: string;
/** Explicit store target for file-backed stores and SQLite migration adapters. */
storePath: string;
/** Thread/topic id for the delivery route, when the transport has one. */
threadId?: string | number;
/** Delivery target persisted as the last route recipient. */
to?: string;
};
/**
* Records stable conversation metadata derived from one inbound message as a
* single storage-sized upsert (createIfMissing by default). Inbound metadata
* must not refresh activity timestamps — idle reset relies on updatedAt from
* real session turns — so existing rows merge with preserve-activity
* semantics while legacy alias keys collapse onto the canonical row.
*/
export async function recordInboundSessionMeta(
params: RecordInboundSessionMetaParams,
): Promise<SessionEntry | null> {
return await recordFileSessionMetaFromInbound(params);
}
/**
* Persists the last known delivery route for one session as a single
* storage-sized patch. Route updates preserve activity timestamps (#49515)
* and merge explicit route/delivery input over the persisted session
* fallback before normalizing the derived last* fields.
*/
export async function updateSessionLastRoute(
params: UpdateSessionLastRouteParams,
): Promise<SessionEntry | null> {
return await updateFileSessionLastRoute(params);
}
/** Resolves one abort target identity without exposing the mutable store. */
export function resolveSessionAbortTarget(
scope: SessionAccessScope,

26
src/config/sessions/store.session-key-normalization.test.ts
View file

@ -4,12 +4,8 @@ import path from "node:path";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
import type { MsgContext } from "../../auto-reply/templating.js";
import { createSuiteTempRootTracker } from "../../test-helpers/temp-dir.js";
import {
clearSessionStoreCacheForTest,
loadSessionStore,
recordSessionMetaFromInbound,
updateLastRoute,
} from "../sessions.js";
import { clearSessionStoreCacheForTest, loadSessionStore } from "../sessions.js";
import { recordInboundSessionMeta, updateSessionLastRoute } from "./session-accessor.js";
const CANONICAL_KEY = "agent:main:webchat:dm:mixed-user";
const MIXED_CASE_KEY = "Agent:Main:WebChat:DM:MiXeD-User";
@ -67,7 +63,7 @@ describe("session store key normalization", () => {
});
it("records inbound metadata under a canonical lowercase key", async () => {
await recordSessionMetaFromInbound({
await recordInboundSessionMeta({
storePath,
sessionKey: MIXED_CASE_KEY,
ctx: createInboundContext(),
@ -79,13 +75,13 @@ describe("session store key normalization", () => {
});
it("does not create a duplicate mixed-case key when last route is updated", async () => {
await recordSessionMetaFromInbound({
await recordInboundSessionMeta({
storePath,
sessionKey: CANONICAL_KEY,
ctx: createInboundContext(),
});
await updateLastRoute({
await updateSessionLastRoute({
storePath,
sessionKey: MIXED_CASE_KEY,
channel: "webchat",
@ -121,7 +117,7 @@ describe("session store key normalization", () => {
);
clearSessionStoreCacheForTest();
await updateLastRoute({
await updateSessionLastRoute({
storePath,
sessionKey: CANONICAL_KEY,
channel: "webchat",
@ -162,7 +158,7 @@ describe("session store key normalization", () => {
);
clearSessionStoreCacheForTest();
await recordSessionMetaFromInbound({
await recordInboundSessionMeta({
storePath,
sessionKey: CANONICAL_KEY,
ctx: {},
@ -200,7 +196,7 @@ describe("session store key normalization", () => {
);
clearSessionStoreCacheForTest();
await recordSessionMetaFromInbound({
await recordInboundSessionMeta({
storePath,
sessionKey: CANONICAL_KEY,
ctx: createInboundContext(),
@ -213,7 +209,7 @@ describe("session store key normalization", () => {
});
it("records Signal group metadata under the mixed-case opaque group id", async () => {
await recordSessionMetaFromInbound({
await recordInboundSessionMeta({
storePath,
sessionKey: `Agent:Main:Signal:Group:${SIGNAL_GROUP_ID}`,
ctx: createSignalGroupContext(),
@ -249,7 +245,7 @@ describe("session store key normalization", () => {
);
clearSessionStoreCacheForTest();
await recordSessionMetaFromInbound({
await recordInboundSessionMeta({
storePath,
sessionKey: SIGNAL_GROUP_KEY,
ctx: createSignalGroupContext(),
@ -263,7 +259,7 @@ describe("session store key normalization", () => {
});
it("stores canonical route metadata and derives legacy delivery fields", async () => {
await updateLastRoute({
await updateSessionLastRoute({
storePath,
sessionKey: CANONICAL_KEY,
route: {

15
src/infra/outbound/outbound-session.test.ts
View file

@ -13,7 +13,7 @@ type InboundMetadataParams = {
};
const mocks = vi.hoisted(() => ({
recordSessionMetaFromInbound: vi.fn(async (_params: InboundMetadataParams) => ({ ok: true })),
recordInboundSessionMeta: vi.fn(async (_params: InboundMetadataParams) => ({ ok: true })),
resolveStorePath: vi.fn(
(_store: unknown, params?: { agentId?: string }) => `/stores/${params?.agentId ?? "main"}.json`,
),
@ -35,13 +35,13 @@ function firstMockArg(
}
vi.mock("../../config/sessions/inbound.runtime.js", () => ({
recordSessionMetaFromInbound: mocks.recordSessionMetaFromInbound,
recordInboundSessionMeta: mocks.recordInboundSessionMeta,
resolveStorePath: mocks.resolveStorePath,
}));
describe("resolveOutboundSessionRoute", () => {
beforeEach(() => {
mocks.recordSessionMetaFromInbound.mockClear();
mocks.recordInboundSessionMeta.mockClear();
mocks.resolveStorePath.mockClear();
setMinimalOutboundSessionPluginRegistryForTests();
});
@ -544,7 +544,7 @@ describe("resolveOutboundSessionRoute", () => {
describe("ensureOutboundSessionEntry", () => {
beforeEach(() => {
mocks.recordSessionMetaFromInbound.mockClear();
mocks.recordInboundSessionMeta.mockClear();
mocks.resolveStorePath.mockClear();
});
@ -569,11 +569,8 @@ describe("ensureOutboundSessionEntry", () => {
expect(mocks.resolveStorePath).toHaveBeenCalledWith("/stores/{agentId}.json", {
agentId: "main",
});
expect(mocks.recordSessionMetaFromInbound).toHaveBeenCalledOnce();
const metadata = firstMockArg(
mocks.recordSessionMetaFromInbound,
"recordSessionMetaFromInbound",
);
expect(mocks.recordInboundSessionMeta).toHaveBeenCalledOnce();
const metadata = firstMockArg(mocks.recordInboundSessionMeta, "recordInboundSessionMeta");
expect(metadata.storePath).toBe("/stores/main.json");
expect(metadata.sessionKey).toBe("agent:main:workspace:channel:c1");
});

4
src/infra/outbound/outbound-session.ts
View file

@ -8,7 +8,7 @@ import { getChannelPlugin } from "../../channels/plugins/index.js";
import type { ChannelPlugin } from "../../channels/plugins/types.plugin.js";
import type { ChannelId } from "../../channels/plugins/types.public.js";
import {
recordSessionMetaFromInbound,
recordInboundSessionMeta,
resolveStorePath,
} from "../../config/sessions/inbound.runtime.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
@ -226,7 +226,7 @@ export async function ensureOutboundSessionEntry(params: {
OriginatingTo: params.route.to,
};
try {
await recordSessionMetaFromInbound({
await recordInboundSessionMeta({
storePath,
sessionKey: params.route.sessionKey,
ctx,

8
src/plugin-sdk/config-runtime.ts
View file

@ -150,14 +150,12 @@ export type {
} from "../config/types.js";
export {
clearSessionStoreCacheForTest,
recordSessionMetaFromInbound,
/**
* @deprecated Use patchSessionEntry/upsertSessionEntry for writes. This
* whole-store helper is kept only during the transition before SQLite
* migration. Callers must migrate away from writing sessions.json directly.
*/
saveSessionStore,
updateLastRoute,
/**
* @deprecated Use patchSessionEntry/upsertSessionEntry for writes. This
* whole-store helper is kept only during the transition before SQLite
@ -166,6 +164,12 @@ export {
updateSessionStore,
resolveSessionStoreEntry,
} from "../config/sessions/store.js";
// SDK-facing names are a shipped plugin contract; internals route through the
// session accessor so the storage backend can change beneath them.
export {
recordInboundSessionMeta as recordSessionMetaFromInbound,
updateSessionLastRoute as updateLastRoute,
} from "../config/sessions/session-accessor.js";
export { resolveSessionKey } from "../config/sessions/session-key.js";
export { resolveStorePath } from "../config/sessions/paths.js";
export type { SessionResetMode } from "../config/sessions/reset.js";

10
src/plugin-sdk/session-store-runtime.ts
View file

@ -227,11 +227,13 @@ export {
export { resolveSessionKey } from "../config/sessions/session-key.js";
export { resolveGroupSessionKey } from "../config/sessions/group.js";
export { canonicalizeMainSessionAlias } from "../config/sessions/main-session.js";
export { clearSessionStoreCacheForTest } from "../config/sessions/store.js";
// SDK-facing names are a shipped plugin contract; internals route through the
// session accessor so the storage backend can change beneath them.
export {
clearSessionStoreCacheForTest,
recordSessionMetaFromInbound,
updateLastRoute,
} from "../config/sessions/store.js";
recordInboundSessionMeta as recordSessionMetaFromInbound,
updateSessionLastRoute as updateLastRoute,
} from "../config/sessions/session-accessor.js";
/**
* @deprecated Use patchSessionEntry/upsertSessionEntry for writes. These
* whole-store helpers are kept only during the transition before SQLite

18
src/plugins/runtime/runtime-channel.ts
View file

@ -63,13 +63,13 @@ import {
resolveChannelGroupRequireMention,
} from "../../config/group-policy.js";
import { resolveMarkdownTableMode } from "../../config/markdown-tables.js";
import {
recordSessionMetaFromInbound,
resolveStorePath,
updateLastRoute,
} from "../../config/sessions.js";
import { resolveStorePath } from "../../config/sessions.js";
import { resolveSessionEntryResetFreshness } from "../../config/sessions/entry-freshness.js";
import { readSessionUpdatedAt } from "../../config/sessions/session-accessor.js";
import {
readSessionUpdatedAt,
recordInboundSessionMeta,
updateSessionLastRoute,
} from "../../config/sessions/session-accessor.js";
import { getChannelActivity, recordChannelActivity } from "../../infra/channel-activity.js";
import {
fetchRemoteMedia,
@ -91,9 +91,11 @@ export function createRuntimeChannel(): PluginRuntime["channel"] {
const sessionRuntime = {
resolveStorePath,
readSessionUpdatedAt,
recordSessionMetaFromInbound,
// Plugin runtime property names are a shipped contract; the implementations
// route through the session accessor boundary.
recordSessionMetaFromInbound: recordInboundSessionMeta,
recordInboundSession,
updateLastRoute,
updateLastRoute: updateSessionLastRoute,
resolveEntryResetFreshness: resolveSessionEntryResetFreshness,
};
const channelRuntime = {

6
test/scripts/check-session-accessor-boundary.test.ts
View file

@ -54,6 +54,8 @@ describe("session accessor boundary guard", () => {
"src/commands/status.summary.ts",
"src/commands/tasks.ts",
"src/config/sessions/combined-store-gateway.ts",
"src/config/sessions/delivery-info.ts",
"src/config/sessions/goals.ts",
"src/cron/isolated-agent/delivery-target.ts",
"src/cron/service/timer.ts",
"src/gateway/session-compaction-checkpoints.ts",
@ -120,6 +122,7 @@ describe("session accessor boundary guard", () => {
"src/agents/command/session-store.ts",
"src/agents/embedded-agent-runner/run.ts",
"src/agents/embedded-agent-runner/run/attempt.ts",
"src/agents/embedded-agent-subscribe.handlers.compaction.runtime.ts",
"src/agents/live-model-switch.ts",
"src/agents/main-session-restart-recovery.ts",
"src/auto-reply/reply/abort.ts",
@ -148,12 +151,15 @@ describe("session accessor boundary guard", () => {
"src/auto-reply/reply/session-usage.ts",
"src/commands/tasks.ts",
"src/config/sessions/cleanup-service.ts",
"src/config/sessions/goals.ts",
"src/gateway/boot.ts",
"src/gateway/server-methods/sessions.ts",
"src/gateway/server-node-events.ts",
"src/gateway/session-compaction-checkpoints.ts",
"src/infra/outbound/outbound-session.ts",
"src/plugins/host-hook-cleanup.ts",
"src/plugins/host-hook-state.ts",
"src/plugins/runtime/runtime-channel.ts",
"src/tui/embedded-backend.ts",
]),
);