mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-08 02:52:15 +00:00
feat(memory): backfill retained session transcripts (#115162)
* feat(memory): enumerate retained session transcripts * feat(memory): add session transcript backfill * test(memory): cover session backfill workflows * docs(memory): document session transcript backfill * fix(memory): preserve oldest session backfill ordering * fix(memory): harden session archive backfill * test(memory): prove owner-turn provenance gating * fix(memory): persist session backfill progress * fix(memory): preserve ranged backfill progress * fix(memory): order per-file backfill candidates * fix(memory): retain paged backfill diary entries * fix(memory): preserve rem backfill diary history * fix(memory): adapt backfill to retained session identity * fix(memory): keep rem backfill preview-only * fix(memory): satisfy backfill CI guards
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
4fb33559fb
commit
f0bf74bad4
@@ -1,5 +1,5 @@
|
||||
---
|
||||
summary: "CLI reference for `openclaw memory` (status/index/search/promote/promote-explain/rem-harness/rem-backfill)"
|
||||
summary: "CLI reference for `openclaw memory` (status/index/search/promote/promote-explain/rem-harness/rem-backfill/session-backfill)"
|
||||
read_when:
|
||||
- You want to index or search semantic memory
|
||||
- You're debugging memory availability or indexing
|
||||
@@ -139,6 +139,50 @@ openclaw memory rem-backfill --rollback [--rollback-short-term] [--json]
|
||||
- `--rollback-short-term`: remove previously staged grounded short-term
|
||||
candidates.
|
||||
|
||||
## `memory session-backfill`
|
||||
|
||||
Distill retained session history through the same provenance and short-term
|
||||
staging pipeline used by dreaming. The default is a read-only preview, ordered
|
||||
from the oldest unprocessed day to the newest.
|
||||
|
||||
```bash
|
||||
openclaw memory session-backfill --agent <id> [--from YYYY-MM-DD] [--to YYYY-MM-DD] \
|
||||
[--limit-days <n>] [--archive-files <path...>] [--rem | --apply] [--json]
|
||||
openclaw memory session-backfill --agent <id> --rollback [--json]
|
||||
```
|
||||
|
||||
| Flag | Default | Effect |
|
||||
| --------------------------- | ------------ | ------------------------------------------------------------------------------------------------------------- |
|
||||
| `--from YYYY-MM-DD` | | Include messages on or after this day in the dreaming timezone. |
|
||||
| `--to YYYY-MM-DD` | | Include messages on or before this day in the dreaming timezone. |
|
||||
| `--limit-days <n>` | `92` | Process at most this many hash-untracked days, oldest first. |
|
||||
| `--archive-files <path...>` | | Also inspect foreign transcript files as untrusted input; embedded owner metadata is not accepted. |
|
||||
| `--rem` | | Write deterministic grounded per-day previews to `DREAMS.md` only. |
|
||||
| `--apply` | preview only | Stage trusted candidates and write reversible `DREAMS.md` diary blocks. |
|
||||
| `--rollback` | | Remove all grounded backfill candidates and shared backfill diary blocks, including `rem-backfill` artifacts. |
|
||||
| `--json` | | Print machine-readable per-day counts and top candidates. |
|
||||
|
||||
The command reads the selected agent's canonical session store, including
|
||||
retained SQLite transcript identities from session rotation. It uses the same
|
||||
tracked message hashes and per-run caps as live session ingestion, so repeated
|
||||
`--apply` runs skip already ingested messages. Owner and agent lines from the
|
||||
canonical store are eligible; tool output, web or non-owner input, and turns
|
||||
without trustworthy owner provenance are excluded. Foreign archive files have
|
||||
no authenticated owner-provenance contract, so their embedded ownership fields
|
||||
remain untrusted and cannot be staged.
|
||||
|
||||
`--apply` writes only the session corpus under `memory/.dreams/`, short-term
|
||||
staging state, and reversible diary entries in `DREAMS.md`. It never writes
|
||||
`MEMORY.md` or `USER.md`; durable promotion remains a separate `memory promote`
|
||||
or dreaming decision. `--rem` and `--apply` are mutually exclusive.
|
||||
|
||||
Backfill rollback is intentionally shared with `memory rem-backfill`: both
|
||||
commands use the same grounded-only staging class and diary markers. Run
|
||||
`session-backfill --rollback` only when you intend to clear both commands'
|
||||
grounded backfill artifacts from that workspace. Rollback preserves transcript
|
||||
ingestion cursors and tracked message hashes, so removed messages are not
|
||||
automatically re-ingested.
|
||||
|
||||
## Dreaming
|
||||
|
||||
Dreaming is the background memory consolidation system with three cooperative
|
||||
|
||||
@@ -112,10 +112,25 @@ There is also a grounded historical backfill lane for review and recovery work:
|
||||
- `memory rem-backfill --path ...` writes reversible grounded diary entries into `DREAMS.md`.
|
||||
- `memory rem-backfill --path ... --stage-short-term` stages grounded durable candidates into the same short-term evidence store the normal deep phase uses.
|
||||
- `memory rem-backfill --rollback` and `--rollback-short-term` remove those staged backfill artifacts without touching ordinary diary entries or live short-term recall.
|
||||
- `memory session-backfill --agent <id>` previews trusted candidates from the agent's retained session history, oldest unprocessed day first.
|
||||
- `memory session-backfill --agent <id> --apply` stages those candidates through the normal short-term store and writes reversible diary blocks without changing `MEMORY.md` or `USER.md`.
|
||||
- `memory session-backfill --agent <id> --rem` writes a deterministic grounded preview per day to `DREAMS.md` without staging candidates or calling a model.
|
||||
- `memory session-backfill --agent <id> --rollback` clears the shared grounded backfill candidates and diary blocks, including artifacts created by `rem-backfill`.
|
||||
|
||||
</Accordion>
|
||||
</AccordionGroup>
|
||||
|
||||
Session backfill uses canonical retained transcript identities, including
|
||||
sessions preserved across rotation. Messages are bucketed in the configured
|
||||
dreaming timezone and share live ingestion's tracked message hashes and signal
|
||||
caps, so bounded reruns continue forward without re-ingesting prior messages.
|
||||
Rollback removes generated artifacts but retains those ingestion checkpoints.
|
||||
Foreign files supplied with `--archive-files` are treated conservatively. Their
|
||||
embedded ownership fields are caller-controlled and therefore remain untrusted;
|
||||
without an authenticated provenance contract, they cannot enter short-term
|
||||
staging. Tool output, web content, and non-owner turns are excluded from the
|
||||
canonical session path as well.
|
||||
|
||||
The Control UI exposes the same diary backfill/reset flow on the agent's Memory tab (Agents page) so you can inspect results in the dream scene before deciding whether grounded candidates deserve promotion. A distinct grounded Scene lane shows which staged short-term entries came from historical replay, which promoted items were grounded-led, and lets you clear only grounded-only staged entries without touching live short-term state.
|
||||
|
||||
## Deep ranking signals
|
||||
|
||||
@@ -1757,6 +1757,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`.
|
||||
- H2: memory promote-explain
|
||||
- H2: memory rem-harness
|
||||
- H2: memory rem-backfill
|
||||
- H2: memory session-backfill
|
||||
- H2: Dreaming
|
||||
- H2: SecretRef gateway dependency
|
||||
- H2: Related
|
||||
|
||||
@@ -16,11 +16,103 @@ import { seedHistoricalDailyMemorySignals } from "./dreaming-phases.js";
|
||||
import type { MemoryCoreRuntimeHost } from "./memory/runtime-host.js";
|
||||
import { previewGroundedRemMarkdown } from "./rem-evidence.js";
|
||||
import { previewRemHarness } from "./rem-harness.js";
|
||||
import { runSessionBackfill, type MemorySessionBackfillOptions } from "./session-backfill.js";
|
||||
import {
|
||||
recordGroundedShortTermCandidates,
|
||||
removeGroundedShortTermCandidates,
|
||||
} from "./short-term-promotion.js";
|
||||
const { heading, muted, warn } = theme;
|
||||
|
||||
export async function runMemorySessionBackfill(
|
||||
opts: MemorySessionBackfillOptions,
|
||||
hostOptions?: MemoryCoreRuntimeHost,
|
||||
) {
|
||||
const { config: cfg, diagnostics } = await loadMemoryCommandConfig("memory session-backfill");
|
||||
emitMemorySecretResolveDiagnostics(diagnostics, { json: Boolean(opts.json) });
|
||||
const agentId = resolveAgent(cfg, opts.agent);
|
||||
await withMemoryManagerForAgent({
|
||||
cfg,
|
||||
agentId,
|
||||
purpose: "status",
|
||||
acquireLocalService: hostOptions?.acquireLocalService,
|
||||
withLease: hostOptions?.withLease,
|
||||
run: async (manager) => {
|
||||
const workspaceDir = manager.status().workspaceDir?.trim();
|
||||
if (!workspaceDir) {
|
||||
defaultRuntime.error("Memory session-backfill requires a resolvable workspace directory.");
|
||||
process.exitCode = 1;
|
||||
return;
|
||||
}
|
||||
if (
|
||||
opts.rollback &&
|
||||
(opts.apply || opts.rem || opts.from || opts.to || opts.archiveFiles?.length)
|
||||
) {
|
||||
defaultRuntime.error(
|
||||
"Memory session-backfill --rollback cannot be combined with input, range, --rem, or --apply options.",
|
||||
);
|
||||
process.exitCode = 1;
|
||||
return;
|
||||
}
|
||||
const remConfig = resolveMemoryRemDreamingConfig({
|
||||
pluginConfig: resolveMemoryPluginConfig(cfg),
|
||||
cfg,
|
||||
});
|
||||
let result;
|
||||
try {
|
||||
result = await runSessionBackfill({
|
||||
agentId,
|
||||
workspaceDir,
|
||||
...(opts.from !== undefined ? { from: opts.from } : {}),
|
||||
...(opts.to !== undefined ? { to: opts.to } : {}),
|
||||
...(opts.limitDays !== undefined ? { limitDays: opts.limitDays } : {}),
|
||||
...(opts.rem !== undefined ? { rem: opts.rem } : {}),
|
||||
...(opts.apply !== undefined ? { apply: opts.apply } : {}),
|
||||
...(opts.rollback !== undefined ? { rollback: opts.rollback } : {}),
|
||||
...(opts.archiveFiles !== undefined ? { archiveFiles: opts.archiveFiles } : {}),
|
||||
...(remConfig.timezone !== undefined ? { timezone: remConfig.timezone } : {}),
|
||||
});
|
||||
} catch (error) {
|
||||
defaultRuntime.error(error instanceof Error ? error.message : String(error));
|
||||
process.exitCode = 1;
|
||||
return;
|
||||
}
|
||||
if (opts.json) {
|
||||
defaultRuntime.writeJson(result);
|
||||
return;
|
||||
}
|
||||
if (result.rollback) {
|
||||
defaultRuntime.log(
|
||||
[
|
||||
`${heading("Session Backfill")} ${muted("(rollback)")}`,
|
||||
muted(`workspace=${shortenHomePath(workspaceDir)}`),
|
||||
muted(`removedDiaryEntries=${result.rollback.removedDiaryEntries}`),
|
||||
muted(`removedStagedEntries=${result.rollback.removedStagedEntries}`),
|
||||
].join("\n"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
const lines = [
|
||||
`${heading("Session Backfill")} ${muted(`(${agentId})`)}`,
|
||||
muted(`workspace=${shortenHomePath(workspaceDir)}`),
|
||||
muted(
|
||||
`days=${result.days.length} candidates=${result.candidateCount} staged=${result.stagedEntries}`,
|
||||
),
|
||||
];
|
||||
for (const day of result.days) {
|
||||
lines.push("", heading(day.day), muted(`candidates=${day.candidateCount}`));
|
||||
lines.push(...day.topCandidates.map((candidate) => `- ${candidate}`));
|
||||
}
|
||||
if (result.days.length === 0) {
|
||||
lines.push("", "No new hash-untracked trusted session candidates.");
|
||||
}
|
||||
if (!result.applied && !result.rem) {
|
||||
lines.push("", muted("Dry run; use --apply to stage candidates."));
|
||||
}
|
||||
defaultRuntime.log(lines.join("\n"));
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export async function runMemoryRemHarness(
|
||||
opts: MemoryRemHarnessOptions,
|
||||
hostOptions?: MemoryCoreRuntimeHost,
|
||||
|
||||
@@ -4,5 +4,9 @@ export {
|
||||
runMemoryPromoteExplain,
|
||||
runMemorySearch,
|
||||
} from "./cli-index-search.runtime.js";
|
||||
export { runMemoryRemBackfill, runMemoryRemHarness } from "./cli-rem.runtime.js";
|
||||
export {
|
||||
runMemoryRemBackfill,
|
||||
runMemoryRemHarness,
|
||||
runMemorySessionBackfill,
|
||||
} from "./cli-rem.runtime.js";
|
||||
export { runMemoryStatus } from "./cli-status.runtime.js";
|
||||
|
||||
@@ -19,6 +19,7 @@ import type {
|
||||
MemorySearchCommandOptions,
|
||||
} from "./cli.types.js";
|
||||
import type { MemoryCoreRuntimeHost } from "./memory/runtime-host.js";
|
||||
import type { MemorySessionBackfillOptions } from "./session-backfill.js";
|
||||
import {
|
||||
DEFAULT_PROMOTION_MIN_RECALL_COUNT,
|
||||
DEFAULT_PROMOTION_MIN_SCORE,
|
||||
@@ -28,6 +29,7 @@ import {
|
||||
const loadMemoryCliRuntime = createLazyRuntimeModule(() => import("./cli.runtime.js"));
|
||||
|
||||
const DECIMAL_NUMBER_RE = /^[+-]?(?:\d+(?:\.\d+)?|\.\d+)$/;
|
||||
const DEFAULT_SESSION_BACKFILL_LIMIT_DAYS = 92;
|
||||
|
||||
async function runMemoryStatus(opts: MemoryCommandOptions, hostOptions?: MemoryCoreRuntimeHost) {
|
||||
const runtime = await loadMemoryCliRuntime();
|
||||
@@ -81,6 +83,14 @@ async function runMemoryRemBackfill(
|
||||
await runtime.runMemoryRemBackfill(opts, hostOptions);
|
||||
}
|
||||
|
||||
async function runMemorySessionBackfill(
|
||||
opts: MemorySessionBackfillOptions,
|
||||
hostOptions?: MemoryCoreRuntimeHost,
|
||||
) {
|
||||
const runtime = await loadMemoryCliRuntime();
|
||||
await runtime.runMemorySessionBackfill(opts, hostOptions);
|
||||
}
|
||||
|
||||
function invalidCliArgument(message: string): Error & { code: string; exitCode: number } {
|
||||
const error = new Error(message) as Error & { code: string; exitCode: number };
|
||||
error.name = "InvalidArgumentError";
|
||||
@@ -159,6 +169,10 @@ export function registerMemoryCli(program: Command, hostOptions?: MemoryCoreRunt
|
||||
"openclaw memory rem-backfill --path ./memory --stage-short-term",
|
||||
"Also seed durable grounded candidates into the live short-term promotion store.",
|
||||
],
|
||||
[
|
||||
"openclaw memory session-backfill --agent main --from 2026-01-01",
|
||||
"Preview trusted candidates from retained session history.",
|
||||
],
|
||||
["openclaw memory status --json", "Output machine-readable JSON (good for scripts)."],
|
||||
])}\n\n${theme.muted("Docs:")} ${formatDocsLink("/cli/memory", "docs.openclaw.ai/cli/memory")}\n`,
|
||||
);
|
||||
@@ -276,6 +290,34 @@ export function registerMemoryCli(program: Command, hostOptions?: MemoryCoreRunt
|
||||
await runMemoryRemBackfill(opts, hostOptions);
|
||||
});
|
||||
|
||||
memory
|
||||
.command("session-backfill")
|
||||
.description("Distill retained session history into staged memory candidates")
|
||||
.option("--agent <id>", "Agent id (default: default agent)")
|
||||
.option("--from <YYYY-MM-DD>", "Oldest transcript day to include")
|
||||
.option("--to <YYYY-MM-DD>", "Newest transcript day to include")
|
||||
.option(
|
||||
"--limit-days <n>",
|
||||
`Maximum unprocessed days (default: ${DEFAULT_SESSION_BACKFILL_LIMIT_DAYS})`,
|
||||
(value: string) => parseMemoryCliPositiveIntegerOption(value, "--limit-days"),
|
||||
DEFAULT_SESSION_BACKFILL_LIMIT_DAYS,
|
||||
)
|
||||
.option("--rem", "Write grounded per-day REM previews to DREAMS.md", false)
|
||||
.option("--apply", "Stage candidates and write DREAMS.md diary entries", false)
|
||||
.option(
|
||||
"--rollback",
|
||||
"Remove all grounded backfill candidates and shared backfill diary entries",
|
||||
false,
|
||||
)
|
||||
.option(
|
||||
"--archive-files <path...>",
|
||||
"Also inspect foreign transcript archive files conservatively",
|
||||
)
|
||||
.option("--json", "Print JSON")
|
||||
.action(async (opts: MemorySessionBackfillOptions) => {
|
||||
await runMemorySessionBackfill(opts, hostOptions);
|
||||
});
|
||||
|
||||
memory.action(() => {
|
||||
memory.outputHelp();
|
||||
process.exitCode = 0;
|
||||
|
||||
@@ -635,12 +635,15 @@ export async function writeBackfillDiaryEntries(params: {
|
||||
bodyLines: string[];
|
||||
sourcePath?: string;
|
||||
}>;
|
||||
preserveExisting?: boolean;
|
||||
timezone?: string;
|
||||
}): Promise<{ dreamsPath: string; written: number; replaced: number }> {
|
||||
return await updateDreamsFile({
|
||||
workspaceDir: params.workspaceDir,
|
||||
updater: (existing, dreamsPath) => {
|
||||
const stripped = stripBackfillDiaryBlocks(existing);
|
||||
const stripped = params.preserveExisting
|
||||
? { updated: existing, removed: 0 }
|
||||
: stripBackfillDiaryBlocks(existing);
|
||||
const startIdx = stripped.updated.indexOf(DIARY_START_MARKER);
|
||||
const endIdx = stripped.updated.indexOf(DIARY_END_MARKER);
|
||||
const inner =
|
||||
@@ -648,22 +651,33 @@ export async function writeBackfillDiaryEntries(params: {
|
||||
? stripped.updated.slice(startIdx + DIARY_START_MARKER.length, endIdx)
|
||||
: "";
|
||||
const preservedBlocks = splitDiaryBlocks(inner);
|
||||
const nextBlocks = [
|
||||
...preservedBlocks,
|
||||
...params.entries.map((entry) =>
|
||||
buildBackfillDiaryEntry({
|
||||
isoDay: entry.isoDay,
|
||||
bodyLines: entry.bodyLines,
|
||||
sourcePath: entry.sourcePath,
|
||||
timezone: params.timezone,
|
||||
}),
|
||||
),
|
||||
];
|
||||
const additions = params.entries.map((entry) =>
|
||||
buildBackfillDiaryEntry({
|
||||
isoDay: entry.isoDay,
|
||||
bodyLines: entry.bodyLines,
|
||||
sourcePath: entry.sourcePath,
|
||||
timezone: params.timezone,
|
||||
}),
|
||||
);
|
||||
const existingFingerprints = new Set(
|
||||
preservedBlocks.map((block) => normalizeDiaryBlockFingerprint(block)),
|
||||
);
|
||||
const appended = params.preserveExisting
|
||||
? additions.filter((block) => {
|
||||
const fingerprint = normalizeDiaryBlockFingerprint(block);
|
||||
if (existingFingerprints.has(fingerprint)) {
|
||||
return false;
|
||||
}
|
||||
existingFingerprints.add(fingerprint);
|
||||
return true;
|
||||
})
|
||||
: additions;
|
||||
const nextBlocks = [...preservedBlocks, ...appended];
|
||||
return {
|
||||
content: replaceDiaryContent(stripped.updated, joinDiaryBlocks(nextBlocks)),
|
||||
result: {
|
||||
dreamsPath,
|
||||
written: params.entries.length,
|
||||
written: appended.length,
|
||||
replaced: stripped.removed,
|
||||
},
|
||||
};
|
||||
|
||||
@@ -2051,4 +2051,26 @@ export async function runDreamingSweepPhases(params: {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Session backfill is a batch driver over the live-ingestion primitives. Keep
|
||||
// these exports narrow so both paths share caps, hashing, state, and rendering.
|
||||
export {
|
||||
SESSION_INGESTION_MAX_MESSAGES_PER_FILE,
|
||||
SESSION_INGESTION_MAX_MESSAGES_PER_SWEEP,
|
||||
SESSION_INGESTION_MIN_MESSAGES_PER_FILE,
|
||||
SESSION_INGESTION_MIN_SNIPPET_CHARS,
|
||||
SESSION_INGESTION_SCORE,
|
||||
appendSessionCorpusLines,
|
||||
buildSessionFileScopeKey,
|
||||
buildSessionRenderedLine,
|
||||
buildSqliteDreamingSessionPath,
|
||||
buildSessionStateKey,
|
||||
hashSessionMessageId,
|
||||
mergeTrackedMessageHashes,
|
||||
normalizeSessionCorpusSnippet,
|
||||
readSessionIngestionState,
|
||||
trimTrackedSessionScopes,
|
||||
writeSessionIngestionState,
|
||||
};
|
||||
export type { SessionIngestionMessage };
|
||||
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */
|
||||
|
||||
470
extensions/memory-core/src/session-backfill.test.ts
Normal file
470
extensions/memory-core/src/session-backfill.test.ts
Normal file
@@ -0,0 +1,470 @@
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { resolveSessionTranscriptsDirForAgent } from "openclaw/plugin-sdk/memory-core-host-runtime-core";
|
||||
import {
|
||||
clearConfigCache,
|
||||
clearRuntimeConfigSnapshot,
|
||||
} from "openclaw/plugin-sdk/runtime-config-snapshot";
|
||||
import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime";
|
||||
import { appendSessionTranscriptMessageByIdentity } from "openclaw/plugin-sdk/session-transcript-runtime";
|
||||
import { formatSqliteSessionFileMarker } from "openclaw/plugin-sdk/sqlite-runtime-testing";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { writeBackfillDiaryEntries } from "./dreaming-narrative.js";
|
||||
import { runSessionBackfill } from "./session-backfill.js";
|
||||
import { readShortTermRecallEntries } from "./short-term-promotion.js";
|
||||
import { createMemoryCoreTestHarness } from "./test-helpers.js";
|
||||
|
||||
const harness = createMemoryCoreTestHarness();
|
||||
|
||||
type TranscriptMessage = {
|
||||
role: "assistant" | "tool" | "user";
|
||||
content: string;
|
||||
timestamp: string;
|
||||
owner?: boolean;
|
||||
};
|
||||
|
||||
async function writeTranscript(filePath: string, messages: TranscriptMessage[]): Promise<void> {
|
||||
await fs.mkdir(path.dirname(filePath), { recursive: true });
|
||||
const records = messages.map((message, index) => ({
|
||||
type: "message",
|
||||
id: `message-${index}`,
|
||||
timestamp: message.timestamp,
|
||||
message: {
|
||||
role: message.role,
|
||||
content: message.content,
|
||||
timestamp: message.timestamp,
|
||||
...(message.owner ? { __openclaw: { senderIsOwner: true } } : {}),
|
||||
},
|
||||
}));
|
||||
await fs.writeFile(filePath, `${records.map((record) => JSON.stringify(record)).join("\n")}\n`);
|
||||
}
|
||||
|
||||
async function seedCanonicalTranscript(
|
||||
sessionId: string,
|
||||
messages: TranscriptMessage[],
|
||||
): Promise<void> {
|
||||
const agentId = "main";
|
||||
const sessionsDir = resolveSessionTranscriptsDirForAgent(agentId);
|
||||
const storePath = path.join(sessionsDir, "sessions.json");
|
||||
const sessionKey = `agent:${agentId}:session-backfill:${sessionId}`;
|
||||
const updatedAt = Math.max(
|
||||
Date.now(),
|
||||
...messages.map((message) => Date.parse(message.timestamp)),
|
||||
);
|
||||
await fs.mkdir(sessionsDir, { recursive: true });
|
||||
const sessionFile = formatSqliteSessionFileMarker({ agentId, sessionId, storePath });
|
||||
const entry = { sessionFile, sessionId, updatedAt };
|
||||
await upsertSessionEntry({ agentId, sessionKey, storePath, entry });
|
||||
for (const message of messages) {
|
||||
await appendSessionTranscriptMessageByIdentity({
|
||||
agentId,
|
||||
sessionId,
|
||||
sessionKey,
|
||||
storePath,
|
||||
message: {
|
||||
role: message.role,
|
||||
content: message.content,
|
||||
timestamp: message.timestamp,
|
||||
...(message.owner ? { __openclaw: { senderIsOwner: true } } : {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
await upsertSessionEntry({ agentId, sessionKey, storePath, entry });
|
||||
}
|
||||
|
||||
async function createIsolatedWorkspace(prefix: string): Promise<string> {
|
||||
const workspaceDir = await harness.createTempWorkspace(prefix);
|
||||
vi.stubEnv("OPENCLAW_STATE_DIR", path.join(workspaceDir, "state"));
|
||||
vi.stubEnv("OPENCLAW_CONFIG_PATH", path.join(workspaceDir, "openclaw.json"));
|
||||
clearRuntimeConfigSnapshot();
|
||||
clearConfigCache();
|
||||
return workspaceDir;
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllEnvs();
|
||||
clearRuntimeConfigSnapshot();
|
||||
clearConfigCache();
|
||||
});
|
||||
|
||||
describe("runSessionBackfill", () => {
|
||||
it("keeps REM preview mode mutually exclusive with apply", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("rem-apply-");
|
||||
|
||||
await expect(
|
||||
runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
rem: true,
|
||||
apply: true,
|
||||
}),
|
||||
).rejects.toThrow("Memory session-backfill --rem cannot be combined with --apply.");
|
||||
});
|
||||
|
||||
it("buckets messages in the configured timezone and processes days oldest first", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("timezone-");
|
||||
await seedCanonicalTranscript("timezone", [
|
||||
{
|
||||
role: "user",
|
||||
content: "Late New York note",
|
||||
timestamp: "2026-01-02T00:30:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: "Early New York note",
|
||||
timestamp: "2026-01-02T05:30:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
]);
|
||||
|
||||
const result = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
timezone: "America/New_York",
|
||||
});
|
||||
|
||||
expect(result.days.map((day) => [day.day, day.candidateCount])).toEqual([
|
||||
["2026-01-01", 1],
|
||||
["2026-01-02", 1],
|
||||
]);
|
||||
});
|
||||
|
||||
it("honors the day limit before moving to newer unprocessed days", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("limit-");
|
||||
await seedCanonicalTranscript(
|
||||
"limited",
|
||||
["2026-01-01", "2026-01-02", "2026-01-03"].map((day) => ({
|
||||
role: "user" as const,
|
||||
content: `Durable note for ${day}`,
|
||||
timestamp: `${day}T12:00:00.000Z`,
|
||||
owner: true,
|
||||
})),
|
||||
);
|
||||
|
||||
const result = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
limitDays: 2,
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
expect(result.days.map((day) => day.day)).toEqual(["2026-01-01", "2026-01-02"]);
|
||||
});
|
||||
|
||||
it("does not advance the cursor past messages excluded by a date range", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("range-cursor-");
|
||||
await seedCanonicalTranscript("range-cursor", [
|
||||
{
|
||||
role: "user",
|
||||
content: "January durable note",
|
||||
timestamp: "2026-01-15T12:00:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: "February durable note",
|
||||
timestamp: "2026-02-15T12:00:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
]);
|
||||
|
||||
const january = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
apply: true,
|
||||
to: "2026-01-31",
|
||||
timezone: "UTC",
|
||||
});
|
||||
const february = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
apply: true,
|
||||
from: "2026-02-01",
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
expect(january.days.map((day) => day.day)).toEqual(["2026-01-15"]);
|
||||
expect(february.days.map((day) => day.day)).toEqual(["2026-02-15"]);
|
||||
});
|
||||
|
||||
it("advances the source cursor beyond the per-file signal cap", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("cursor-");
|
||||
await seedCanonicalTranscript(
|
||||
"cursor",
|
||||
Array.from({ length: 100 }, (_, index) => ({
|
||||
role: "user" as const,
|
||||
content: `Durable cursor note ${index}`,
|
||||
timestamp: new Date(Date.parse("2026-01-01T00:00:00.000Z") + index * 60_000).toISOString(),
|
||||
owner: true,
|
||||
})),
|
||||
);
|
||||
|
||||
const run = () =>
|
||||
runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
apply: true,
|
||||
nowMs: Date.parse("2026-01-02T12:00:00.000Z"),
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
expect((await run()).candidateCount).toBe(80);
|
||||
expect((await run()).candidateCount).toBe(20);
|
||||
expect((await run()).candidateCount).toBe(0);
|
||||
const dreams = await fs.readFile(path.join(workspaceDir, "DREAMS.md"), "utf-8");
|
||||
expect(dreams.match(/openclaw:dreaming:backfill-entry/g)).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("applies the total cap after finding the oldest candidate across sources", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("oldest-cap-");
|
||||
for (let sourceIndex = 0; sourceIndex < 16; sourceIndex += 1) {
|
||||
await seedCanonicalTranscript(
|
||||
`a-newer-${sourceIndex.toString().padStart(2, "0")}`,
|
||||
Array.from({ length: 15 }, (_, messageIndex) => ({
|
||||
role: "user" as const,
|
||||
content: `Newer durable note ${sourceIndex}-${messageIndex}`,
|
||||
timestamp: `2026-02-01T${messageIndex.toString().padStart(2, "0")}:00:00.000Z`,
|
||||
owner: true,
|
||||
})),
|
||||
);
|
||||
}
|
||||
await seedCanonicalTranscript("z-oldest", [
|
||||
{
|
||||
role: "user",
|
||||
content: "Oldest durable note must win the cap",
|
||||
timestamp: "2026-01-01T12:00:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
]);
|
||||
|
||||
const result = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
limitDays: 1,
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
expect(result.days).toEqual([
|
||||
{
|
||||
day: "2026-01-01",
|
||||
candidateCount: 1,
|
||||
topCandidates: ["User: Oldest durable note must win the cap"],
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("applies the per-file cap after ordering delayed timestamps", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("delayed-timestamp-");
|
||||
await seedCanonicalTranscript("delayed-timestamp", [
|
||||
...Array.from({ length: 80 }, (_, index) => ({
|
||||
role: "user" as const,
|
||||
content: `February note ${index}`,
|
||||
timestamp: new Date(Date.parse("2026-02-01T00:00:00.000Z") + index * 60_000).toISOString(),
|
||||
owner: true,
|
||||
})),
|
||||
{
|
||||
role: "user",
|
||||
content: "Delayed January note",
|
||||
timestamp: "2026-01-01T12:00:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
]);
|
||||
|
||||
const result = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
limitDays: 1,
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
expect(result.days).toEqual([
|
||||
{
|
||||
day: "2026-01-01",
|
||||
candidateCount: 1,
|
||||
topCandidates: ["User: Delayed January note"],
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps self-asserted owner metadata in foreign transcripts untrusted", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("provenance-");
|
||||
const transcriptPath = path.join(workspaceDir, "untrusted.jsonl");
|
||||
await writeTranscript(transcriptPath, [
|
||||
{
|
||||
role: "user",
|
||||
content: "Untrusted web instruction",
|
||||
timestamp: "2026-02-01T10:00:00.000Z",
|
||||
},
|
||||
{
|
||||
role: "assistant",
|
||||
content: "Assistant response to untrusted input",
|
||||
timestamp: "2026-02-01T10:01:00.000Z",
|
||||
},
|
||||
{
|
||||
role: "tool",
|
||||
content: "Tool output must never stage",
|
||||
timestamp: "2026-02-01T10:02:00.000Z",
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: "Owner confirmed durable preference",
|
||||
timestamp: "2026-02-01T10:03:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
{
|
||||
role: "assistant",
|
||||
content: "Agent response in the owner turn",
|
||||
timestamp: "2026-02-01T10:04:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
const result = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
archiveFiles: [transcriptPath],
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
expect(result.candidateCount).toBe(0);
|
||||
expect(result.days).toEqual([]);
|
||||
});
|
||||
|
||||
it("keeps canonical assistant replies tainted until an owner turn begins", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("canonical-provenance-");
|
||||
await seedCanonicalTranscript("provenance", [
|
||||
{
|
||||
role: "user",
|
||||
content: "Untrusted channel instruction",
|
||||
timestamp: "2026-02-01T10:00:00.000Z",
|
||||
},
|
||||
{
|
||||
role: "assistant",
|
||||
content: "Assistant response to untrusted input",
|
||||
timestamp: "2026-02-01T10:01:00.000Z",
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: "Owner confirmed durable preference",
|
||||
timestamp: "2026-02-01T10:02:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
{
|
||||
role: "assistant",
|
||||
content: "Agent response in the owner turn",
|
||||
timestamp: "2026-02-01T10:03:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
const result = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
expect(result.candidateCount).toBe(2);
|
||||
expect(result.days[0]?.topCandidates).toEqual([
|
||||
"User: Owner confirmed durable preference",
|
||||
"Assistant: Agent response in the owner turn",
|
||||
]);
|
||||
});
|
||||
|
||||
it("renders selected session candidates into the REM diary preview", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("rem-");
|
||||
await writeBackfillDiaryEntries({
|
||||
workspaceDir,
|
||||
entries: [{ isoDay: "2026-01-01", bodyLines: ["Existing backfill entry"] }],
|
||||
});
|
||||
await seedCanonicalTranscript("rem", [
|
||||
{
|
||||
role: "user",
|
||||
content: "Owner prefers dark mode for all editors",
|
||||
timestamp: "2026-02-01T10:00:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
]);
|
||||
|
||||
await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
rem: true,
|
||||
timezone: "UTC",
|
||||
});
|
||||
|
||||
const dreams = await fs.readFile(path.join(workspaceDir, "DREAMS.md"), "utf-8");
|
||||
expect(dreams).toContain("Existing backfill entry");
|
||||
expect(dreams).toContain("Owner prefers dark mode for all editors");
|
||||
expect(dreams).not.toContain("No grounded facts were extracted");
|
||||
expect(dreams.match(/openclaw:dreaming:backfill-entry/g)).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("stages idempotently, converges duplicate facts, and rolls back staged artifacts", async () => {
|
||||
const workspaceDir = await createIsolatedWorkspace("apply-");
|
||||
await seedCanonicalTranscript("repeat", [
|
||||
{
|
||||
role: "user",
|
||||
content: "The preferred editor is Nova",
|
||||
timestamp: "2026-03-01T10:00:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: "The preferred editor is Nova",
|
||||
timestamp: "2026-03-01T11:00:00.000Z",
|
||||
owner: true,
|
||||
},
|
||||
]);
|
||||
|
||||
const first = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
apply: true,
|
||||
nowMs: Date.parse("2026-03-02T12:00:00.000Z"),
|
||||
timezone: "UTC",
|
||||
});
|
||||
const afterFirst = await readShortTermRecallEntries({ workspaceDir });
|
||||
|
||||
// Count-level proof stays stable across the sibling claim-key implementation.
|
||||
expect(first.stagedEntries).toBe(1);
|
||||
expect(afterFirst).toHaveLength(1);
|
||||
expect(afterFirst[0]?.snippet).toBe("The preferred editor is Nova");
|
||||
|
||||
const second = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
apply: true,
|
||||
nowMs: Date.parse("2026-03-02T12:00:00.000Z"),
|
||||
timezone: "UTC",
|
||||
});
|
||||
expect(second.candidateCount).toBe(0);
|
||||
expect(second.stagedEntries).toBe(0);
|
||||
expect(await readShortTermRecallEntries({ workspaceDir })).toHaveLength(1);
|
||||
|
||||
const dreamsPath = path.join(workspaceDir, "DREAMS.md");
|
||||
expect(await fs.readFile(dreamsPath, "utf-8")).toContain("openclaw:dreaming:backfill-entry");
|
||||
|
||||
const rollback = await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
rollback: true,
|
||||
});
|
||||
expect(rollback.rollback).toEqual({
|
||||
removedDiaryEntries: 1,
|
||||
removedStagedEntries: 1,
|
||||
});
|
||||
expect(await readShortTermRecallEntries({ workspaceDir })).toHaveLength(0);
|
||||
expect(await fs.readFile(dreamsPath, "utf-8")).not.toContain(
|
||||
"openclaw:dreaming:backfill-entry",
|
||||
);
|
||||
expect(
|
||||
(
|
||||
await runSessionBackfill({
|
||||
agentId: "main",
|
||||
workspaceDir,
|
||||
apply: true,
|
||||
timezone: "UTC",
|
||||
})
|
||||
).candidateCount,
|
||||
).toBe(0);
|
||||
});
|
||||
});
|
||||
680
extensions/memory-core/src/session-backfill.ts
Normal file
680
extensions/memory-core/src/session-backfill.ts
Normal file
@@ -0,0 +1,680 @@
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import {
|
||||
buildSessionEntry,
|
||||
listSessionTranscriptCorpusEntriesForAgent,
|
||||
sessionPathForFile,
|
||||
type SessionTranscriptCorpusEntry,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-qmd";
|
||||
import type { MemorySearchResult } from "openclaw/plugin-sdk/memory-core-host-runtime-files";
|
||||
import { formatMemoryDreamingDay } from "openclaw/plugin-sdk/memory-core-host-status";
|
||||
import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path";
|
||||
import type { SessionIngestionFileState } from "./dreaming-ingestion-state.js";
|
||||
import { removeBackfillDiaryEntries, writeBackfillDiaryEntries } from "./dreaming-narrative.js";
|
||||
import {
|
||||
SESSION_INGESTION_MAX_MESSAGES_PER_FILE,
|
||||
SESSION_INGESTION_MAX_MESSAGES_PER_SWEEP,
|
||||
SESSION_INGESTION_MIN_MESSAGES_PER_FILE,
|
||||
SESSION_INGESTION_MIN_SNIPPET_CHARS,
|
||||
SESSION_INGESTION_SCORE,
|
||||
appendSessionCorpusLines,
|
||||
buildSessionFileScopeKey,
|
||||
buildSessionRenderedLine,
|
||||
buildSqliteDreamingSessionPath,
|
||||
buildSessionStateKey,
|
||||
hashSessionMessageId,
|
||||
mergeTrackedMessageHashes,
|
||||
normalizeSessionCorpusSnippet,
|
||||
readSessionIngestionState,
|
||||
trimTrackedSessionScopes,
|
||||
writeSessionIngestionState,
|
||||
type SessionIngestionMessage,
|
||||
} from "./dreaming-phases.js";
|
||||
import { previewGroundedRemMarkdown } from "./rem-evidence.js";
|
||||
import {
|
||||
readShortTermRecallEntries,
|
||||
recordGroundedShortTermCandidates,
|
||||
removeGroundedShortTermCandidates,
|
||||
} from "./short-term-promotion.js";
|
||||
|
||||
const DEFAULT_SESSION_BACKFILL_LIMIT_DAYS = 92;
|
||||
const SESSION_BACKFILL_QUERY_PREFIX = "__dreaming_session_backfill__";
|
||||
const SESSION_CORPUS_RELATIVE_DIR = path.join("memory", ".dreams", "session-corpus");
|
||||
const TOP_CANDIDATE_LIMIT = 5;
|
||||
const MEMORY_DAY_RE = /^\d{4}-\d{2}-\d{2}$/;
|
||||
|
||||
export type MemorySessionBackfillOptions = {
|
||||
agent?: string;
|
||||
from?: string;
|
||||
to?: string;
|
||||
limitDays?: number;
|
||||
rem?: boolean;
|
||||
apply?: boolean;
|
||||
rollback?: boolean;
|
||||
archiveFiles?: string[];
|
||||
json?: boolean;
|
||||
};
|
||||
|
||||
type SessionBackfillSource = {
|
||||
agentId: string;
|
||||
absolutePath: string;
|
||||
foreign: boolean;
|
||||
sessionPath: string;
|
||||
stateKey: string;
|
||||
scope: string;
|
||||
legacyScope?: string;
|
||||
sessionId?: string;
|
||||
sessionKey?: string;
|
||||
storePath?: string;
|
||||
updatedAtMs?: number;
|
||||
generatedByDreamingNarrative?: boolean;
|
||||
generatedByCronRun?: boolean;
|
||||
sessionKind: "interactive";
|
||||
};
|
||||
|
||||
type SessionBackfillCandidate = SessionIngestionMessage & {
|
||||
hash: string;
|
||||
contentIndex: number;
|
||||
legacyHash?: string;
|
||||
lineNumber: number;
|
||||
scope: string;
|
||||
};
|
||||
|
||||
type SessionBackfillScan = {
|
||||
candidates: SessionBackfillCandidate[];
|
||||
contentHash: string;
|
||||
lineCount: number;
|
||||
mtimeMs: number;
|
||||
progressBlockIndex?: number;
|
||||
scannedEndIndex: number;
|
||||
size: number;
|
||||
stateKey: string;
|
||||
};
|
||||
|
||||
type SessionBackfillCollection = {
|
||||
byDay: Map<string, SessionBackfillCandidate[]>;
|
||||
scans: SessionBackfillScan[];
|
||||
};
|
||||
|
||||
type SessionBackfillDay = {
|
||||
day: string;
|
||||
candidateCount: number;
|
||||
topCandidates: string[];
|
||||
};
|
||||
|
||||
type SessionBackfillResult = {
|
||||
agentId: string;
|
||||
workspaceDir: string;
|
||||
applied: boolean;
|
||||
rem: boolean;
|
||||
days: SessionBackfillDay[];
|
||||
candidateCount: number;
|
||||
stagedEntries: number;
|
||||
writtenDiaryEntries: number;
|
||||
replacedDiaryEntries: number;
|
||||
rollback?: {
|
||||
removedDiaryEntries: number;
|
||||
removedStagedEntries: number;
|
||||
};
|
||||
};
|
||||
|
||||
type RunSessionBackfillParams = {
|
||||
agentId: string;
|
||||
workspaceDir: string;
|
||||
from?: string;
|
||||
to?: string;
|
||||
limitDays?: number;
|
||||
rem?: boolean;
|
||||
apply?: boolean;
|
||||
rollback?: boolean;
|
||||
archiveFiles?: string[];
|
||||
nowMs?: number;
|
||||
timezone?: string;
|
||||
};
|
||||
|
||||
function normalizeMemoryDay(value: string | undefined, flag: string): string | undefined {
|
||||
if (value === undefined) {
|
||||
return undefined;
|
||||
}
|
||||
const day = value.trim();
|
||||
if (!MEMORY_DAY_RE.test(day)) {
|
||||
throw new Error(`${flag} must use YYYY-MM-DD.`);
|
||||
}
|
||||
const parsed = new Date(`${day}T00:00:00.000Z`);
|
||||
if (!Number.isFinite(parsed.getTime()) || parsed.toISOString().slice(0, 10) !== day) {
|
||||
throw new Error(`${flag} must be a valid calendar day.`);
|
||||
}
|
||||
return day;
|
||||
}
|
||||
|
||||
function resolveLimitDays(value: number | undefined): number {
|
||||
if (value === undefined) {
|
||||
return DEFAULT_SESSION_BACKFILL_LIMIT_DAYS;
|
||||
}
|
||||
if (!Number.isInteger(value) || value <= 0) {
|
||||
throw new Error("--limit-days must be a positive integer.");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function sourceFromCorpusEntry(entry: SessionTranscriptCorpusEntry): SessionBackfillSource | null {
|
||||
if (
|
||||
entry.sessionKind !== "interactive" ||
|
||||
entry.generatedByDreamingNarrative ||
|
||||
entry.generatedByCronRun
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
const sessionPath =
|
||||
entry.transcriptSource === "sqlite"
|
||||
? buildSqliteDreamingSessionPath(entry.agentId, entry.sessionId)
|
||||
: sessionPathForFile(entry.sessionFile);
|
||||
return {
|
||||
agentId: entry.agentId,
|
||||
absolutePath: entry.sessionFile,
|
||||
foreign: false,
|
||||
sessionPath,
|
||||
stateKey: buildSessionStateKey(entry.agentId, sessionPath),
|
||||
scope:
|
||||
entry.transcriptSource === "sqlite"
|
||||
? `${entry.agentId}:${sessionPath}`
|
||||
: buildSessionFileScopeKey(entry.agentId, entry.sessionFile),
|
||||
...(entry.transcriptSource === "sqlite"
|
||||
? { legacyScope: `${entry.agentId}:${entry.sessionId}` }
|
||||
: {}),
|
||||
...(entry.transcriptSource === "sqlite" ? { sessionId: entry.sessionId } : {}),
|
||||
...(entry.sessionKey ? { sessionKey: entry.sessionKey } : {}),
|
||||
...(entry.storePath ? { storePath: entry.storePath } : {}),
|
||||
...(entry.updatedAtMs !== undefined ? { updatedAtMs: entry.updatedAtMs } : {}),
|
||||
...(entry.generatedByDreamingNarrative
|
||||
? { generatedByDreamingNarrative: entry.generatedByDreamingNarrative }
|
||||
: {}),
|
||||
...(entry.generatedByCronRun ? { generatedByCronRun: entry.generatedByCronRun } : {}),
|
||||
sessionKind: "interactive",
|
||||
};
|
||||
}
|
||||
|
||||
function sourceFromArchiveFile(agentId: string, archiveFile: string): SessionBackfillSource {
|
||||
const absolutePath = path.resolve(archiveFile);
|
||||
const sessionPath = sessionPathForFile(absolutePath);
|
||||
return {
|
||||
agentId,
|
||||
absolutePath,
|
||||
foreign: true,
|
||||
sessionPath,
|
||||
stateKey: `session-backfill:${absolutePath.replaceAll("\\", "/")}`,
|
||||
// Foreign files do not inherit canonical session identity from a matching basename.
|
||||
scope: `archive:${agentId}:${absolutePath.replaceAll("\\", "/")}`,
|
||||
sessionKind: "interactive",
|
||||
};
|
||||
}
|
||||
|
||||
async function listSessionBackfillSources(params: {
|
||||
agentId: string;
|
||||
archiveFiles: string[];
|
||||
}): Promise<SessionBackfillSource[]> {
|
||||
const corpus = await listSessionTranscriptCorpusEntriesForAgent(params.agentId, {
|
||||
includeRetainedSqlite: true,
|
||||
});
|
||||
const sources = corpus
|
||||
.map(sourceFromCorpusEntry)
|
||||
.filter((entry): entry is SessionBackfillSource => entry !== null);
|
||||
const canonicalPaths = new Set(sources.map((entry) => path.resolve(entry.absolutePath)));
|
||||
for (const archiveFile of params.archiveFiles) {
|
||||
const source = sourceFromArchiveFile(params.agentId, archiveFile);
|
||||
if (!canonicalPaths.has(source.absolutePath)) {
|
||||
sources.push(source);
|
||||
canonicalPaths.add(source.absolutePath);
|
||||
}
|
||||
}
|
||||
return sources.toSorted((a, b) =>
|
||||
a.sessionPath === b.sessionPath
|
||||
? a.absolutePath.localeCompare(b.absolutePath)
|
||||
: a.sessionPath.localeCompare(b.sessionPath),
|
||||
);
|
||||
}
|
||||
|
||||
function candidateDayInRange(
|
||||
day: string,
|
||||
from: string | undefined,
|
||||
to: string | undefined,
|
||||
): boolean {
|
||||
return (from === undefined || day >= from) && (to === undefined || day <= to);
|
||||
}
|
||||
|
||||
function compareSessionBackfillCandidates(
|
||||
a: SessionBackfillCandidate,
|
||||
b: SessionBackfillCandidate,
|
||||
): number {
|
||||
if (a.day !== b.day) {
|
||||
return a.day.localeCompare(b.day);
|
||||
}
|
||||
if (a.provenance.observedAt !== b.provenance.observedAt) {
|
||||
return a.provenance.observedAt - b.provenance.observedAt;
|
||||
}
|
||||
if (a.scope !== b.scope) {
|
||||
return a.scope.localeCompare(b.scope);
|
||||
}
|
||||
return a.lineNumber - b.lineNumber;
|
||||
}
|
||||
|
||||
async function collectSessionBackfillCandidates(params: {
|
||||
sources: SessionBackfillSource[];
|
||||
files: Record<string, SessionIngestionFileState>;
|
||||
seenMessages: Record<string, string[]>;
|
||||
from?: string;
|
||||
to?: string;
|
||||
timezone?: string;
|
||||
}): Promise<SessionBackfillCollection> {
|
||||
const candidates: SessionBackfillCandidate[] = [];
|
||||
const scans: SessionBackfillScan[] = [];
|
||||
const perFileCap = Math.min(
|
||||
SESSION_INGESTION_MAX_MESSAGES_PER_FILE,
|
||||
Math.max(
|
||||
SESSION_INGESTION_MIN_MESSAGES_PER_FILE,
|
||||
Math.ceil(SESSION_INGESTION_MAX_MESSAGES_PER_SWEEP / Math.max(1, params.sources.length)),
|
||||
),
|
||||
);
|
||||
|
||||
for (const source of params.sources) {
|
||||
const entry = await buildSessionEntry(source.absolutePath, {
|
||||
agentId: source.agentId,
|
||||
sessionKind: source.sessionKind,
|
||||
...(source.generatedByDreamingNarrative !== undefined
|
||||
? { generatedByDreamingNarrative: source.generatedByDreamingNarrative }
|
||||
: {}),
|
||||
...(source.generatedByCronRun !== undefined
|
||||
? { generatedByCronRun: source.generatedByCronRun }
|
||||
: {}),
|
||||
...(source.sessionKey !== undefined ? { sessionKey: source.sessionKey } : {}),
|
||||
...(source.sessionId !== undefined ? { sessionId: source.sessionId } : {}),
|
||||
...(source.storePath !== undefined ? { storePath: source.storePath } : {}),
|
||||
...(source.updatedAtMs !== undefined ? { updatedAtMs: source.updatedAtMs } : {}),
|
||||
});
|
||||
if (!entry || entry.generatedByDreamingNarrative || entry.generatedByCronRun) {
|
||||
continue;
|
||||
}
|
||||
const seen = new Set(params.seenMessages[source.scope] ?? []);
|
||||
const legacySeen = source.legacyScope
|
||||
? new Set(params.seenMessages[source.legacyScope] ?? [])
|
||||
: undefined;
|
||||
const lines = entry.content.length > 0 ? entry.content.split("\n") : [];
|
||||
const previous = params.files[source.stateKey];
|
||||
const unchanged =
|
||||
previous?.mtimeMs === Math.floor(entry.mtimeMs) &&
|
||||
previous.size === Math.floor(entry.size) &&
|
||||
previous.contentHash === entry.hash &&
|
||||
previous.lineCount === lines.length;
|
||||
const startIndex = unchanged ? Math.min(previous.lastContentLine, lines.length) : 0;
|
||||
const sourceCandidates: SessionBackfillCandidate[] = [];
|
||||
let progressBlockIndex: number | undefined;
|
||||
let scannedEndIndex = startIndex;
|
||||
for (let index = startIndex; index < lines.length; index += 1) {
|
||||
scannedEndIndex = index + 1;
|
||||
const snippet = normalizeSessionCorpusSnippet(lines[index] ?? "");
|
||||
if (snippet.length < SESSION_INGESTION_MIN_SNIPPET_CHARS) {
|
||||
continue;
|
||||
}
|
||||
const lineNumber = entry.lineMap[index] ?? index + 1;
|
||||
const timestampMs = entry.messageTimestampsMs[index] ?? 0;
|
||||
const parsedProvenance = entry.lineProvenance[index] ?? {
|
||||
originClass: "untrusted" as const,
|
||||
sessionKind: "interactive" as const,
|
||||
observedAt: timestampMs || entry.mtimeMs,
|
||||
};
|
||||
// Foreign JSONL is caller-controlled, so embedded owner metadata is not
|
||||
// authenticated. Only the canonical session store can establish trust.
|
||||
const provenance = source.foreign
|
||||
? { ...parsedProvenance, originClass: "untrusted" as const }
|
||||
: parsedProvenance;
|
||||
// Canonical parsing emits `agent` only inside an authenticated owner
|
||||
// turn; replies to non-owner input retain the turn's untrusted taint.
|
||||
if (provenance.originClass !== "owner" && provenance.originClass !== "agent") {
|
||||
continue;
|
||||
}
|
||||
const day = formatMemoryDreamingDay(
|
||||
timestampMs > 0 ? timestampMs : entry.mtimeMs,
|
||||
params.timezone,
|
||||
);
|
||||
if (!candidateDayInRange(day, params.from, params.to)) {
|
||||
progressBlockIndex ??= index;
|
||||
continue;
|
||||
}
|
||||
const dedupeBasis = timestampMs > 0 ? `ts:${Math.floor(timestampMs)}` : `line:${lineNumber}`;
|
||||
const hash = hashSessionMessageId(`${source.scope}\n${dedupeBasis}\n${snippet}`);
|
||||
const legacyHash = source.legacyScope
|
||||
? hashSessionMessageId(`${source.legacyScope}\n${dedupeBasis}\n${snippet}`)
|
||||
: undefined;
|
||||
if (seen.has(hash) || (legacyHash !== undefined && legacySeen?.has(legacyHash))) {
|
||||
continue;
|
||||
}
|
||||
const rendered = buildSessionRenderedLine({
|
||||
agentId: source.agentId,
|
||||
sessionPath: source.sessionPath,
|
||||
lineNumber,
|
||||
snippet,
|
||||
});
|
||||
const candidate = {
|
||||
contentIndex: index,
|
||||
day,
|
||||
hash,
|
||||
...(legacyHash ? { legacyHash } : {}),
|
||||
lineNumber,
|
||||
provenance,
|
||||
rendered,
|
||||
scope: source.scope,
|
||||
snippet,
|
||||
} satisfies SessionBackfillCandidate;
|
||||
sourceCandidates.push(candidate);
|
||||
seen.add(hash);
|
||||
}
|
||||
candidates.push(
|
||||
...sourceCandidates.toSorted(compareSessionBackfillCandidates).slice(0, perFileCap),
|
||||
);
|
||||
scans.push({
|
||||
candidates: sourceCandidates,
|
||||
contentHash: entry.hash,
|
||||
lineCount: lines.length,
|
||||
mtimeMs: Math.floor(entry.mtimeMs),
|
||||
...(progressBlockIndex !== undefined ? { progressBlockIndex } : {}),
|
||||
scannedEndIndex,
|
||||
size: Math.floor(entry.size),
|
||||
stateKey: source.stateKey,
|
||||
});
|
||||
}
|
||||
const selected = candidates
|
||||
.toSorted(compareSessionBackfillCandidates)
|
||||
.slice(0, SESSION_INGESTION_MAX_MESSAGES_PER_SWEEP);
|
||||
const byDay = new Map<string, SessionBackfillCandidate[]>();
|
||||
for (const candidate of selected) {
|
||||
const bucket = byDay.get(candidate.day) ?? [];
|
||||
bucket.push(candidate);
|
||||
byDay.set(candidate.day, bucket);
|
||||
}
|
||||
return { byDay, scans };
|
||||
}
|
||||
|
||||
function mergeSessionBackfillFileProgress(params: {
|
||||
current: Record<string, SessionIngestionFileState>;
|
||||
scans: SessionBackfillScan[];
|
||||
selectedDays: Array<{ candidates: SessionBackfillCandidate[] }>;
|
||||
}): Record<string, SessionIngestionFileState> {
|
||||
const selectedHashes = new Set(
|
||||
params.selectedDays.flatMap((day) => day.candidates.map((candidate) => candidate.hash)),
|
||||
);
|
||||
const files = { ...params.current };
|
||||
for (const scan of params.scans) {
|
||||
const firstUnselected = scan.candidates.find(
|
||||
(candidate) => !selectedHashes.has(candidate.hash),
|
||||
);
|
||||
const progressStops = [
|
||||
scan.scannedEndIndex,
|
||||
...(firstUnselected ? [firstUnselected.contentIndex] : []),
|
||||
...(scan.progressBlockIndex !== undefined ? [scan.progressBlockIndex] : []),
|
||||
];
|
||||
files[scan.stateKey] = {
|
||||
mtimeMs: scan.mtimeMs,
|
||||
size: scan.size,
|
||||
contentHash: scan.contentHash,
|
||||
lineCount: scan.lineCount,
|
||||
lastContentLine: Math.min(...progressStops),
|
||||
};
|
||||
}
|
||||
return files;
|
||||
}
|
||||
|
||||
function summarizeDay(day: string, candidates: SessionBackfillCandidate[]): SessionBackfillDay {
|
||||
return {
|
||||
day,
|
||||
candidateCount: candidates.length,
|
||||
topCandidates: candidates.slice(0, TOP_CANDIDATE_LIMIT).map((entry) => entry.snippet),
|
||||
};
|
||||
}
|
||||
|
||||
function buildSummaryDiaryLines(day: SessionBackfillDay): string[] {
|
||||
return [
|
||||
`Session backfill found ${day.candidateCount} trusted candidate${day.candidateCount === 1 ? "" : "s"}.`,
|
||||
...day.topCandidates.map((candidate) => `- ${candidate}`),
|
||||
];
|
||||
}
|
||||
|
||||
function groundedMarkdownToDiaryLines(markdown: string): string[] {
|
||||
return markdown
|
||||
.split(/\r?\n/)
|
||||
.map((line) => line.replace(/^##\s+/, "").trimEnd())
|
||||
.filter((line, index, lines) => !(line.length === 0 && lines[index - 1]?.length === 0));
|
||||
}
|
||||
|
||||
async function buildRemDiaryEntries(params: {
|
||||
days: Array<{ day: string; candidates: SessionBackfillCandidate[] }>;
|
||||
}): Promise<Array<{ isoDay: string; sourcePath: string; bodyLines: string[] }>> {
|
||||
const scratchDir = await fs.mkdtemp(
|
||||
path.join(resolvePreferredOpenClawTmpDir(), "openclaw-session-backfill-"),
|
||||
);
|
||||
try {
|
||||
const entries: Array<{ isoDay: string; sourcePath: string; bodyLines: string[] }> = [];
|
||||
for (const day of params.days) {
|
||||
const results = await appendSessionCorpusLines({
|
||||
workspaceDir: scratchDir,
|
||||
day: day.day,
|
||||
lines: day.candidates,
|
||||
});
|
||||
if (results.length === 0) {
|
||||
continue;
|
||||
}
|
||||
const corpusPath = path.join(scratchDir, SESSION_CORPUS_RELATIVE_DIR, `${day.day}.txt`);
|
||||
const inputPath = path.join(scratchDir, "memory", `${day.day}.md`);
|
||||
const corpus = await fs.readFile(corpusPath, "utf-8");
|
||||
await fs.writeFile(inputPath, `## Session transcript\n\n${corpus}`);
|
||||
const preview = await previewGroundedRemMarkdown({
|
||||
workspaceDir: scratchDir,
|
||||
inputPaths: [inputPath],
|
||||
});
|
||||
const file = preview.files.at(0);
|
||||
const hasGroundedContent = Boolean(
|
||||
file &&
|
||||
(file.facts.length > 0 ||
|
||||
file.reflections.length > 0 ||
|
||||
file.memoryImplications.length > 0 ||
|
||||
file.candidates.length > 0),
|
||||
);
|
||||
entries.push({
|
||||
isoDay: day.day,
|
||||
sourcePath: results[0]?.path ?? `memory/.dreams/session-corpus/${day.day}.txt`,
|
||||
bodyLines:
|
||||
hasGroundedContent && file
|
||||
? groundedMarkdownToDiaryLines(file.renderedMarkdown)
|
||||
: buildSummaryDiaryLines(summarizeDay(day.day, day.candidates)),
|
||||
});
|
||||
}
|
||||
return entries;
|
||||
} finally {
|
||||
await fs.rm(scratchDir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
function uniqueGroundedItems(results: MemorySearchResult[]): MemorySearchResult[] {
|
||||
const seen = new Set<string>();
|
||||
return results.flatMap((result) => {
|
||||
const snippet = result.snippet.replace(/^(?:Assistant|User):\s*/i, "").trim();
|
||||
const key = snippet.replace(/\s+/g, " ").toLowerCase();
|
||||
if (!key || seen.has(key)) {
|
||||
return [];
|
||||
}
|
||||
seen.add(key);
|
||||
return [{ ...result, snippet }];
|
||||
});
|
||||
}
|
||||
|
||||
async function applySessionBackfillDays(params: {
|
||||
workspaceDir: string;
|
||||
days: Array<{ day: string; candidates: SessionBackfillCandidate[] }>;
|
||||
nowMs: number;
|
||||
timezone?: string;
|
||||
}): Promise<number> {
|
||||
const before = await readShortTermRecallEntries({
|
||||
workspaceDir: params.workspaceDir,
|
||||
nowMs: params.nowMs,
|
||||
});
|
||||
for (const day of params.days) {
|
||||
const results = await appendSessionCorpusLines({
|
||||
workspaceDir: params.workspaceDir,
|
||||
day: day.day,
|
||||
lines: day.candidates,
|
||||
});
|
||||
const grounded = uniqueGroundedItems(results);
|
||||
if (grounded.length === 0) {
|
||||
continue;
|
||||
}
|
||||
// Standard grounded staging owns claim identity. Exact duplicates are
|
||||
// collapsed here; claim-hash keying also converges the same fact across sources.
|
||||
await recordGroundedShortTermCandidates({
|
||||
workspaceDir: params.workspaceDir,
|
||||
query: `${SESSION_BACKFILL_QUERY_PREFIX}:${day.day}`,
|
||||
items: grounded.map((result) => ({
|
||||
path: result.path,
|
||||
startLine: result.startLine,
|
||||
endLine: result.endLine,
|
||||
snippet: result.snippet,
|
||||
score: SESSION_INGESTION_SCORE,
|
||||
dayBucket: day.day,
|
||||
})),
|
||||
dedupeByQueryPerDay: true,
|
||||
nowMs: params.nowMs,
|
||||
...(params.timezone !== undefined ? { timezone: params.timezone } : {}),
|
||||
});
|
||||
}
|
||||
const after = await readShortTermRecallEntries({
|
||||
workspaceDir: params.workspaceDir,
|
||||
nowMs: params.nowMs,
|
||||
});
|
||||
return Math.max(0, after.length - before.length);
|
||||
}
|
||||
|
||||
export async function runSessionBackfill(
|
||||
params: RunSessionBackfillParams,
|
||||
): Promise<SessionBackfillResult> {
|
||||
const workspaceDir = params.workspaceDir.trim();
|
||||
if (!workspaceDir) {
|
||||
throw new Error("Memory session-backfill requires a resolvable workspace directory.");
|
||||
}
|
||||
if (params.rem && params.apply) {
|
||||
throw new Error("Memory session-backfill --rem cannot be combined with --apply.");
|
||||
}
|
||||
const nowMs = Number.isFinite(params.nowMs) ? (params.nowMs as number) : Date.now();
|
||||
if (params.rollback) {
|
||||
// Backfill diary markers and grounded-only candidates are a shared artifact
|
||||
// class with rem-backfill; the stable removal APIs intentionally clear both.
|
||||
// Ingestion cursors and hashes remain: rollback must not re-ingest tracked messages.
|
||||
const [diary, staged] = await Promise.all([
|
||||
removeBackfillDiaryEntries({ workspaceDir }),
|
||||
removeGroundedShortTermCandidates({ workspaceDir }),
|
||||
]);
|
||||
return {
|
||||
agentId: params.agentId,
|
||||
workspaceDir,
|
||||
applied: false,
|
||||
rem: false,
|
||||
days: [],
|
||||
candidateCount: 0,
|
||||
stagedEntries: 0,
|
||||
writtenDiaryEntries: 0,
|
||||
replacedDiaryEntries: 0,
|
||||
rollback: {
|
||||
removedDiaryEntries: diary.removed,
|
||||
removedStagedEntries: staged.removed,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const from = normalizeMemoryDay(params.from, "--from");
|
||||
const to = normalizeMemoryDay(params.to, "--to");
|
||||
if (from !== undefined && to !== undefined && from > to) {
|
||||
throw new Error("--from must not be after --to.");
|
||||
}
|
||||
const limitDays = resolveLimitDays(params.limitDays);
|
||||
const state = await readSessionIngestionState(workspaceDir);
|
||||
const sources = await listSessionBackfillSources({
|
||||
agentId: params.agentId,
|
||||
archiveFiles: params.archiveFiles ?? [],
|
||||
});
|
||||
const collected = await collectSessionBackfillCandidates({
|
||||
sources,
|
||||
files: state.files,
|
||||
seenMessages: state.seenMessages,
|
||||
...(from !== undefined ? { from } : {}),
|
||||
...(to !== undefined ? { to } : {}),
|
||||
...(params.timezone !== undefined ? { timezone: params.timezone } : {}),
|
||||
});
|
||||
const selectedDays = [...collected.byDay.keys()]
|
||||
.toSorted()
|
||||
.slice(0, limitDays)
|
||||
.map((day) => ({ day, candidates: collected.byDay.get(day) ?? [] }));
|
||||
const days = selectedDays.map((entry) => summarizeDay(entry.day, entry.candidates));
|
||||
const candidateCount = days.reduce((sum, day) => sum + day.candidateCount, 0);
|
||||
let writtenDiaryEntries = 0;
|
||||
let replacedDiaryEntries = 0;
|
||||
let stagedEntries = 0;
|
||||
|
||||
if (selectedDays.length > 0 && (params.rem || params.apply)) {
|
||||
const diaryEntries = params.rem
|
||||
? await buildRemDiaryEntries({ days: selectedDays })
|
||||
: selectedDays.map((entry) => ({
|
||||
isoDay: entry.day,
|
||||
sourcePath: `memory/.dreams/session-corpus/${entry.day}.txt`,
|
||||
bodyLines: buildSummaryDiaryLines(summarizeDay(entry.day, entry.candidates)),
|
||||
}));
|
||||
const diary = await writeBackfillDiaryEntries({
|
||||
workspaceDir,
|
||||
entries: diaryEntries,
|
||||
preserveExisting: true,
|
||||
...(params.timezone !== undefined ? { timezone: params.timezone } : {}),
|
||||
});
|
||||
writtenDiaryEntries = diary.written;
|
||||
replacedDiaryEntries = diary.replaced;
|
||||
}
|
||||
|
||||
if (params.apply) {
|
||||
if (selectedDays.length > 0) {
|
||||
stagedEntries = await applySessionBackfillDays({
|
||||
workspaceDir,
|
||||
days: selectedDays,
|
||||
nowMs,
|
||||
...(params.timezone !== undefined ? { timezone: params.timezone } : {}),
|
||||
});
|
||||
}
|
||||
const nextSeenMessages = { ...state.seenMessages };
|
||||
for (const { candidates } of selectedDays) {
|
||||
const hashesByScope = new Map<string, string[]>();
|
||||
for (const candidate of candidates) {
|
||||
const hashes = hashesByScope.get(candidate.scope) ?? [];
|
||||
hashes.push(candidate.hash);
|
||||
hashesByScope.set(candidate.scope, hashes);
|
||||
}
|
||||
for (const [scope, hashes] of hashesByScope) {
|
||||
nextSeenMessages[scope] = mergeTrackedMessageHashes(nextSeenMessages[scope] ?? [], hashes);
|
||||
}
|
||||
}
|
||||
await writeSessionIngestionState(workspaceDir, {
|
||||
...state,
|
||||
files: mergeSessionBackfillFileProgress({
|
||||
current: state.files,
|
||||
scans: collected.scans,
|
||||
selectedDays,
|
||||
}),
|
||||
seenMessages: trimTrackedSessionScopes(nextSeenMessages),
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
agentId: params.agentId,
|
||||
workspaceDir,
|
||||
applied: Boolean(params.apply),
|
||||
rem: Boolean(params.rem),
|
||||
days,
|
||||
candidateCount,
|
||||
stagedEntries,
|
||||
writtenDiaryEntries,
|
||||
replacedDiaryEntries,
|
||||
};
|
||||
}
|
||||
@@ -21,6 +21,7 @@ export {
|
||||
type SessionFileState,
|
||||
type SessionTranscriptClassification,
|
||||
type SessionTranscriptCorpusEntry,
|
||||
type SessionTranscriptCorpusOptions,
|
||||
} from "./host/session-files.js";
|
||||
export {
|
||||
isSessionArchiveArtifactName,
|
||||
|
||||
@@ -17,6 +17,7 @@ export {
|
||||
isUsageCountedSessionTranscriptFileName,
|
||||
loadTranscriptEventsSync,
|
||||
listSessionEntries,
|
||||
listSessionTranscriptInstances,
|
||||
onSessionTranscriptUpdate,
|
||||
parseSqliteSessionFileMarker,
|
||||
parseUsageCountedSessionIdFromFileName,
|
||||
@@ -30,6 +31,7 @@ export {
|
||||
stripInboundMetadata,
|
||||
stripInternalRuntimeContext,
|
||||
type SessionEntry,
|
||||
type SessionTranscriptInstance,
|
||||
} from "./openclaw-runtime.js";
|
||||
|
||||
/** Extracts the agent id from a canonical `agents/<id>/sessions` directory path. */
|
||||
|
||||
@@ -52,6 +52,10 @@ export {
|
||||
parseUsageCountedSessionIdFromFileName,
|
||||
} from "../../../../src/config/sessions/artifacts.js";
|
||||
export { canonicalizeMainSessionAlias } from "../../../../src/config/sessions/main-session.js";
|
||||
export {
|
||||
listSessionTranscriptInstances,
|
||||
type SessionTranscriptInstance,
|
||||
} from "../../../../src/config/sessions/session-history.js";
|
||||
export {
|
||||
resolveSessionFilePath,
|
||||
resolveSessionTranscriptsDirForAgent,
|
||||
|
||||
@@ -9,7 +9,9 @@ import {
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
|
||||
import { markInboundContextLabel } from "../../../../src/auto-reply/reply/inbound-context-marker.js";
|
||||
import {
|
||||
appendTranscriptMessage,
|
||||
persistSessionTranscriptTurn,
|
||||
replaceSessionEntry,
|
||||
upsertSessionEntry,
|
||||
} from "../../../../src/config/sessions/session-accessor.js";
|
||||
import {
|
||||
@@ -119,6 +121,67 @@ describe("listSessionFilesForAgent", () => {
|
||||
});
|
||||
|
||||
describe("listSessionTranscriptCorpusEntriesForAgent", () => {
|
||||
it("includes rotated SQLite sessions only when retained history is requested", async () => {
|
||||
const sessionsDir = path.join(tmpDir, "agents", "main", "sessions");
|
||||
const storePath = path.join(sessionsDir, "sessions.json");
|
||||
const sessionKey = "agent:main:main";
|
||||
fsSync.mkdirSync(sessionsDir, { recursive: true });
|
||||
|
||||
await upsertSessionEntry(
|
||||
{ agentId: "main", sessionKey, storePath },
|
||||
{ sessionId: "retained-old", updatedAt: 10 },
|
||||
);
|
||||
await appendTranscriptMessage(
|
||||
{ agentId: "main", sessionId: "retained-old", sessionKey, storePath },
|
||||
{ message: { role: "assistant", content: "retained transcript" } },
|
||||
);
|
||||
await replaceSessionEntry(
|
||||
{ agentId: "main", sessionKey, storePath },
|
||||
{ sessionId: "retained-old", updatedAt: 15 },
|
||||
);
|
||||
await upsertSessionEntry(
|
||||
{ agentId: "main", sessionKey, storePath },
|
||||
{ sessionId: "retained-new", updatedAt: 20 },
|
||||
);
|
||||
await appendTranscriptMessage(
|
||||
{ agentId: "main", sessionId: "retained-new", sessionKey, storePath },
|
||||
{ message: { role: "assistant", content: "current transcript" } },
|
||||
);
|
||||
|
||||
const currentOnly = await listSessionTranscriptCorpusEntriesForAgent("main");
|
||||
expect(currentOnly.map((entry) => entry.sessionId)).toEqual(["retained-new"]);
|
||||
|
||||
const withHistory = await listSessionTranscriptCorpusEntriesForAgent("main", {
|
||||
includeRetainedSqlite: true,
|
||||
});
|
||||
expect(withHistory).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
artifactKind: "active-session",
|
||||
sessionId: "retained-new",
|
||||
transcriptSource: "sqlite",
|
||||
}),
|
||||
expect.objectContaining({
|
||||
artifactKind: "retained-session",
|
||||
sessionId: "retained-old",
|
||||
transcriptSource: "sqlite",
|
||||
}),
|
||||
]),
|
||||
);
|
||||
const retained = withHistory.find((entry) => entry.sessionId === "retained-old");
|
||||
expect(
|
||||
requireSessionEntry(
|
||||
await buildSessionEntry(retained?.sessionFile ?? "", {
|
||||
...(retained?.agentId !== undefined ? { agentId: retained.agentId } : {}),
|
||||
...(retained?.sessionId !== undefined ? { sessionId: retained.sessionId } : {}),
|
||||
...(retained?.sessionKey !== undefined ? { sessionKey: retained.sessionKey } : {}),
|
||||
...(retained?.storePath !== undefined ? { storePath: retained.storePath } : {}),
|
||||
...(retained?.updatedAtMs !== undefined ? { updatedAtMs: retained.updatedAtMs } : {}),
|
||||
}),
|
||||
).content,
|
||||
).toBe("Assistant: retained transcript");
|
||||
});
|
||||
|
||||
it("treats accessor-backed entries as live SQLite transcripts", async () => {
|
||||
const sessionsDir = path.join(tmpDir, "agents", "main", "sessions");
|
||||
fsSync.mkdirSync(sessionsDir, { recursive: true });
|
||||
|
||||
@@ -40,6 +40,7 @@ import type { MemoryEntryProvenance, MemoryOriginClass, MemorySessionKind } from
|
||||
export {
|
||||
listSessionTranscriptCorpusEntriesForAgent,
|
||||
type SessionTranscriptCorpusEntry,
|
||||
type SessionTranscriptCorpusOptions,
|
||||
} from "./session-transcript-corpus.js";
|
||||
|
||||
// Keep the historical one-line-per-message export shape for normal turns, but
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
isSessionArchiveArtifactName,
|
||||
isUsageCountedSessionTranscriptFileName,
|
||||
listSessionEntries,
|
||||
listSessionTranscriptInstances,
|
||||
parseSqliteSessionFileMarker,
|
||||
parseUsageCountedSessionIdFromFileName,
|
||||
readTranscriptContentRevisionSync,
|
||||
@@ -19,10 +20,19 @@ import {
|
||||
resolveSessionTranscriptsDirForAgent,
|
||||
resolveStorePath,
|
||||
type SessionEntry,
|
||||
type SessionTranscriptInstance,
|
||||
} from "./openclaw-runtime-session.js";
|
||||
import type { MemorySessionKind } from "./types.js";
|
||||
|
||||
type SessionTranscriptCorpusArtifactKind = "active-session" | "archive-artifact";
|
||||
type SessionTranscriptCorpusArtifactKind =
|
||||
| "active-session"
|
||||
| "retained-session"
|
||||
| "archive-artifact";
|
||||
|
||||
export type SessionTranscriptCorpusOptions = {
|
||||
/** Include rotated SQLite transcript identities retained behind current logical sessions. */
|
||||
includeRetainedSqlite?: boolean;
|
||||
};
|
||||
|
||||
export type SessionTranscriptCorpusEntry = {
|
||||
agentId: string;
|
||||
@@ -343,6 +353,46 @@ function toSessionStoreCorpusEntry(
|
||||
};
|
||||
}
|
||||
|
||||
function toRetainedSessionCorpusEntry(
|
||||
agentId: string,
|
||||
instance: SessionTranscriptInstance,
|
||||
sessionKey: string,
|
||||
storePath: string,
|
||||
cronGeneratedSessionKeys: ReadonlySet<string>,
|
||||
): SessionTranscriptCorpusEntry | null {
|
||||
// Retained rows predate the current logical session entry. Only rows whose
|
||||
// exclusion-sensitive ownership was captured may enter historical ingestion.
|
||||
if (
|
||||
!instance.provenanceKnown ||
|
||||
instance.acpOwned ||
|
||||
instance.entry.pluginOwnerId ||
|
||||
instance.entry.hookExternalContentSource
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
const classification = classifySessionEntry(sessionKey, instance.entry, cronGeneratedSessionKeys);
|
||||
const contentRevision = sqliteContentRevision({
|
||||
agentId,
|
||||
sessionId: instance.sessionId,
|
||||
...(sessionKey ? { sessionKey } : {}),
|
||||
storePath,
|
||||
});
|
||||
return {
|
||||
agentId,
|
||||
artifactKind: "retained-session",
|
||||
sessionFile: sessionKey,
|
||||
sessionId: instance.sessionId,
|
||||
...(contentRevision ? { contentRevision } : {}),
|
||||
storePath,
|
||||
transcriptSource: "sqlite",
|
||||
updatedAtMs: instance.updatedAtMs,
|
||||
...(sessionKey ? { sessionKey } : {}),
|
||||
...(classification.generatedByDreamingNarrative ? { generatedByDreamingNarrative: true } : {}),
|
||||
...(classification.generatedByCronRun ? { generatedByCronRun: true } : {}),
|
||||
sessionKind: classification.sessionKind,
|
||||
};
|
||||
}
|
||||
|
||||
function listSessionTranscriptArtifactFiles(sessionsDir: string): string[] {
|
||||
try {
|
||||
return fsSync
|
||||
@@ -422,6 +472,7 @@ function toArtifactCorpusEntry(
|
||||
|
||||
export function listSessionTranscriptCorpusEntriesForAgentSync(
|
||||
agentId: string,
|
||||
options: SessionTranscriptCorpusOptions = {},
|
||||
): SessionTranscriptCorpusEntry[] {
|
||||
const normalizedAgentId = normalizeAgentId(agentId);
|
||||
const cfg = getRuntimeConfig();
|
||||
@@ -450,7 +501,18 @@ export function listSessionTranscriptCorpusEntriesForAgentSync(
|
||||
hydrateSkillPromptRefs: false,
|
||||
storePath,
|
||||
});
|
||||
const cronGeneratedSessionKeys = collectCronGeneratedSessionKeys(sessionEntries);
|
||||
const retainedInstances = options.includeRetainedSqlite
|
||||
? listSessionTranscriptInstances({
|
||||
agentId: normalizedAgentId,
|
||||
hydrateSkillPromptRefs: false,
|
||||
readConsistency: "latest",
|
||||
storePath,
|
||||
})
|
||||
: [];
|
||||
const cronGeneratedSessionKeys = collectCronGeneratedSessionKeys([
|
||||
...retainedInstances.map(({ entry, sessionKey }) => ({ entry, sessionKey })),
|
||||
...sessionEntries,
|
||||
]);
|
||||
for (const summary of sessionEntries) {
|
||||
const sessionKey = isSharedFixedStore
|
||||
? summary.sessionKey
|
||||
@@ -491,6 +553,38 @@ export function listSessionTranscriptCorpusEntriesForAgentSync(
|
||||
const corpusEntries = [...activeEntriesBySessionId.values()].filter(
|
||||
(entry) => entry.transcriptSource === "sqlite",
|
||||
);
|
||||
if (options.includeRetainedSqlite) {
|
||||
for (const instance of retainedInstances) {
|
||||
if (activeEntriesBySessionId.has(instance.sessionId)) {
|
||||
continue;
|
||||
}
|
||||
const sessionKey = isSharedFixedStore
|
||||
? instance.sessionKey
|
||||
: canonicalizeMainSessionAlias({
|
||||
cfg,
|
||||
agentId: normalizedAgentId,
|
||||
sessionKey: instance.sessionKey,
|
||||
});
|
||||
const ownerAgentId = resolveSessionAgentId({
|
||||
config: cfg,
|
||||
sessionKey,
|
||||
...(isSharedFixedStore ? {} : { fallbackAgentId: normalizedAgentId }),
|
||||
});
|
||||
if (ownerAgentId !== normalizedAgentId) {
|
||||
continue;
|
||||
}
|
||||
const entry = toRetainedSessionCorpusEntry(
|
||||
ownerAgentId,
|
||||
instance,
|
||||
sessionKey,
|
||||
storePath,
|
||||
cronGeneratedSessionKeys,
|
||||
);
|
||||
if (entry?.transcriptSource === "sqlite") {
|
||||
corpusEntries.push(entry);
|
||||
}
|
||||
}
|
||||
}
|
||||
const scannedArtifactPaths = new Set<string>();
|
||||
for (const artifactDir of artifactDirsByPath.values()) {
|
||||
for (const artifactPath of listSessionTranscriptArtifactFiles(artifactDir)) {
|
||||
@@ -544,6 +638,7 @@ export function listSessionTranscriptCorpusEntriesForAgentSync(
|
||||
*/
|
||||
export async function listSessionTranscriptCorpusEntriesForAgent(
|
||||
agentId: string,
|
||||
options: SessionTranscriptCorpusOptions = {},
|
||||
): Promise<SessionTranscriptCorpusEntry[]> {
|
||||
return listSessionTranscriptCorpusEntriesForAgentSync(agentId);
|
||||
return listSessionTranscriptCorpusEntriesForAgentSync(agentId, options);
|
||||
}
|
||||
|
||||
@@ -28,4 +28,5 @@ export type {
|
||||
QmdQueryResult,
|
||||
SessionFileEntry,
|
||||
SessionTranscriptCorpusEntry,
|
||||
SessionTranscriptCorpusOptions,
|
||||
} from "../../packages/memory-host-sdk/src/engine-qmd.js";
|
||||
|
||||
Reference in New Issue
Block a user