mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 10:16:44 +00:00
refactor: add SDK transcript identity target API (#95030)
This commit is contained in:
Josh Lehman
@@ -1,2 +1,2 @@
|
||||
118c0f05ded3d3671e4caca646f8c5c13799757705fec2d769b1657367ec0243 plugin-sdk-api-baseline.json
|
||||
6795c59b8ce6c8203bfca5d932b562d3d2b718e93701faa3a52e57cb45d277d4 plugin-sdk-api-baseline.jsonl
|
||||
9edb033535fe1325c18b431190672dc3a826dba312e376c13c98fcf9043060dd plugin-sdk-api-baseline.json
|
||||
78f26963fe2e6d7903ce2e1067699200d825f391c0010df46f48d9abd2915e65 plugin-sdk-api-baseline.jsonl
|
||||
|
||||
@@ -166,7 +166,9 @@ two-party event loops that do not go through the shared inbound reply runner.
|
||||
|
||||
Prefer `getSessionEntry(...)`, `listSessionEntries(...)`, `patchSessionEntry(...)`, or `upsertSessionEntry(...)` for session workflows. These helpers address sessions by agent/session identity so plugins do not depend on the legacy `sessions.json` storage shape. Use `preserveActivity: true` for metadata-only patches that should not refresh session activity, and `replaceEntry: true` only when the callback returns a complete entry and deleted fields must stay deleted.
|
||||
|
||||
`loadSessionStore(...)`, `saveSessionStore(...)`, `updateSessionStore(...)`, and `resolveSessionFilePath(...)` are kept only during the transition before SQLite migration for plugins that still intentionally depend on the legacy whole-store or transcript-file shape. New plugin code must not use those helpers, and existing callers must migrate to entry helpers before the SQLite storage flip.
|
||||
For transcript reads and writes, import `openclaw/plugin-sdk/session-transcript-runtime` and use `resolveSessionTranscriptIdentity(...)`, `resolveSessionTranscriptTarget(...)`, `readSessionTranscriptEvents(...)`, `appendSessionTranscriptMessageByIdentity(...)`, `publishSessionTranscriptUpdateByIdentity(...)`, or `withSessionTranscriptWriteLock(...)` with `{ agentId, sessionKey, sessionId }`. These APIs let plugins identify a transcript, read its events, append messages, publish updates, and run related operations under the same transcript write lock. Pass `sessionFile` only when adapting code that already receives an active transcript artifact and needs each helper to operate on that same artifact.
|
||||
|
||||
`loadSessionStore(...)`, `saveSessionStore(...)`, `updateSessionStore(...)`, and `resolveSessionFilePath(...)` are compatibility helpers for plugins that still intentionally depend on the legacy whole-store or transcript-file shape. New plugin code must not use those helpers, and existing callers should migrate to entry helpers.
|
||||
|
||||
</Accordion>
|
||||
<Accordion title="api.runtime.agent.defaults">
|
||||
|
||||
@@ -248,6 +248,7 @@ usage endpoint failed or returned no usable usage data.
|
||||
| `plugin-sdk/reply-reference` | `createReplyReferencePlanner` |
|
||||
| `plugin-sdk/reply-chunking` | Narrow text/markdown chunking helpers |
|
||||
| `plugin-sdk/session-store-runtime` | Session workflow helpers (`getSessionEntry`, `listSessionEntries`, `patchSessionEntry`, `upsertSessionEntry`), legacy session store path/session-key helpers, updated-at reads, and transition-only whole-store/file-path compatibility helpers |
|
||||
| `plugin-sdk/session-transcript-runtime` | Transcript identity, scoped target/read/write helpers, update publishing, write locks, and transcript memory hit keys |
|
||||
| `plugin-sdk/sqlite-runtime` | Focused SQLite agent-schema, path, and transaction helpers for first-party runtime |
|
||||
| `plugin-sdk/cron-store-runtime` | Cron store path/load/save helpers |
|
||||
| `plugin-sdk/state-paths` | State/OAuth dir path helpers |
|
||||
|
||||
@@ -975,6 +975,10 @@
|
||||
"types": "./dist/plugin-sdk/session-store-runtime.d.ts",
|
||||
"default": "./dist/plugin-sdk/session-store-runtime.js"
|
||||
},
|
||||
"./plugin-sdk/session-transcript-runtime": {
|
||||
"types": "./dist/plugin-sdk/session-transcript-runtime.d.ts",
|
||||
"default": "./dist/plugin-sdk/session-transcript-runtime.js"
|
||||
},
|
||||
"./plugin-sdk/sqlite-runtime": {
|
||||
"types": "./dist/plugin-sdk/sqlite-runtime.d.ts",
|
||||
"default": "./dist/plugin-sdk/sqlite-runtime.js"
|
||||
|
||||
@@ -114,6 +114,9 @@ export const pluginSdkDocMetadata = {
|
||||
"runtime-store": {
|
||||
category: "runtime",
|
||||
},
|
||||
"session-transcript-runtime": {
|
||||
category: "runtime",
|
||||
},
|
||||
"sqlite-runtime": {
|
||||
category: "runtime",
|
||||
},
|
||||
|
||||
@@ -217,6 +217,7 @@
|
||||
"session-binding-runtime",
|
||||
"session-key-runtime",
|
||||
"session-store-runtime",
|
||||
"session-transcript-runtime",
|
||||
"sqlite-runtime",
|
||||
"sqlite-runtime-testing",
|
||||
"session-transcript-hit",
|
||||
|
||||
@@ -125,6 +125,7 @@ const defaultPublicDeprecatedExportsByEntrypointBudget = Object.freeze({
|
||||
"channel-policy": 8,
|
||||
"channel-route": 5,
|
||||
"session-store-runtime": 1,
|
||||
"session-transcript-runtime": 1,
|
||||
"group-access": 13,
|
||||
"media-generation-runtime-shared": 3,
|
||||
"music-generation-core": 20,
|
||||
@@ -161,9 +162,9 @@ let budgets;
|
||||
let publicDeprecatedExportsByEntrypointBudget;
|
||||
try {
|
||||
budgets = {
|
||||
publicEntrypoints: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_ENTRYPOINTS", 320),
|
||||
publicExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", 10301),
|
||||
publicFunctionExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", 5171),
|
||||
publicEntrypoints: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_ENTRYPOINTS", 321),
|
||||
publicExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", 10331),
|
||||
publicFunctionExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", 5183),
|
||||
publicDeprecatedExports: readBudgetEnv(
|
||||
"OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_DEPRECATED_EXPORTS",
|
||||
3245,
|
||||
|
||||
@@ -866,7 +866,11 @@ export async function loadTranscriptEvents(
|
||||
const transcript = await resolveTranscriptReadAccess(scope);
|
||||
const events: TranscriptEvent[] = [];
|
||||
for await (const line of streamSessionTranscriptLines(transcript.sessionFile)) {
|
||||
events.push(JSON.parse(line) as TranscriptEvent);
|
||||
try {
|
||||
events.push(JSON.parse(line) as TranscriptEvent);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
@@ -26,7 +26,10 @@ import {
|
||||
streamSessionTranscriptLinesReverse,
|
||||
} from "./transcript-stream.js";
|
||||
import { isCanonicalSessionTranscriptEntry } from "./transcript-tree.js";
|
||||
import { resolveOwnedSessionTranscriptWriteLockRunner } from "./transcript-write-context.js";
|
||||
import {
|
||||
resolveOwnedSessionTranscriptWriteLockRunner,
|
||||
type OwnedSessionTranscriptPublishedEntry,
|
||||
} from "./transcript-write-context.js";
|
||||
import { CURRENT_SESSION_VERSION } from "./version.js";
|
||||
|
||||
const SESSION_MANAGER_APPEND_MAX_BYTES = 8 * 1024 * 1024;
|
||||
@@ -385,6 +388,13 @@ export type AppendSessionTranscriptMessageResult<TMessage> = {
|
||||
appended: boolean;
|
||||
};
|
||||
|
||||
export type SessionTranscriptAppendTransactionContext = {
|
||||
appendEvent: (event: unknown) => Promise<void>;
|
||||
appendMessage: <TMessage>(
|
||||
params: Omit<AppendSessionTranscriptMessageParams<TMessage>, "config" | "transcriptPath">,
|
||||
) => Promise<AppendSessionTranscriptMessageResult<TMessage> | undefined>;
|
||||
};
|
||||
|
||||
function isTranscriptAgentMessage(value: unknown): value is AgentMessage {
|
||||
return (
|
||||
typeof value === "object" &&
|
||||
@@ -464,6 +474,57 @@ export async function appendSessionTranscriptMessageWithOwnedWriteLock<TMessage>
|
||||
return await activeLockRunner(() => appendSessionTranscriptMessageLocked(params));
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs a group of transcript appends through one append queue and write lock.
|
||||
*/
|
||||
export async function runSessionTranscriptAppendTransaction<T>(
|
||||
params: Pick<AppendSessionTranscriptMessageParams, "config" | "transcriptPath">,
|
||||
run: (context: SessionTranscriptAppendTransactionContext) => Promise<T> | T,
|
||||
): Promise<T> {
|
||||
const publishedEntries: OwnedSessionTranscriptPublishedEntry[] = [];
|
||||
const runTransaction = async (): Promise<T> =>
|
||||
await run({
|
||||
appendEvent: async (event) => {
|
||||
const result = await appendSessionTranscriptEventLocked({
|
||||
config: params.config,
|
||||
event,
|
||||
transcriptPath: params.transcriptPath,
|
||||
});
|
||||
publishedEntries.push({ kind: "serialized", serialized: result.serializedEntry });
|
||||
},
|
||||
appendMessage: async (messageParams) => {
|
||||
const result = await appendSessionTranscriptMessageLocked({
|
||||
...messageParams,
|
||||
config: params.config,
|
||||
onHeaderCreated: (header) => {
|
||||
publishedEntries.push({ kind: "header", serialized: header });
|
||||
},
|
||||
transcriptPath: params.transcriptPath,
|
||||
});
|
||||
if (result?.appended === true) {
|
||||
publishedEntries.push({ kind: "id", id: result.messageId });
|
||||
}
|
||||
return result;
|
||||
},
|
||||
});
|
||||
const activeLockRunner = resolveOwnedSessionTranscriptWriteLockRunner({
|
||||
sessionFile: params.transcriptPath,
|
||||
});
|
||||
if (activeLockRunner) {
|
||||
return await activeLockRunner(
|
||||
() => withSessionTranscriptAppendQueue(params.transcriptPath, runTransaction),
|
||||
{
|
||||
publishOwnedWrite: true,
|
||||
resolvePublishedEntries: () => publishedEntries,
|
||||
resolvePublishedEntriesAfterFailure: () => publishedEntries,
|
||||
},
|
||||
);
|
||||
}
|
||||
return await withSessionTranscriptAppendQueue(params.transcriptPath, () =>
|
||||
withSessionTranscriptWriteLock(params, runTransaction),
|
||||
);
|
||||
}
|
||||
|
||||
type AppendSessionTranscriptEventParams = {
|
||||
config?: OpenClawConfig;
|
||||
event: unknown;
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
// Session transcript hit tests cover transcript match formatting and path resolution.
|
||||
import fs from "node:fs";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { SessionEntry } from "../config/sessions/types.js";
|
||||
import {
|
||||
extractTranscriptIdentityFromSessionsMemoryHit,
|
||||
extractTranscriptStemFromSessionsMemoryHit,
|
||||
formatSessionTranscriptMemoryHitKey,
|
||||
parseSessionTranscriptMemoryHitKey,
|
||||
resolveSessionTranscriptMemoryHitKeyToSessionKeys,
|
||||
resolveTranscriptStemToSessionKeys,
|
||||
} from "./session-transcript-hit.js";
|
||||
|
||||
@@ -273,3 +277,34 @@ describe("resolveTranscriptStemToSessionKeys", () => {
|
||||
).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("session transcript memory hit key compatibility exports", () => {
|
||||
it("keeps hit-subpath memory helpers off the runtime writer import path", () => {
|
||||
const source = fs.readFileSync(new URL("./session-transcript-hit.ts", import.meta.url), "utf8");
|
||||
|
||||
expect(source).not.toContain("session-transcript-runtime.js");
|
||||
});
|
||||
|
||||
it("exports storage-neutral memory hit key helpers from the legacy hit subpath", () => {
|
||||
const key = formatSessionTranscriptMemoryHitKey({
|
||||
agentId: "main",
|
||||
sessionId: "session:legacy",
|
||||
});
|
||||
const store: Record<string, SessionEntry> = {
|
||||
"agent:main:discord:direct:42": {
|
||||
sessionFile: "/tmp/not-the-identity.jsonl",
|
||||
sessionId: "session:legacy",
|
||||
updatedAt: 10,
|
||||
},
|
||||
};
|
||||
|
||||
expect(key).toBe("transcript:main:session%3Alegacy");
|
||||
expect(parseSessionTranscriptMemoryHitKey(key)).toMatchObject({
|
||||
agentId: "main",
|
||||
sessionId: "session:legacy",
|
||||
});
|
||||
expect(resolveSessionTranscriptMemoryHitKeyToSessionKeys({ key, store })).toEqual([
|
||||
"agent:main:discord:direct:42",
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -5,6 +5,19 @@ import { uniqueStrings } from "../../packages/normalization-core/src/string-norm
|
||||
import { parseUsageCountedSessionIdFromFileName } from "../config/sessions/artifacts.js";
|
||||
import type { SessionEntry } from "../config/sessions/types.js";
|
||||
import { normalizeAgentId } from "../routing/session-key.js";
|
||||
export {
|
||||
formatSessionTranscriptMemoryHitKey,
|
||||
parseSessionTranscriptMemoryHitKey,
|
||||
resolveSessionTranscriptMemoryHitKeyToSessionKeys,
|
||||
} from "./session-transcript-memory-hit.js";
|
||||
export type {
|
||||
ResolveSessionTranscriptMemoryHitKeyParams,
|
||||
SessionTranscriptIdentity,
|
||||
SessionTranscriptMemoryHitIdentity,
|
||||
SessionTranscriptMemoryHitKey,
|
||||
SessionTranscriptMemoryHitKeyParams,
|
||||
SessionTranscriptReadParams,
|
||||
} from "./session-transcript-memory-hit.js";
|
||||
|
||||
export { loadCombinedSessionStoreForGateway } from "../config/sessions/combined-store-gateway.js";
|
||||
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
import { normalizeOptionalString } from "../../packages/normalization-core/src/string-coerce.js";
|
||||
import { uniqueStrings } from "../../packages/normalization-core/src/string-normalization.js";
|
||||
import type { SessionEntry } from "../config/sessions/types.js";
|
||||
import { normalizeAgentId, resolveAgentIdFromSessionKey } from "../routing/session-key.js";
|
||||
|
||||
const SESSION_TRANSCRIPT_MEMORY_HIT_PREFIX = "transcript";
|
||||
|
||||
export type SessionTranscriptIdentity = {
|
||||
agentId: string;
|
||||
memoryKey: SessionTranscriptMemoryHitKey;
|
||||
sessionId: string;
|
||||
sessionKey: string;
|
||||
};
|
||||
|
||||
export type SessionTranscriptMemoryHitIdentity = {
|
||||
agentId: string;
|
||||
key: SessionTranscriptMemoryHitKey;
|
||||
sessionId: string;
|
||||
};
|
||||
|
||||
export type SessionTranscriptMemoryHitKey = `transcript:${string}:${string}`;
|
||||
|
||||
export type SessionTranscriptReadParams = {
|
||||
agentId?: string;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
hydrateSkillPromptRefs?: boolean;
|
||||
sessionId: string;
|
||||
sessionKey: string;
|
||||
storePath?: string;
|
||||
threadId?: string | number;
|
||||
};
|
||||
|
||||
export type SessionTranscriptMemoryHitKeyParams = {
|
||||
agentId: string;
|
||||
sessionId: string;
|
||||
};
|
||||
|
||||
export type ResolveSessionTranscriptMemoryHitKeyParams = {
|
||||
includeSyntheticFallback?: boolean;
|
||||
key: string;
|
||||
store: Record<string, SessionEntry>;
|
||||
};
|
||||
|
||||
function requireMemoryKeySegment(value: string, label: string): string {
|
||||
const normalized = normalizeOptionalString(value);
|
||||
if (!normalized) {
|
||||
throw new Error(`Cannot build session transcript memory hit key without ${label}.`);
|
||||
}
|
||||
return encodeURIComponent(normalized);
|
||||
}
|
||||
|
||||
function decodeMemoryKeySegment(value: string): string | null {
|
||||
try {
|
||||
return normalizeOptionalString(decodeURIComponent(value)) ?? null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function syntheticSessionKey(identity: SessionTranscriptMemoryHitIdentity): string {
|
||||
return `agent:${identity.agentId}:${identity.sessionId}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds the memory hit key for one session transcript.
|
||||
*/
|
||||
export function formatSessionTranscriptMemoryHitKey(
|
||||
params: SessionTranscriptMemoryHitKeyParams,
|
||||
): SessionTranscriptMemoryHitKey {
|
||||
const agentId = requireMemoryKeySegment(normalizeAgentId(params.agentId), "agentId");
|
||||
const sessionId = requireMemoryKeySegment(params.sessionId, "sessionId");
|
||||
return `${SESSION_TRANSCRIPT_MEMORY_HIT_PREFIX}:${agentId}:${sessionId}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parses a session transcript memory hit key.
|
||||
*/
|
||||
export function parseSessionTranscriptMemoryHitKey(
|
||||
key: string,
|
||||
): SessionTranscriptMemoryHitIdentity | null {
|
||||
const parts = key.split(":");
|
||||
if (parts.length !== 3 || parts[0] !== SESSION_TRANSCRIPT_MEMORY_HIT_PREFIX) {
|
||||
return null;
|
||||
}
|
||||
const agentId = decodeMemoryKeySegment(parts[1] ?? "");
|
||||
const sessionId = decodeMemoryKeySegment(parts[2] ?? "");
|
||||
if (!agentId || !sessionId) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
agentId: normalizeAgentId(agentId),
|
||||
key: formatSessionTranscriptMemoryHitKey({ agentId, sessionId }),
|
||||
sessionId,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Maps a session transcript memory hit key back to visible session store keys.
|
||||
*/
|
||||
export function resolveSessionTranscriptMemoryHitKeyToSessionKeys(
|
||||
params: ResolveSessionTranscriptMemoryHitKeyParams,
|
||||
): string[] {
|
||||
const identity = parseSessionTranscriptMemoryHitKey(params.key);
|
||||
if (!identity) {
|
||||
return [];
|
||||
}
|
||||
const matches = Object.entries(params.store)
|
||||
.filter(([sessionKey, entry]) => {
|
||||
return (
|
||||
entry.sessionId === identity.sessionId &&
|
||||
normalizeAgentId(resolveAgentIdFromSessionKey(sessionKey)) === identity.agentId
|
||||
);
|
||||
})
|
||||
.map(([sessionKey]) => sessionKey);
|
||||
const deduped = uniqueStrings(matches);
|
||||
if (deduped.length > 0) {
|
||||
return deduped;
|
||||
}
|
||||
return params.includeSyntheticFallback === false ? [] : [syntheticSessionKey(identity)];
|
||||
}
|
||||
@@ -0,0 +1,406 @@
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { appendTranscriptEvent, upsertSessionEntry } from "../config/sessions/session-accessor.js";
|
||||
import { loadSessionStore } from "../config/sessions/store.js";
|
||||
import { withOwnedSessionTranscriptWrites } from "../config/sessions/transcript-write-context.js";
|
||||
import * as transcriptEvents from "../sessions/transcript-events.js";
|
||||
import {
|
||||
appendSessionTranscriptMessageByIdentity,
|
||||
formatSessionTranscriptMemoryHitKey,
|
||||
parseSessionTranscriptMemoryHitKey,
|
||||
publishSessionTranscriptUpdateByIdentity,
|
||||
readSessionTranscriptEvents,
|
||||
resolveSessionTranscriptIdentity,
|
||||
resolveSessionTranscriptTarget,
|
||||
resolveSessionTranscriptMemoryHitKeyToSessionKeys,
|
||||
withSessionTranscriptWriteLock,
|
||||
} from "./session-transcript-runtime.js";
|
||||
|
||||
describe("session transcript runtime SDK", () => {
|
||||
let tempDir: string;
|
||||
let storePath: string;
|
||||
|
||||
beforeEach(() => {
|
||||
tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sdk-transcript-"));
|
||||
storePath = path.join(tempDir, "sessions.json");
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
fs.rmSync(tempDir, { force: true, recursive: true });
|
||||
});
|
||||
|
||||
it("resolves transcript identity and reads events without returning sessionFile", async () => {
|
||||
const scope = {
|
||||
agentId: "Main",
|
||||
sessionId: "session-with-colon",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const event = { id: "event-1", type: "metadata" };
|
||||
|
||||
await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 });
|
||||
await appendTranscriptEvent(scope, event);
|
||||
|
||||
const identity = await resolveSessionTranscriptIdentity(scope);
|
||||
|
||||
expect(identity).toEqual({
|
||||
agentId: "main",
|
||||
memoryKey: "transcript:main:session-with-colon",
|
||||
sessionId: scope.sessionId,
|
||||
sessionKey: "agent:main:main",
|
||||
});
|
||||
expect(identity).not.toHaveProperty("sessionFile");
|
||||
await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([event]);
|
||||
});
|
||||
|
||||
it("does not persist sessionFile metadata for identity-only reads", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionId: "read-only-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
|
||||
await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 });
|
||||
|
||||
await expect(resolveSessionTranscriptIdentity(scope)).resolves.toMatchObject({
|
||||
memoryKey: "transcript:main:read-only-session",
|
||||
});
|
||||
await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([]);
|
||||
expect(loadSessionStore(storePath)[scope.sessionKey]?.sessionFile).toBeUndefined();
|
||||
});
|
||||
|
||||
it("skips malformed transcript lines when reading by scoped identity", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "malformed-lines.jsonl"),
|
||||
sessionId: "malformed-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const firstEvent = { id: "event-valid-1", type: "message" };
|
||||
const secondEvent = { id: "event-valid-2", type: "message" };
|
||||
|
||||
fs.writeFileSync(
|
||||
scope.sessionFile,
|
||||
[
|
||||
JSON.stringify(firstEvent),
|
||||
"{malformed-json",
|
||||
JSON.stringify(secondEvent),
|
||||
"not-json",
|
||||
"",
|
||||
].join("\n"),
|
||||
"utf8",
|
||||
);
|
||||
|
||||
await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([firstEvent, secondEvent]);
|
||||
});
|
||||
|
||||
it("binds scoped reads to an explicit active transcript file without exposing it", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "active-session.jsonl"),
|
||||
sessionId: "active-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const event = { id: "event-active", type: "metadata" };
|
||||
|
||||
await upsertSessionEntry(scope, {
|
||||
sessionFile: path.join(tempDir, "store-default.jsonl"),
|
||||
sessionId: scope.sessionId,
|
||||
updatedAt: 10,
|
||||
});
|
||||
await appendTranscriptEvent(scope, event);
|
||||
|
||||
const target = await resolveSessionTranscriptTarget(scope);
|
||||
|
||||
expect(target).toEqual({
|
||||
agentId: "main",
|
||||
memoryKey: "transcript:main:active-session",
|
||||
sessionId: "active-session",
|
||||
sessionKey: "agent:main:main",
|
||||
targetKind: "active-session-file",
|
||||
});
|
||||
expect(target).not.toHaveProperty("sessionFile");
|
||||
await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([event]);
|
||||
expect(fs.readFileSync(scope.sessionFile, "utf8")).toContain("event-active");
|
||||
});
|
||||
|
||||
it("appends messages by the same explicit scoped transcript target", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "mirror-target.jsonl"),
|
||||
sessionId: "mirror-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const message = {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "hello" }],
|
||||
timestamp: 1,
|
||||
};
|
||||
|
||||
const appended = await appendSessionTranscriptMessageByIdentity({
|
||||
...scope,
|
||||
message,
|
||||
});
|
||||
|
||||
expect(appended).toBeDefined();
|
||||
expect(appended?.message).toMatchObject(message);
|
||||
await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([
|
||||
expect.objectContaining({ type: "session" }),
|
||||
expect.objectContaining({ message: expect.objectContaining({ role: "assistant" }) }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("publishes updates with the resolved scoped transcript identity", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "publish-target.jsonl"),
|
||||
sessionId: "publish-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const emitSpy = vi.spyOn(transcriptEvents, "emitSessionTranscriptUpdate");
|
||||
|
||||
await publishSessionTranscriptUpdateByIdentity({
|
||||
...scope,
|
||||
update: {
|
||||
agentId: "stale-agent",
|
||||
messageId: "message-from-direct-publish",
|
||||
sessionKey: "agent:stale:other",
|
||||
},
|
||||
});
|
||||
|
||||
expect(emitSpy).toHaveBeenCalledWith({
|
||||
agentId: "main",
|
||||
messageId: "message-from-direct-publish",
|
||||
sessionFile: scope.sessionFile,
|
||||
sessionKey: "agent:main:main",
|
||||
});
|
||||
});
|
||||
|
||||
it("locks read and append helpers to one scoped transcript target", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "locked-target.jsonl"),
|
||||
sessionId: "locked-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
|
||||
const target = await withSessionTranscriptWriteLock(scope, async (locked) => {
|
||||
expect(await locked.readEvents()).toEqual([]);
|
||||
await locked.appendMessage({
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "locked" }],
|
||||
timestamp: 1,
|
||||
},
|
||||
});
|
||||
return locked.target;
|
||||
});
|
||||
|
||||
expect(target).toMatchObject({
|
||||
sessionId: "locked-session",
|
||||
targetKind: "active-session-file",
|
||||
});
|
||||
expect(target).not.toHaveProperty("sessionFile");
|
||||
await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([
|
||||
expect.objectContaining({ type: "session" }),
|
||||
expect.objectContaining({ message: expect.objectContaining({ role: "assistant" }) }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("uses the owned active transcript write context for scoped locked appends", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "owned-active-target.jsonl"),
|
||||
sessionId: "owned-active-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const storeDefaultFile = path.join(tempDir, "store-default-owned.jsonl");
|
||||
const lockEvents: string[] = [];
|
||||
|
||||
await upsertSessionEntry(scope, {
|
||||
sessionFile: storeDefaultFile,
|
||||
sessionId: scope.sessionId,
|
||||
updatedAt: 10,
|
||||
});
|
||||
|
||||
const target = await withOwnedSessionTranscriptWrites(
|
||||
{
|
||||
sessionFile: scope.sessionFile,
|
||||
sessionKey: scope.sessionKey,
|
||||
withSessionWriteLock: async (run) => {
|
||||
lockEvents.push("lock");
|
||||
return await run();
|
||||
},
|
||||
},
|
||||
async () =>
|
||||
await withSessionTranscriptWriteLock(scope, async (locked) => {
|
||||
await locked.appendMessage({
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "owned locked" }],
|
||||
timestamp: 1,
|
||||
},
|
||||
});
|
||||
return locked.target;
|
||||
}),
|
||||
);
|
||||
|
||||
expect(lockEvents).toEqual(["lock"]);
|
||||
expect(target).toMatchObject({
|
||||
sessionId: "owned-active-session",
|
||||
targetKind: "active-session-file",
|
||||
});
|
||||
expect(fs.readFileSync(scope.sessionFile, "utf8")).toContain("owned locked");
|
||||
expect(fs.existsSync(storeDefaultFile)).toBe(false);
|
||||
await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([
|
||||
expect.objectContaining({ type: "session" }),
|
||||
expect.objectContaining({ message: expect.objectContaining({ role: "assistant" }) }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("publishes queued locked updates after callback appends are visible", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "queued-publish-target.jsonl"),
|
||||
sessionId: "queued-publish-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const observedUpdates: Array<{
|
||||
callbackCompleted: boolean;
|
||||
fileText: string;
|
||||
update: unknown;
|
||||
}> = [];
|
||||
let callbackCompleted = false;
|
||||
const emitSpy = vi
|
||||
.spyOn(transcriptEvents, "emitSessionTranscriptUpdate")
|
||||
.mockImplementation((update) => {
|
||||
observedUpdates.push({
|
||||
callbackCompleted,
|
||||
fileText: fs.readFileSync(scope.sessionFile, "utf8"),
|
||||
update,
|
||||
});
|
||||
});
|
||||
|
||||
const result = await withSessionTranscriptWriteLock(scope, async (locked) => {
|
||||
await locked.appendMessage({
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "queued publish" }],
|
||||
timestamp: 1,
|
||||
},
|
||||
});
|
||||
await locked.publishUpdate({
|
||||
messageId: "message-from-callback",
|
||||
});
|
||||
expect(emitSpy).not.toHaveBeenCalled();
|
||||
callbackCompleted = true;
|
||||
return "complete";
|
||||
});
|
||||
|
||||
expect(result).toBe("complete");
|
||||
expect(emitSpy).toHaveBeenCalledTimes(1);
|
||||
expect(observedUpdates).toEqual([
|
||||
{
|
||||
callbackCompleted: true,
|
||||
fileText: expect.stringContaining("queued publish"),
|
||||
update: expect.objectContaining({
|
||||
messageId: "message-from-callback",
|
||||
agentId: "main",
|
||||
sessionFile: scope.sessionFile,
|
||||
sessionKey: scope.sessionKey,
|
||||
}),
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not publish queued locked updates when the callback throws", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionFile: path.join(tempDir, "failed-queued-publish-target.jsonl"),
|
||||
sessionId: "failed-queued-publish-session",
|
||||
sessionKey: "agent:main:main",
|
||||
storePath,
|
||||
};
|
||||
const emitSpy = vi.spyOn(transcriptEvents, "emitSessionTranscriptUpdate");
|
||||
|
||||
await expect(
|
||||
withSessionTranscriptWriteLock(scope, async (locked) => {
|
||||
await locked.appendMessage({
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "durable but failed" }],
|
||||
timestamp: 1,
|
||||
},
|
||||
});
|
||||
await locked.publishUpdate({ sessionKey: scope.sessionKey });
|
||||
throw new Error("stop before commit");
|
||||
}),
|
||||
).rejects.toThrow("stop before commit");
|
||||
expect(emitSpy).not.toHaveBeenCalled();
|
||||
expect(fs.readFileSync(scope.sessionFile, "utf8")).toContain("durable but failed");
|
||||
});
|
||||
|
||||
it("round-trips encoded memory hit keys with opaque session ids", () => {
|
||||
const key = formatSessionTranscriptMemoryHitKey({
|
||||
agentId: "SECONDARY",
|
||||
sessionId: "my-plugin:task/1",
|
||||
});
|
||||
|
||||
expect(key).toBe("transcript:secondary:my-plugin%3Atask%2F1");
|
||||
expect(parseSessionTranscriptMemoryHitKey(key)).toEqual({
|
||||
agentId: "secondary",
|
||||
key,
|
||||
sessionId: "my-plugin:task/1",
|
||||
});
|
||||
});
|
||||
|
||||
it("resolves memory hit keys by agent and session id instead of transcript basename", async () => {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionId: "session-id",
|
||||
sessionKey: "agent:main:telegram:direct:123",
|
||||
storePath,
|
||||
};
|
||||
await upsertSessionEntry(scope, {
|
||||
sessionFile: path.join(tempDir, "legacy-file-name.jsonl"),
|
||||
sessionId: scope.sessionId,
|
||||
updatedAt: 10,
|
||||
});
|
||||
|
||||
const keys = resolveSessionTranscriptMemoryHitKeyToSessionKeys({
|
||||
key: formatSessionTranscriptMemoryHitKey(scope),
|
||||
store: loadSessionStore(storePath),
|
||||
});
|
||||
|
||||
expect(keys).toEqual(["agent:main:telegram:direct:123"]);
|
||||
});
|
||||
|
||||
it("can avoid synthetic fallback keys for strict live-store checks", () => {
|
||||
const key = formatSessionTranscriptMemoryHitKey({
|
||||
agentId: "main",
|
||||
sessionId: "deleted-session",
|
||||
});
|
||||
|
||||
expect(resolveSessionTranscriptMemoryHitKeyToSessionKeys({ key, store: {} })).toEqual([
|
||||
"agent:main:deleted-session",
|
||||
]);
|
||||
expect(
|
||||
resolveSessionTranscriptMemoryHitKeyToSessionKeys({
|
||||
includeSyntheticFallback: false,
|
||||
key,
|
||||
store: {},
|
||||
}),
|
||||
).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,208 @@
|
||||
import {
|
||||
appendTranscriptMessage,
|
||||
publishTranscriptUpdate,
|
||||
resolveSessionTranscriptRuntimeReadTarget,
|
||||
resolveSessionTranscriptRuntimeTarget,
|
||||
type TranscriptMessageAppendOptions,
|
||||
type TranscriptMessageAppendResult,
|
||||
type TranscriptUpdatePayload,
|
||||
} from "../config/sessions/session-accessor.js";
|
||||
import { runSessionTranscriptAppendTransaction } from "../config/sessions/transcript-append.js";
|
||||
import { streamSessionTranscriptLines } from "../config/sessions/transcript-stream.js";
|
||||
import { normalizeAgentId } from "../routing/session-key.js";
|
||||
import {
|
||||
formatSessionTranscriptMemoryHitKey,
|
||||
parseSessionTranscriptMemoryHitKey,
|
||||
resolveSessionTranscriptMemoryHitKeyToSessionKeys,
|
||||
type ResolveSessionTranscriptMemoryHitKeyParams,
|
||||
type SessionTranscriptIdentity,
|
||||
type SessionTranscriptMemoryHitIdentity,
|
||||
type SessionTranscriptMemoryHitKey,
|
||||
type SessionTranscriptMemoryHitKeyParams,
|
||||
type SessionTranscriptReadParams,
|
||||
} from "./session-transcript-memory-hit.js";
|
||||
|
||||
export {
|
||||
formatSessionTranscriptMemoryHitKey,
|
||||
parseSessionTranscriptMemoryHitKey,
|
||||
resolveSessionTranscriptMemoryHitKeyToSessionKeys,
|
||||
};
|
||||
export type {
|
||||
ResolveSessionTranscriptMemoryHitKeyParams,
|
||||
SessionTranscriptIdentity,
|
||||
SessionTranscriptMemoryHitIdentity,
|
||||
SessionTranscriptMemoryHitKey,
|
||||
SessionTranscriptMemoryHitKeyParams,
|
||||
SessionTranscriptReadParams,
|
||||
};
|
||||
|
||||
export type SessionTranscriptEvent = unknown;
|
||||
|
||||
export type SessionTranscriptTargetParams = SessionTranscriptReadParams & {
|
||||
/**
|
||||
* @deprecated Prefer `{ agentId, sessionKey, sessionId }`. Pass this only
|
||||
* when adapting code that already receives an active transcript artifact and
|
||||
* needs each helper to operate on that same artifact.
|
||||
*/
|
||||
sessionFile?: string;
|
||||
};
|
||||
|
||||
export type SessionTranscriptTarget = SessionTranscriptIdentity & {
|
||||
targetKind: "active-session-file" | "runtime-session";
|
||||
};
|
||||
|
||||
export type SessionTranscriptAppendMessageParams<TMessage> = SessionTranscriptTargetParams &
|
||||
TranscriptMessageAppendOptions<TMessage>;
|
||||
|
||||
export type SessionTranscriptWriteLockParams = SessionTranscriptTargetParams & {
|
||||
config?: TranscriptMessageAppendOptions<unknown>["config"];
|
||||
};
|
||||
|
||||
export type SessionTranscriptWriteLockContext = {
|
||||
appendMessage: <TMessage>(
|
||||
options: Omit<TranscriptMessageAppendOptions<TMessage>, "config">,
|
||||
) => Promise<TranscriptMessageAppendResult<TMessage> | undefined>;
|
||||
publishUpdate: (update?: TranscriptUpdatePayload) => Promise<void>;
|
||||
readEvents: () => Promise<SessionTranscriptEvent[]>;
|
||||
target: SessionTranscriptTarget;
|
||||
};
|
||||
|
||||
/**
|
||||
* Resolves the public identity for a transcript without returning its file path.
|
||||
*/
|
||||
export async function resolveSessionTranscriptIdentity(
|
||||
params: SessionTranscriptReadParams,
|
||||
): Promise<SessionTranscriptIdentity> {
|
||||
const target = await resolveSessionTranscriptRuntimeReadTarget(params);
|
||||
const agentId = normalizeAgentId(target.agentId);
|
||||
return {
|
||||
agentId,
|
||||
memoryKey: formatSessionTranscriptMemoryHitKey({ agentId, sessionId: target.sessionId }),
|
||||
sessionId: target.sessionId,
|
||||
sessionKey: target.sessionKey,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves the public target for transcript operations without exposing the
|
||||
* current storage path as identity.
|
||||
*/
|
||||
export async function resolveSessionTranscriptTarget(
|
||||
params: SessionTranscriptTargetParams,
|
||||
): Promise<SessionTranscriptTarget> {
|
||||
const target = await resolveSessionTranscriptRuntimeReadTarget(params);
|
||||
return projectPublicTarget({
|
||||
...target,
|
||||
targetKind: params.sessionFile?.trim() ? "active-session-file" : "runtime-session",
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads transcript events by public session identity instead of file path.
|
||||
*/
|
||||
export async function readSessionTranscriptEvents(
|
||||
params: SessionTranscriptTargetParams,
|
||||
): Promise<SessionTranscriptEvent[]> {
|
||||
const target = await resolveSessionTranscriptRuntimeReadTarget(params);
|
||||
const events: SessionTranscriptEvent[] = [];
|
||||
for await (const line of streamSessionTranscriptLines(target.sessionFile)) {
|
||||
try {
|
||||
events.push(JSON.parse(line) as SessionTranscriptEvent);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
/**
|
||||
* Appends a transcript message by scoped transcript target.
|
||||
*/
|
||||
export async function appendSessionTranscriptMessageByIdentity<TMessage>(
|
||||
params: SessionTranscriptAppendMessageParams<TMessage>,
|
||||
): Promise<TranscriptMessageAppendResult<TMessage> | undefined> {
|
||||
return await appendTranscriptMessage(params, params);
|
||||
}
|
||||
|
||||
/**
|
||||
* Publishes a transcript update by scoped transcript target.
|
||||
*/
|
||||
export async function publishSessionTranscriptUpdateByIdentity(
|
||||
params: SessionTranscriptTargetParams & { update?: TranscriptUpdatePayload },
|
||||
): Promise<void> {
|
||||
const target = await resolveSessionTranscriptRuntimeTarget(params);
|
||||
await publishTranscriptUpdate(
|
||||
{
|
||||
...params,
|
||||
sessionFile: target.sessionFile,
|
||||
},
|
||||
{
|
||||
...params.update,
|
||||
agentId: target.agentId,
|
||||
sessionKey: target.sessionKey,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs transcript work under the write lock for the resolved scoped target.
|
||||
*/
|
||||
export async function withSessionTranscriptWriteLock<T>(
|
||||
params: SessionTranscriptWriteLockParams,
|
||||
run: (context: SessionTranscriptWriteLockContext) => Promise<T> | T,
|
||||
): Promise<T> {
|
||||
const storageTarget = await resolveSessionTranscriptRuntimeTarget(params);
|
||||
const target = projectPublicTarget({
|
||||
...storageTarget,
|
||||
targetKind: params.sessionFile?.trim() ? "active-session-file" : "runtime-session",
|
||||
});
|
||||
const boundScope = {
|
||||
...params,
|
||||
sessionFile: storageTarget.sessionFile,
|
||||
};
|
||||
// Treat publishUpdate as a post-commit callback: future transactional stores
|
||||
// must not expose updates when the scoped write callback fails.
|
||||
const queuedUpdates: Array<TranscriptUpdatePayload | undefined> = [];
|
||||
const result = await runSessionTranscriptAppendTransaction(
|
||||
{
|
||||
config: params.config,
|
||||
transcriptPath: storageTarget.sessionFile,
|
||||
},
|
||||
(transaction) =>
|
||||
run({
|
||||
target,
|
||||
readEvents: () => readSessionTranscriptEvents(boundScope),
|
||||
appendMessage: (options) =>
|
||||
transaction.appendMessage({
|
||||
...options,
|
||||
sessionId: params.sessionId,
|
||||
}),
|
||||
publishUpdate: async (update) => {
|
||||
queuedUpdates.push(update ? { ...update } : undefined);
|
||||
},
|
||||
}),
|
||||
);
|
||||
for (const update of queuedUpdates) {
|
||||
await publishSessionTranscriptUpdateByIdentity({
|
||||
...boundScope,
|
||||
update,
|
||||
});
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function projectPublicTarget(target: {
|
||||
agentId: string;
|
||||
sessionId: string;
|
||||
sessionKey: string;
|
||||
targetKind: SessionTranscriptTarget["targetKind"];
|
||||
}): SessionTranscriptTarget {
|
||||
const agentId = normalizeAgentId(target.agentId);
|
||||
return {
|
||||
agentId,
|
||||
memoryKey: formatSessionTranscriptMemoryHitKey({ agentId, sessionId: target.sessionId }),
|
||||
sessionId: target.sessionId,
|
||||
sessionKey: target.sessionKey,
|
||||
targetKind: target.targetKind,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user