feat(openclaw): add config-aware history backfill CLI (#878)
* Add OpenClaw history backfill CLI * Fix backfill resume and local daemon behavior * Fix backfill checkpoint finalization semantics * Fix symlinked backfill CLI entrypoint detection * fix(ci): skip PR status write for fork approvals --------- Co-authored-by: Nicolò Boschi <boschi1997@gmail.com>
This commit is contained in:
parent
5a61ac50e9
commit
72fd3d59db
12 changed files with 1288 additions and 6 deletions
3
.github/workflows/test.yml
vendored
3
.github/workflows/test.yml
vendored
|
|
@ -2524,7 +2524,8 @@ jobs:
|
|||
core.setOutput('run_url', runUrl);
|
||||
|
||||
- name: Report status to PR
|
||||
uses: actions/github-script@v8
|
||||
if: github.event.pull_request.head.repo.full_name == github.repository
|
||||
uses: actions/github-script@v7
|
||||
with:
|
||||
script: |
|
||||
await github.rest.repos.createCommitStatus({
|
||||
|
|
|
|||
1
hindsight-integrations/openclaw/.gitignore
vendored
1
hindsight-integrations/openclaw/.gitignore
vendored
|
|
@ -2,3 +2,4 @@ node_modules/
|
|||
dist/
|
||||
*.log
|
||||
.DS_Store
|
||||
.tmp/
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ That's it! The plugin will automatically start capturing and recalling memories.
|
|||
|
||||
- **Auto-capture** and **auto-recall** of memories each turn, injected into system prompt space so recalled memories stay out of the visible chat transcript
|
||||
- **Memory isolation** — configurable per agent, channel, user, or provider via `dynamicBankGranularity`
|
||||
- **Historical backfill CLI** — import prior OpenClaw session history into Hindsight using the active plugin bank-routing config by default
|
||||
- **Retention controls** — choose which message roles to retain, toggle auto-retain on/off, and stamp retained documents with consistent tags/source metadata
|
||||
|
||||
## Configuration
|
||||
|
|
@ -146,6 +147,51 @@ tail -f ~/.hindsight/profiles/openclaw.log
|
|||
uvx hindsight-embed@latest profile list
|
||||
```
|
||||
|
||||
## Backfilling Existing OpenClaw History
|
||||
|
||||
The package includes a config-aware backfill CLI for importing historical OpenClaw sessions into Hindsight.
|
||||
|
||||
By default it mirrors the active plugin settings for:
|
||||
|
||||
- `dynamicBankId`
|
||||
- `dynamicBankGranularity`
|
||||
- `bankIdPrefix`
|
||||
- local daemon vs external `hindsightApiUrl`
|
||||
|
||||
Dry-run example:
|
||||
|
||||
```bash
|
||||
npx --package @vectorize-io/hindsight-openclaw hindsight-openclaw-backfill \
|
||||
--openclaw-root ~/.openclaw \
|
||||
--dry-run
|
||||
```
|
||||
|
||||
Direct invocation from a built checkout:
|
||||
|
||||
```bash
|
||||
node dist/backfill.js --openclaw-root ~/.openclaw --dry-run
|
||||
```
|
||||
|
||||
Migration-oriented overrides are explicit:
|
||||
|
||||
```bash
|
||||
node dist/backfill.js \
|
||||
--openclaw-root ~/.openclaw \
|
||||
--bank-strategy agent \
|
||||
--agent proj-run \
|
||||
--resume \
|
||||
--max-pending-operations 10
|
||||
```
|
||||
|
||||
Useful options:
|
||||
|
||||
- `--agent <id>` limit import to selected agents
|
||||
- `--exclude-archive` ignore `sessions-archive-from-migration_backup`
|
||||
- `--bank-strategy mirror-config|agent|fixed`
|
||||
- `--resume` skip only entries already finalized as completed
|
||||
- `--checkpoint <path>` store progress outside the default location
|
||||
- `--wait-until-drained` block until the touched bank queues have finished and checkpoint state can be finalized
|
||||
|
||||
## Links
|
||||
|
||||
- [Hindsight Documentation](https://vectorize.io/hindsight)
|
||||
|
|
|
|||
|
|
@ -4,6 +4,9 @@
|
|||
"description": "Hindsight memory plugin for OpenClaw - biomimetic long-term memory with fact extraction",
|
||||
"main": "dist/index.js",
|
||||
"types": "dist/index.d.ts",
|
||||
"bin": {
|
||||
"hindsight-openclaw-backfill": "dist/backfill.js"
|
||||
},
|
||||
"type": "module",
|
||||
"openclaw": {
|
||||
"extensions": [
|
||||
|
|
@ -34,6 +37,7 @@
|
|||
"build": "tsc",
|
||||
"dev": "tsc --watch",
|
||||
"clean": "rm -rf dist",
|
||||
"backfill:help": "node dist/backfill.js --help",
|
||||
"test": "vitest run src",
|
||||
"test:watch": "vitest src",
|
||||
"test:integration": "vitest run --config vitest.integration.config.ts",
|
||||
|
|
|
|||
122
hindsight-integrations/openclaw/src/backfill-lib.test.ts
Normal file
122
hindsight-integrations/openclaw/src/backfill-lib.test.ts
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'fs';
|
||||
import { join } from 'path';
|
||||
import { tmpdir } from 'os';
|
||||
import {
|
||||
buildBackfillPlan,
|
||||
loadPluginConfigFromOpenClawRoot,
|
||||
stableDocumentId,
|
||||
} from './backfill-lib.js';
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
function makeTempRoot(): string {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'hindsight-openclaw-backfill-'));
|
||||
tempDirs.push(dir);
|
||||
return dir;
|
||||
}
|
||||
|
||||
function writeOpenClawConfig(root: string, config: Record<string, unknown>) {
|
||||
writeFileSync(join(root, 'openclaw.json'), JSON.stringify(config, null, 2));
|
||||
}
|
||||
|
||||
function writeSession(root: string, agentId: string, fileName: string, lines: unknown[], archive = false) {
|
||||
const dir = archive
|
||||
? join(root, 'agents', agentId, 'sessions-archive-from-migration_backup')
|
||||
: join(root, 'agents', agentId, 'sessions');
|
||||
mkdirSync(dir, { recursive: true });
|
||||
writeFileSync(join(dir, fileName), lines.map((line) => JSON.stringify(line)).join('\n') + '\n');
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
for (const dir of tempDirs.splice(0)) {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
describe('backfill planning', () => {
|
||||
it('mirrors plugin bank routing from config by default', () => {
|
||||
const root = makeTempRoot();
|
||||
writeOpenClawConfig(root, {
|
||||
plugins: {
|
||||
entries: {
|
||||
'hindsight-openclaw': {
|
||||
config: {
|
||||
dynamicBankId: true,
|
||||
dynamicBankGranularity: ['agent', 'provider', 'channel'],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
writeSession(root, 'proj-run', 'one.jsonl', [
|
||||
{ type: 'session', id: 'session-1', sessionKey: 'agent:proj-run:discord:channel:123' },
|
||||
{ type: 'message', message: { role: 'user', content: 'hello' } },
|
||||
{ type: 'message', message: { role: 'assistant', content: 'world' } },
|
||||
]);
|
||||
|
||||
const config = loadPluginConfigFromOpenClawRoot(root);
|
||||
const result = buildBackfillPlan(config, {
|
||||
openclawRoot: root,
|
||||
includeArchive: true,
|
||||
bankStrategy: 'mirror-config',
|
||||
});
|
||||
|
||||
expect(result.discoveredSessions).toBe(1);
|
||||
expect(result.entries).toHaveLength(1);
|
||||
expect(result.entries[0].bankId).toBe('proj-run::discord::channel%3A123');
|
||||
expect(result.entries[0].documentId).toBe(stableDocumentId({
|
||||
filePath: result.entries[0].filePath,
|
||||
agentId: 'proj-run',
|
||||
sessionId: 'session-1',
|
||||
sessionKey: 'agent:proj-run:discord:channel:123',
|
||||
messages: [],
|
||||
}, result.entries[0].bankId));
|
||||
});
|
||||
|
||||
it('supports migration overrides for agent-only banks', () => {
|
||||
const root = makeTempRoot();
|
||||
writeOpenClawConfig(root, { plugins: { entries: { 'hindsight-openclaw': { config: {} } } } });
|
||||
writeSession(root, 'proj-debug', 'two.jsonl', [
|
||||
{ type: 'session', id: 'session-2', sessionKey: 'agent:proj-debug:discord:group:abc' },
|
||||
{ type: 'message', message: { role: 'user', content: 'hello' } },
|
||||
{ type: 'message', message: { role: 'assistant', content: 'world' } },
|
||||
]);
|
||||
|
||||
const config = loadPluginConfigFromOpenClawRoot(root);
|
||||
const result = buildBackfillPlan(config, {
|
||||
openclawRoot: root,
|
||||
includeArchive: true,
|
||||
bankStrategy: 'agent',
|
||||
});
|
||||
|
||||
expect(result.entries).toHaveLength(1);
|
||||
expect(result.entries[0].bankId).toBe('proj-debug');
|
||||
});
|
||||
|
||||
it('can exclude archive sessions', () => {
|
||||
const root = makeTempRoot();
|
||||
writeOpenClawConfig(root, { plugins: { entries: { 'hindsight-openclaw': { config: {} } } } });
|
||||
writeSession(root, 'main', 'live.jsonl', [
|
||||
{ type: 'session', id: 'live' },
|
||||
{ type: 'message', message: { role: 'user', content: 'live' } },
|
||||
{ type: 'message', message: { role: 'assistant', content: 'reply' } },
|
||||
]);
|
||||
writeSession(root, 'main', 'archive.jsonl', [
|
||||
{ type: 'session', id: 'archive' },
|
||||
{ type: 'message', message: { role: 'user', content: 'archived' } },
|
||||
{ type: 'message', message: { role: 'assistant', content: 'reply' } },
|
||||
], true);
|
||||
|
||||
const config = loadPluginConfigFromOpenClawRoot(root);
|
||||
const result = buildBackfillPlan(config, {
|
||||
openclawRoot: root,
|
||||
includeArchive: false,
|
||||
bankStrategy: 'mirror-config',
|
||||
});
|
||||
|
||||
expect(result.discoveredSessions).toBe(1);
|
||||
expect(result.entries).toHaveLength(1);
|
||||
expect(result.entries[0].sessionId).toBe('live');
|
||||
});
|
||||
});
|
||||
294
hindsight-integrations/openclaw/src/backfill-lib.ts
Normal file
294
hindsight-integrations/openclaw/src/backfill-lib.ts
Normal file
|
|
@ -0,0 +1,294 @@
|
|||
import { homedir } from 'os';
|
||||
import { dirname, join, resolve } from 'path';
|
||||
import { readFileSync, existsSync, mkdirSync, writeFileSync, readdirSync } from 'fs';
|
||||
import { deriveBankId, prepareRetentionTranscript } from './index.js';
|
||||
import type { PluginConfig, PluginHookAgentContext } from './types.js';
|
||||
|
||||
export interface BackfillCliOptions {
|
||||
openclawRoot: string;
|
||||
includeArchive: boolean;
|
||||
selectedAgents?: Set<string>;
|
||||
limit?: number;
|
||||
bankStrategy: 'mirror-config' | 'agent' | 'fixed';
|
||||
fixedBank?: string;
|
||||
}
|
||||
|
||||
export interface SessionMessage {
|
||||
role: 'user' | 'assistant' | 'system' | 'tool';
|
||||
content: string | Array<{ type?: string; text?: string }>;
|
||||
}
|
||||
|
||||
export interface ParsedSessionFile {
|
||||
filePath: string;
|
||||
agentId: string;
|
||||
sessionId: string;
|
||||
sessionKey?: string;
|
||||
startedAt?: string;
|
||||
messages: SessionMessage[];
|
||||
}
|
||||
|
||||
export interface BackfillPlanEntry {
|
||||
filePath: string;
|
||||
agentId: string;
|
||||
sessionId: string;
|
||||
startedAt?: string;
|
||||
bankId: string;
|
||||
documentId: string;
|
||||
transcript: string;
|
||||
messageCount: number;
|
||||
}
|
||||
|
||||
export interface BackfillCheckpointEntry {
|
||||
status: 'enqueued' | 'completed' | 'failed';
|
||||
bankId: string;
|
||||
filePath: string;
|
||||
sessionId: string;
|
||||
updatedAt: string;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface BackfillCheckpoint {
|
||||
version: 1;
|
||||
entries: Record<string, BackfillCheckpointEntry>;
|
||||
}
|
||||
|
||||
interface RawBackfillCheckpointEntry extends Omit<BackfillCheckpointEntry, 'status'> {
|
||||
status: BackfillCheckpointEntry['status'] | 'queued';
|
||||
}
|
||||
|
||||
interface RawBackfillCheckpoint {
|
||||
version: 1;
|
||||
entries: Record<string, RawBackfillCheckpointEntry>;
|
||||
}
|
||||
|
||||
interface SessionDirectory {
|
||||
agentId: string;
|
||||
path: string;
|
||||
}
|
||||
|
||||
const DEFAULT_PLUGIN_CONFIG: PluginConfig = {
|
||||
dynamicBankId: true,
|
||||
retainRoles: ['user', 'assistant'],
|
||||
};
|
||||
|
||||
export function defaultOpenClawRoot(): string {
|
||||
return resolve(join(homedir(), '.openclaw'));
|
||||
}
|
||||
|
||||
export function defaultCheckpointPath(openclawRoot: string): string {
|
||||
return join(openclawRoot, 'data', 'hindsight-backfill-checkpoint.json');
|
||||
}
|
||||
|
||||
export function loadPluginConfigFromOpenClawRoot(openclawRoot: string): PluginConfig {
|
||||
const configPath = join(openclawRoot, 'openclaw.json');
|
||||
const raw = JSON.parse(readFileSync(configPath, 'utf8')) as {
|
||||
plugins?: { entries?: Record<string, { config?: PluginConfig }> };
|
||||
};
|
||||
return {
|
||||
...DEFAULT_PLUGIN_CONFIG,
|
||||
...(raw.plugins?.entries?.['hindsight-openclaw']?.config || {}),
|
||||
};
|
||||
}
|
||||
|
||||
function extractTextContent(content: unknown): string {
|
||||
if (typeof content === 'string') {
|
||||
return content;
|
||||
}
|
||||
if (Array.isArray(content)) {
|
||||
return content
|
||||
.filter((block): block is { type?: string; text?: string } => !!block && typeof block === 'object')
|
||||
.filter((block) => block.type === 'text' && typeof block.text === 'string')
|
||||
.map((block) => block.text || '')
|
||||
.join('\n');
|
||||
}
|
||||
return '';
|
||||
}
|
||||
|
||||
function readJsonLines(filePath: string): unknown[] {
|
||||
const content = readFileSync(filePath, 'utf8');
|
||||
return content
|
||||
.split('\n')
|
||||
.map((line) => line.trim())
|
||||
.filter(Boolean)
|
||||
.map((line) => JSON.parse(line));
|
||||
}
|
||||
|
||||
export function parseSessionFile(filePath: string, agentId: string): ParsedSessionFile {
|
||||
const records = readJsonLines(filePath) as Array<Record<string, any>>;
|
||||
let sessionId = filePath.split('/').pop()?.replace(/\.jsonl$/, '') || 'session';
|
||||
let sessionKey: string | undefined;
|
||||
let startedAt: string | undefined;
|
||||
const messages: SessionMessage[] = [];
|
||||
|
||||
for (const record of records) {
|
||||
if (record.type === 'session') {
|
||||
sessionId = typeof record.id === 'string' ? record.id : sessionId;
|
||||
startedAt = typeof record.timestamp === 'string' ? record.timestamp : startedAt;
|
||||
sessionKey = typeof record.sessionKey === 'string' ? record.sessionKey : sessionKey;
|
||||
continue;
|
||||
}
|
||||
if (record.type !== 'message' || !record.message || typeof record.message !== 'object') {
|
||||
continue;
|
||||
}
|
||||
const message = record.message as Record<string, unknown>;
|
||||
const role = message.role;
|
||||
if (role !== 'user' && role !== 'assistant' && role !== 'system' && role !== 'tool') {
|
||||
continue;
|
||||
}
|
||||
const text = extractTextContent(message.content);
|
||||
if (!text.trim()) {
|
||||
continue;
|
||||
}
|
||||
messages.push({
|
||||
role,
|
||||
content: typeof message.content === 'string' ? message.content : [{ type: 'text', text }],
|
||||
});
|
||||
if (!sessionKey && typeof record.sessionKey === 'string') {
|
||||
sessionKey = record.sessionKey;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
filePath,
|
||||
agentId,
|
||||
sessionId,
|
||||
sessionKey,
|
||||
startedAt,
|
||||
messages,
|
||||
};
|
||||
}
|
||||
|
||||
function sessionDirectories(openclawRoot: string, includeArchive: boolean): SessionDirectory[] {
|
||||
const agentsRoot = join(openclawRoot, 'agents');
|
||||
if (!existsSync(agentsRoot)) {
|
||||
return [];
|
||||
}
|
||||
const result: SessionDirectory[] = [];
|
||||
for (const entry of readdirSync(agentsRoot, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
const agentId = entry.name;
|
||||
const sessionsDir = join(agentsRoot, agentId, 'sessions');
|
||||
if (existsSync(sessionsDir)) {
|
||||
result.push({ agentId, path: sessionsDir });
|
||||
}
|
||||
if (includeArchive) {
|
||||
const archiveDir = join(agentsRoot, agentId, 'sessions-archive-from-migration_backup');
|
||||
if (existsSync(archiveDir)) {
|
||||
result.push({ agentId, path: archiveDir });
|
||||
}
|
||||
}
|
||||
}
|
||||
return result.sort((a, b) => a.agentId.localeCompare(b.agentId) || a.path.localeCompare(b.path));
|
||||
}
|
||||
|
||||
export function discoverSessionFiles(openclawRoot: string, includeArchive: boolean): Array<{ agentId: string; filePath: string }> {
|
||||
const sessions: Array<{ agentId: string; filePath: string }> = [];
|
||||
for (const dir of sessionDirectories(openclawRoot, includeArchive)) {
|
||||
for (const entry of readdirSync(dir.path, { withFileTypes: true })) {
|
||||
if (!entry.isFile() || !entry.name.endsWith('.jsonl')) continue;
|
||||
sessions.push({
|
||||
agentId: dir.agentId,
|
||||
filePath: join(dir.path, entry.name),
|
||||
});
|
||||
}
|
||||
}
|
||||
return sessions.sort((a, b) => a.agentId.localeCompare(b.agentId) || a.filePath.localeCompare(b.filePath));
|
||||
}
|
||||
|
||||
function backfillContextForSession(session: ParsedSessionFile): PluginHookAgentContext {
|
||||
return {
|
||||
agentId: session.agentId,
|
||||
sessionKey: session.sessionKey,
|
||||
};
|
||||
}
|
||||
|
||||
function deriveTargetBank(
|
||||
session: ParsedSessionFile,
|
||||
pluginConfig: PluginConfig,
|
||||
bankStrategy: BackfillCliOptions['bankStrategy'],
|
||||
fixedBank?: string,
|
||||
): string {
|
||||
if (bankStrategy === 'agent') {
|
||||
return session.agentId;
|
||||
}
|
||||
if (bankStrategy === 'fixed') {
|
||||
if (!fixedBank) {
|
||||
throw new Error('fixed bank strategy requires --fixed-bank');
|
||||
}
|
||||
return fixedBank;
|
||||
}
|
||||
return deriveBankId(backfillContextForSession(session), pluginConfig);
|
||||
}
|
||||
|
||||
export function stableDocumentId(session: ParsedSessionFile, bankId: string): string {
|
||||
return `backfill::${bankId}::${session.agentId}::${session.sessionId}`;
|
||||
}
|
||||
|
||||
export function buildBackfillPlan(
|
||||
pluginConfig: PluginConfig,
|
||||
opts: BackfillCliOptions,
|
||||
): { entries: BackfillPlanEntry[]; discoveredSessions: number; skippedEmpty: number } {
|
||||
const entries: BackfillPlanEntry[] = [];
|
||||
let discoveredSessions = 0;
|
||||
let skippedEmpty = 0;
|
||||
|
||||
for (const candidate of discoverSessionFiles(opts.openclawRoot, opts.includeArchive)) {
|
||||
if (opts.selectedAgents && !opts.selectedAgents.has(candidate.agentId)) {
|
||||
continue;
|
||||
}
|
||||
discoveredSessions += 1;
|
||||
const parsed = parseSessionFile(candidate.filePath, candidate.agentId);
|
||||
const retention = prepareRetentionTranscript(parsed.messages, pluginConfig, true);
|
||||
if (!retention) {
|
||||
skippedEmpty += 1;
|
||||
continue;
|
||||
}
|
||||
const bankId = deriveTargetBank(parsed, pluginConfig, opts.bankStrategy, opts.fixedBank);
|
||||
entries.push({
|
||||
filePath: parsed.filePath,
|
||||
agentId: parsed.agentId,
|
||||
sessionId: parsed.sessionId,
|
||||
startedAt: parsed.startedAt,
|
||||
bankId,
|
||||
documentId: stableDocumentId(parsed, bankId),
|
||||
transcript: retention.transcript,
|
||||
messageCount: retention.messageCount,
|
||||
});
|
||||
if (opts.limit && entries.length >= opts.limit) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return { entries, discoveredSessions, skippedEmpty };
|
||||
}
|
||||
|
||||
export function loadCheckpoint(checkpointPath: string): BackfillCheckpoint {
|
||||
if (!existsSync(checkpointPath)) {
|
||||
return { version: 1, entries: {} };
|
||||
}
|
||||
const raw = JSON.parse(readFileSync(checkpointPath, 'utf8')) as RawBackfillCheckpoint;
|
||||
if (raw.version !== 1 || !raw.entries || typeof raw.entries !== 'object') {
|
||||
return { version: 1, entries: {} };
|
||||
}
|
||||
return {
|
||||
version: 1,
|
||||
entries: Object.fromEntries(
|
||||
Object.entries(raw.entries).map(([key, entry]) => [
|
||||
key,
|
||||
{
|
||||
...entry,
|
||||
status: entry.status === 'queued' ? 'enqueued' : entry.status,
|
||||
},
|
||||
]),
|
||||
) as Record<string, BackfillCheckpointEntry>,
|
||||
};
|
||||
}
|
||||
|
||||
export function saveCheckpoint(checkpointPath: string, checkpoint: BackfillCheckpoint): void {
|
||||
mkdirSync(dirname(checkpointPath), { recursive: true });
|
||||
writeFileSync(checkpointPath, JSON.stringify(checkpoint, null, 2) + '\n', 'utf8');
|
||||
}
|
||||
|
||||
export function checkpointKey(entry: Pick<BackfillPlanEntry, 'bankId' | 'documentId'>): string {
|
||||
return `${entry.bankId}::${entry.documentId}`;
|
||||
}
|
||||
151
hindsight-integrations/openclaw/src/backfill.test.ts
Normal file
151
hindsight-integrations/openclaw/src/backfill.test.ts
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import { mkdtempSync, symlinkSync, writeFileSync } from 'fs';
|
||||
import { join } from 'path';
|
||||
import { tmpdir } from 'os';
|
||||
import { pathToFileURL } from 'url';
|
||||
import type { BankStats, PluginConfig } from './types.js';
|
||||
import type { BackfillCheckpoint, BackfillPlanEntry } from './backfill-lib.js';
|
||||
|
||||
const managerStart = vi.fn();
|
||||
const managerStop = vi.fn();
|
||||
const managerGetBaseUrl = vi.fn(() => 'http://127.0.0.1:9077');
|
||||
|
||||
vi.mock('./embed-manager.js', () => ({
|
||||
HindsightEmbedManager: vi.fn(class {
|
||||
start = managerStart;
|
||||
stop = managerStop;
|
||||
getBaseUrl = managerGetBaseUrl;
|
||||
}),
|
||||
}));
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
vi.unstubAllGlobals();
|
||||
managerStart.mockReset();
|
||||
managerStop.mockReset();
|
||||
managerGetBaseUrl.mockClear();
|
||||
});
|
||||
|
||||
function makeEntry(bankId: string, sessionId: string): BackfillPlanEntry {
|
||||
return {
|
||||
filePath: `/tmp/${sessionId}.jsonl`,
|
||||
agentId: bankId,
|
||||
sessionId,
|
||||
bankId,
|
||||
documentId: `backfill::${bankId}::${sessionId}`,
|
||||
transcript: '[role: user]\nhello\n[user:end]',
|
||||
messageCount: 1,
|
||||
};
|
||||
}
|
||||
|
||||
function makeStats(overrides: Partial<BankStats> = {}): BankStats {
|
||||
return {
|
||||
bank_id: 'bank',
|
||||
total_nodes: 0,
|
||||
total_links: 0,
|
||||
total_documents: 0,
|
||||
pending_operations: 0,
|
||||
failed_operations: 0,
|
||||
pending_consolidation: 0,
|
||||
last_consolidated_at: null,
|
||||
total_observations: 0,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe('backfill helpers', () => {
|
||||
it('resume skips only completed entries', async () => {
|
||||
const { filterEntriesForResume, splitResumeEntries } = await import('./backfill.js');
|
||||
const entries = [makeEntry('bank-a', '1'), makeEntry('bank-a', '2'), makeEntry('bank-a', '3')];
|
||||
const checkpoint: BackfillCheckpoint = {
|
||||
version: 1,
|
||||
entries: {
|
||||
'bank-a::backfill::bank-a::1': { status: 'completed', bankId: 'bank-a', filePath: '/tmp/1', sessionId: '1', updatedAt: 'now' },
|
||||
'bank-a::backfill::bank-a::2': { status: 'enqueued', bankId: 'bank-a', filePath: '/tmp/2', sessionId: '2', updatedAt: 'now' },
|
||||
'bank-a::backfill::bank-a::3': { status: 'failed', bankId: 'bank-a', filePath: '/tmp/3', sessionId: '3', updatedAt: 'now' },
|
||||
},
|
||||
};
|
||||
const resumable = filterEntriesForResume(entries, checkpoint, true);
|
||||
expect(resumable.map((entry) => entry.sessionId)).toEqual(['2', '3']);
|
||||
expect(splitResumeEntries(resumable, checkpoint, false).entriesToEnqueue.map((entry) => entry.sessionId)).toEqual(['2', '3']);
|
||||
expect(splitResumeEntries(resumable, checkpoint, true)).toEqual({
|
||||
entriesToEnqueue: [entries[2]],
|
||||
alreadyEnqueuedKeys: ['bank-a::backfill::bank-a::2'],
|
||||
});
|
||||
});
|
||||
|
||||
it('normalizes legacy queued checkpoint entries', async () => {
|
||||
const { loadCheckpoint } = await import('./backfill-lib.js');
|
||||
const dir = mkdtempSync(join(tmpdir(), 'hindsight-backfill-'));
|
||||
const checkpointPath = join(dir, 'checkpoint.json');
|
||||
writeFileSync(checkpointPath, JSON.stringify({
|
||||
version: 1,
|
||||
entries: {
|
||||
legacy: { status: 'queued', bankId: 'bank-a', filePath: '/tmp/a', sessionId: 'a', updatedAt: 'now' },
|
||||
},
|
||||
}), 'utf8');
|
||||
|
||||
const checkpoint = loadCheckpoint(checkpointPath);
|
||||
expect(checkpoint.entries.legacy.status).toBe('enqueued');
|
||||
});
|
||||
|
||||
it('marks drained entries completed and leaves aggregate-failure banks enqueued', async () => {
|
||||
const { applyDrainResults } = await import('./backfill.js');
|
||||
const checkpoint: BackfillCheckpoint = {
|
||||
version: 1,
|
||||
entries: {
|
||||
a: { status: 'enqueued', bankId: 'bank-a', filePath: '/tmp/a', sessionId: 'a', updatedAt: 'now' },
|
||||
b: { status: 'enqueued', bankId: 'bank-b', filePath: '/tmp/b', sessionId: 'b', updatedAt: 'now' },
|
||||
},
|
||||
};
|
||||
const touchedEntriesByBank = new Map([
|
||||
['bank-a', ['a']],
|
||||
['bank-b', ['b']],
|
||||
]);
|
||||
const finalStatsByBank = new Map<string, BankStats>([
|
||||
['bank-a', makeStats({ bank_id: 'bank-a', pending_operations: 0, failed_operations: 0 })],
|
||||
['bank-b', makeStats({ bank_id: 'bank-b', pending_operations: 0, failed_operations: 2 })],
|
||||
]);
|
||||
const initialFailedByBank = new Map([
|
||||
['bank-a', 0],
|
||||
['bank-b', 0],
|
||||
]);
|
||||
|
||||
const result = applyDrainResults(checkpoint, touchedEntriesByBank, finalStatsByBank, initialFailedByBank);
|
||||
expect(result.completed).toBe(1);
|
||||
expect(result.unresolved).toBe(1);
|
||||
expect(result.warnings).toEqual([
|
||||
'bank bank-b reported 2 new failed operations during drain; leaving 1 checkpoint entries enqueued',
|
||||
]);
|
||||
expect(checkpoint.entries.a.status).toBe('completed');
|
||||
expect(checkpoint.entries.b.status).toBe('enqueued');
|
||||
});
|
||||
|
||||
it('starts local daemon when no external API is configured and health check fails', async () => {
|
||||
const fetchMock = vi.fn().mockRejectedValue(new Error('offline'));
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
const { createBackfillRuntime } = await import('./backfill.js');
|
||||
const pluginConfig: PluginConfig = {
|
||||
apiPort: 9077,
|
||||
llmProvider: 'openai-codex',
|
||||
llmModel: 'gpt-5.4',
|
||||
};
|
||||
const runtime = await createBackfillRuntime(pluginConfig);
|
||||
expect(managerStart).toHaveBeenCalledTimes(1);
|
||||
expect(runtime.apiUrl).toBe('http://127.0.0.1:9077');
|
||||
await runtime.stop();
|
||||
expect(managerStop).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('treats a symlinked bin path as direct execution', async () => {
|
||||
const { isDirectExecution } = await import('./backfill.js');
|
||||
const dir = mkdtempSync(join(tmpdir(), 'hindsight-backfill-bin-'));
|
||||
const modulePath = join(process.cwd(), 'dist', 'backfill.js');
|
||||
const symlinkPath = join(dir, 'hindsight-openclaw-backfill');
|
||||
symlinkSync(modulePath, symlinkPath);
|
||||
|
||||
const moduleUrl = pathToFileURL(modulePath).href;
|
||||
expect(isDirectExecution(symlinkPath, moduleUrl)).toBe(true);
|
||||
expect(isDirectExecution(join(dir, 'other-entrypoint'), moduleUrl)).toBe(false);
|
||||
});
|
||||
});
|
||||
561
hindsight-integrations/openclaw/src/backfill.ts
Normal file
561
hindsight-integrations/openclaw/src/backfill.ts
Normal file
|
|
@ -0,0 +1,561 @@
|
|||
#!/usr/bin/env node
|
||||
import { existsSync, realpathSync } from 'fs';
|
||||
import { join, resolve } from 'path';
|
||||
import { fileURLToPath, pathToFileURL } from 'url';
|
||||
import { HindsightEmbedManager } from './embed-manager.js';
|
||||
import { HindsightClient } from './client.js';
|
||||
import { buildClientOptions, detectExternalApi, detectLLMConfig } from './index.js';
|
||||
import type { BankStats, PluginConfig } from './types.js';
|
||||
import {
|
||||
buildBackfillPlan,
|
||||
checkpointKey,
|
||||
defaultCheckpointPath,
|
||||
defaultOpenClawRoot,
|
||||
loadCheckpoint,
|
||||
loadPluginConfigFromOpenClawRoot,
|
||||
saveCheckpoint,
|
||||
type BackfillCheckpoint,
|
||||
type BackfillPlanEntry,
|
||||
type BackfillCliOptions,
|
||||
} from './backfill-lib.js';
|
||||
|
||||
interface ParsedArgs {
|
||||
openclawRoot: string;
|
||||
profile: string;
|
||||
agents: string[];
|
||||
includeArchive: boolean;
|
||||
limit?: number;
|
||||
dryRun: boolean;
|
||||
json: boolean;
|
||||
resume: boolean;
|
||||
checkpointPath: string;
|
||||
bankStrategy: 'mirror-config' | 'agent' | 'fixed';
|
||||
fixedBank?: string;
|
||||
apiUrl?: string;
|
||||
apiToken?: string;
|
||||
maxPendingOperations?: number;
|
||||
waitUntilDrained: boolean;
|
||||
}
|
||||
|
||||
interface BackfillRuntime {
|
||||
apiUrl: string;
|
||||
apiToken?: string;
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
interface BankRuntime {
|
||||
client: HindsightClient;
|
||||
touchedEntryKeys: string[];
|
||||
initialFailedOperations: number;
|
||||
missionApplied: boolean;
|
||||
}
|
||||
|
||||
function usage(): string {
|
||||
return [
|
||||
'Usage: hindsight-openclaw-backfill [options]',
|
||||
'',
|
||||
'Options:',
|
||||
' --openclaw-root <path> OpenClaw root directory (default: ~/.openclaw)',
|
||||
' --profile <name> Logical profile name for reporting (default: openclaw)',
|
||||
' --agent <id> Restrict import to a specific agent (repeatable)',
|
||||
' --include-archive Include migration archives (default)',
|
||||
' --exclude-archive Exclude migration archives',
|
||||
' --limit <n> Stop after enqueueing N sessions',
|
||||
' --dry-run Build and print the import plan without enqueueing',
|
||||
' --json Print final summary as JSON',
|
||||
' --resume Skip entries already marked completed in the checkpoint',
|
||||
' --checkpoint <path> Path to checkpoint JSON',
|
||||
' --bank-strategy <mode> mirror-config | agent | fixed',
|
||||
' --fixed-bank <id> Required when bank strategy is fixed',
|
||||
' --api-url <url> Hindsight API base URL override',
|
||||
' --api-token <token> Hindsight API bearer token override',
|
||||
' --max-pending-operations <n> Wait until target bank queue is <= n before enqueueing',
|
||||
' --wait-until-drained Wait for touched banks to drain and finalize checkpoint state',
|
||||
' -h, --help Show this help',
|
||||
].join('\n');
|
||||
}
|
||||
|
||||
function parseArgs(argv: string[]): ParsedArgs {
|
||||
const args: ParsedArgs = {
|
||||
openclawRoot: defaultOpenClawRoot(),
|
||||
profile: 'openclaw',
|
||||
agents: [],
|
||||
includeArchive: true,
|
||||
dryRun: false,
|
||||
json: false,
|
||||
resume: false,
|
||||
checkpointPath: '',
|
||||
bankStrategy: 'mirror-config',
|
||||
waitUntilDrained: false,
|
||||
};
|
||||
|
||||
for (let i = 0; i < argv.length; i++) {
|
||||
const arg = argv[i];
|
||||
const next = () => {
|
||||
const value = argv[++i];
|
||||
if (!value) {
|
||||
throw new Error(`missing value for ${arg}`);
|
||||
}
|
||||
return value;
|
||||
};
|
||||
|
||||
switch (arg) {
|
||||
case '--openclaw-root':
|
||||
args.openclawRoot = resolve(next());
|
||||
break;
|
||||
case '--profile':
|
||||
args.profile = next();
|
||||
break;
|
||||
case '--agent':
|
||||
args.agents.push(next());
|
||||
break;
|
||||
case '--include-archive':
|
||||
args.includeArchive = true;
|
||||
break;
|
||||
case '--exclude-archive':
|
||||
args.includeArchive = false;
|
||||
break;
|
||||
case '--limit':
|
||||
args.limit = Number(next());
|
||||
break;
|
||||
case '--dry-run':
|
||||
args.dryRun = true;
|
||||
break;
|
||||
case '--json':
|
||||
args.json = true;
|
||||
break;
|
||||
case '--resume':
|
||||
args.resume = true;
|
||||
break;
|
||||
case '--checkpoint':
|
||||
args.checkpointPath = resolve(next());
|
||||
break;
|
||||
case '--bank-strategy': {
|
||||
const value = next();
|
||||
if (value !== 'mirror-config' && value !== 'agent' && value !== 'fixed') {
|
||||
throw new Error(`invalid bank strategy: ${value}`);
|
||||
}
|
||||
args.bankStrategy = value;
|
||||
break;
|
||||
}
|
||||
case '--fixed-bank':
|
||||
args.fixedBank = next();
|
||||
break;
|
||||
case '--api-url':
|
||||
args.apiUrl = next();
|
||||
break;
|
||||
case '--api-token':
|
||||
args.apiToken = next();
|
||||
break;
|
||||
case '--max-pending-operations':
|
||||
args.maxPendingOperations = Number(next());
|
||||
break;
|
||||
case '--wait-until-drained':
|
||||
args.waitUntilDrained = true;
|
||||
break;
|
||||
case '-h':
|
||||
case '--help':
|
||||
console.log(usage());
|
||||
process.exit(0);
|
||||
default:
|
||||
throw new Error(`unknown argument: ${arg}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (!args.checkpointPath) {
|
||||
args.checkpointPath = defaultCheckpointPath(args.openclawRoot);
|
||||
}
|
||||
if (args.bankStrategy === 'fixed' && !args.fixedBank) {
|
||||
throw new Error('--fixed-bank is required when --bank-strategy fixed is used');
|
||||
}
|
||||
return args;
|
||||
}
|
||||
|
||||
function inferApiSettings(pluginConfig: PluginConfig, explicitApiUrl?: string, explicitApiToken?: string): { apiUrl: string; apiToken?: string } {
|
||||
const apiUrl = explicitApiUrl
|
||||
|| process.env.HINDSIGHT_EMBED_API_URL
|
||||
|| pluginConfig.hindsightApiUrl
|
||||
|| `http://127.0.0.1:${pluginConfig.apiPort || 9077}`;
|
||||
const apiToken = explicitApiToken
|
||||
|| process.env.HINDSIGHT_EMBED_API_TOKEN
|
||||
|| pluginConfig.hindsightApiToken;
|
||||
return { apiUrl, apiToken: apiToken || undefined };
|
||||
}
|
||||
|
||||
async function checkHealth(apiUrl: string, apiToken?: string): Promise<boolean> {
|
||||
try {
|
||||
const response = await fetch(`${apiUrl.replace(/\/$/, '')}/health`, {
|
||||
method: 'GET',
|
||||
headers: apiToken ? { Authorization: `Bearer ${apiToken}` } : undefined,
|
||||
signal: AbortSignal.timeout(5000),
|
||||
});
|
||||
return response.ok;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function filterEntriesForResume(entries: BackfillPlanEntry[], checkpoint: BackfillCheckpoint, resume: boolean): BackfillPlanEntry[] {
|
||||
if (!resume) {
|
||||
return entries;
|
||||
}
|
||||
return entries.filter((entry) => checkpoint.entries[checkpointKey(entry)]?.status !== 'completed');
|
||||
}
|
||||
|
||||
export function splitResumeEntries(
|
||||
entries: BackfillPlanEntry[],
|
||||
checkpoint: BackfillCheckpoint,
|
||||
waitUntilDrained: boolean,
|
||||
): { entriesToEnqueue: BackfillPlanEntry[]; alreadyEnqueuedKeys: string[] } {
|
||||
const entriesToEnqueue: BackfillPlanEntry[] = [];
|
||||
const alreadyEnqueuedKeys: string[] = [];
|
||||
for (const entry of entries) {
|
||||
const status = checkpoint.entries[checkpointKey(entry)]?.status;
|
||||
if (status === 'enqueued') {
|
||||
if (waitUntilDrained) {
|
||||
alreadyEnqueuedKeys.push(checkpointKey(entry));
|
||||
} else {
|
||||
entriesToEnqueue.push(entry);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
entriesToEnqueue.push(entry);
|
||||
}
|
||||
return { entriesToEnqueue, alreadyEnqueuedKeys };
|
||||
}
|
||||
|
||||
export function applyDrainResults(
|
||||
checkpoint: BackfillCheckpoint,
|
||||
touchedEntriesByBank: Map<string, string[]>,
|
||||
finalStatsByBank: Map<string, BankStats>,
|
||||
initialFailedOperationsByBank: Map<string, number>,
|
||||
): { completed: number; unresolved: number; warnings: string[] } {
|
||||
let completed = 0;
|
||||
let unresolved = 0;
|
||||
const warnings: string[] = [];
|
||||
|
||||
for (const [bankId, entryKeys] of touchedEntriesByBank.entries()) {
|
||||
const stats = finalStatsByBank.get(bankId);
|
||||
const initialFailed = initialFailedOperationsByBank.get(bankId) ?? 0;
|
||||
const hasNewFailures = !!stats && stats.failed_operations > initialFailed;
|
||||
|
||||
if (hasNewFailures) {
|
||||
warnings.push(
|
||||
`bank ${bankId} reported ${stats!.failed_operations - initialFailed} new failed operations during drain; leaving ${entryKeys.length} checkpoint entries enqueued`,
|
||||
);
|
||||
} else if (!stats || stats.pending_operations > 0) {
|
||||
warnings.push(
|
||||
`bank ${bankId} did not finish draining cleanly; leaving ${entryKeys.length} checkpoint entries enqueued`,
|
||||
);
|
||||
}
|
||||
|
||||
for (const entryKey of entryKeys) {
|
||||
const existing = checkpoint.entries[entryKey];
|
||||
if (!existing || existing.status !== 'enqueued') {
|
||||
continue;
|
||||
}
|
||||
if (!hasNewFailures && stats && stats.pending_operations === 0) {
|
||||
checkpoint.entries[entryKey] = {
|
||||
...existing,
|
||||
status: 'completed',
|
||||
updatedAt: new Date().toISOString(),
|
||||
error: undefined,
|
||||
};
|
||||
completed += 1;
|
||||
} else {
|
||||
unresolved += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return { completed, unresolved, warnings };
|
||||
}
|
||||
|
||||
export async function createBackfillRuntime(
|
||||
pluginConfig: PluginConfig,
|
||||
explicitApiUrl?: string,
|
||||
explicitApiToken?: string,
|
||||
): Promise<BackfillRuntime> {
|
||||
const explicit = inferApiSettings(pluginConfig, explicitApiUrl, explicitApiToken);
|
||||
const externalApi = detectExternalApi(pluginConfig);
|
||||
const useExternalApi = !!(explicitApiUrl || explicitApiToken || externalApi.apiUrl || pluginConfig.hindsightApiUrl);
|
||||
|
||||
if (useExternalApi) {
|
||||
return {
|
||||
apiUrl: explicit.apiUrl,
|
||||
apiToken: explicit.apiToken,
|
||||
async stop() {},
|
||||
};
|
||||
}
|
||||
|
||||
if (await checkHealth(explicit.apiUrl, explicit.apiToken)) {
|
||||
return {
|
||||
apiUrl: explicit.apiUrl,
|
||||
apiToken: explicit.apiToken,
|
||||
async stop() {},
|
||||
};
|
||||
}
|
||||
|
||||
const llmConfig = detectLLMConfig(pluginConfig);
|
||||
const manager = new HindsightEmbedManager(
|
||||
pluginConfig.apiPort || 9077,
|
||||
llmConfig.provider || '',
|
||||
llmConfig.apiKey || '',
|
||||
llmConfig.model,
|
||||
llmConfig.baseUrl,
|
||||
pluginConfig.daemonIdleTimeout ?? 0,
|
||||
pluginConfig.embedVersion,
|
||||
pluginConfig.embedPackagePath,
|
||||
);
|
||||
await manager.start();
|
||||
|
||||
return {
|
||||
apiUrl: manager.getBaseUrl(),
|
||||
apiToken: undefined,
|
||||
async stop() {
|
||||
await manager.stop();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function waitForBankQueue(client: HindsightClient, maxPendingOperations: number): Promise<void> {
|
||||
for (;;) {
|
||||
try {
|
||||
const stats = await client.getBankStats();
|
||||
if (stats.pending_operations <= maxPendingOperations) {
|
||||
return;
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof Error && error.message.includes('HTTP 404')) {
|
||||
return;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 3000));
|
||||
}
|
||||
}
|
||||
|
||||
async function getInitialBankStats(client: HindsightClient): Promise<BankStats | null> {
|
||||
try {
|
||||
return await client.getBankStats();
|
||||
} catch (error) {
|
||||
if (error instanceof Error && error.message.includes('HTTP 404')) {
|
||||
return null;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForBanksToDrain(clientsByBankId: Map<string, HindsightClient>): Promise<Map<string, BankStats>> {
|
||||
for (;;) {
|
||||
const stats = await Promise.all(
|
||||
Array.from(clientsByBankId.entries()).map(async ([bankId, client]) => ({ bankId, stats: await client.getBankStats() })),
|
||||
);
|
||||
const statsByBank = new Map(stats.map(({ bankId, stats: bankStats }) => [bankId, bankStats]));
|
||||
const pending = stats.filter(({ stats: bankStats }) => bankStats.pending_operations > 0);
|
||||
if (pending.length === 0) {
|
||||
return statsByBank;
|
||||
}
|
||||
console.log(
|
||||
pending
|
||||
.map(({ bankId, stats: bankStats }) => `${bankId}\tpending_operations=${bankStats.pending_operations}\tfailed_operations=${bankStats.failed_operations}\tpending_consolidation=${bankStats.pending_consolidation}`)
|
||||
.join('\n'),
|
||||
);
|
||||
await new Promise((resolve) => setTimeout(resolve, 5000));
|
||||
}
|
||||
}
|
||||
|
||||
export async function runCli(argv: string[] = process.argv.slice(2)): Promise<void> {
|
||||
const args = parseArgs(argv);
|
||||
if (!existsSync(join(args.openclawRoot, 'openclaw.json'))) {
|
||||
throw new Error(`could not find openclaw.json under ${args.openclawRoot}`);
|
||||
}
|
||||
|
||||
const pluginConfig = loadPluginConfigFromOpenClawRoot(args.openclawRoot);
|
||||
const backfillOptions: BackfillCliOptions = {
|
||||
openclawRoot: args.openclawRoot,
|
||||
includeArchive: args.includeArchive,
|
||||
selectedAgents: args.agents.length ? new Set(args.agents) : undefined,
|
||||
limit: args.limit,
|
||||
bankStrategy: args.bankStrategy,
|
||||
fixedBank: args.fixedBank,
|
||||
};
|
||||
const checkpoint = loadCheckpoint(args.checkpointPath);
|
||||
const { entries, discoveredSessions, skippedEmpty } = buildBackfillPlan(pluginConfig, backfillOptions);
|
||||
const plannedEntries = filterEntriesForResume(entries, checkpoint, args.resume);
|
||||
const { entriesToEnqueue, alreadyEnqueuedKeys } = splitResumeEntries(plannedEntries, checkpoint, args.waitUntilDrained);
|
||||
|
||||
if (args.dryRun) {
|
||||
for (const entry of plannedEntries) {
|
||||
console.log(`${entry.agentId}\t${entry.bankId}\t${entry.sessionId}\tmsgs=${entry.messageCount}\tchars=${entry.transcript.length}`);
|
||||
}
|
||||
const summary = {
|
||||
profile: args.profile,
|
||||
dry_run: true,
|
||||
discovered_sessions: discoveredSessions,
|
||||
planned_sessions: plannedEntries.length,
|
||||
skipped_empty: skippedEmpty,
|
||||
bank_strategy: args.bankStrategy,
|
||||
checkpoint_path: args.checkpointPath,
|
||||
};
|
||||
console.log(args.json ? JSON.stringify(summary, null, 2) : JSON.stringify(summary));
|
||||
return;
|
||||
}
|
||||
|
||||
const llmConfig = detectLLMConfig(pluginConfig);
|
||||
const runtime = await createBackfillRuntime(pluginConfig, args.apiUrl, args.apiToken);
|
||||
const clientsByBankId = new Map<string, BankRuntime>();
|
||||
let imported = 0;
|
||||
let failed = 0;
|
||||
let finalized = 0;
|
||||
|
||||
try {
|
||||
for (const entryKey of alreadyEnqueuedKeys) {
|
||||
const checkpointEntry = checkpoint.entries[entryKey];
|
||||
if (!checkpointEntry) continue;
|
||||
let bankRuntime = clientsByBankId.get(checkpointEntry.bankId);
|
||||
if (!bankRuntime) {
|
||||
const client = new HindsightClient({
|
||||
...buildClientOptions(llmConfig, pluginConfig, { apiUrl: runtime.apiUrl, apiToken: runtime.apiToken ?? null }),
|
||||
apiUrl: runtime.apiUrl,
|
||||
apiToken: runtime.apiToken,
|
||||
});
|
||||
client.setBankId(checkpointEntry.bankId);
|
||||
bankRuntime = {
|
||||
client,
|
||||
touchedEntryKeys: [],
|
||||
initialFailedOperations: (await getInitialBankStats(client))?.failed_operations ?? 0,
|
||||
missionApplied: false,
|
||||
};
|
||||
clientsByBankId.set(checkpointEntry.bankId, bankRuntime);
|
||||
}
|
||||
bankRuntime.touchedEntryKeys.push(entryKey);
|
||||
}
|
||||
|
||||
for (const entry of entriesToEnqueue) {
|
||||
let bankRuntime = clientsByBankId.get(entry.bankId);
|
||||
if (!bankRuntime) {
|
||||
const client = new HindsightClient({
|
||||
...buildClientOptions(llmConfig, pluginConfig, { apiUrl: runtime.apiUrl, apiToken: runtime.apiToken ?? null }),
|
||||
apiUrl: runtime.apiUrl,
|
||||
apiToken: runtime.apiToken,
|
||||
});
|
||||
client.setBankId(entry.bankId);
|
||||
bankRuntime = {
|
||||
client,
|
||||
touchedEntryKeys: [],
|
||||
initialFailedOperations: (await getInitialBankStats(client))?.failed_operations ?? 0,
|
||||
missionApplied: false,
|
||||
};
|
||||
clientsByBankId.set(entry.bankId, bankRuntime);
|
||||
}
|
||||
const client = bankRuntime.client;
|
||||
|
||||
if (!bankRuntime.missionApplied && pluginConfig.bankMission) {
|
||||
await client.setBankMission(pluginConfig.bankMission);
|
||||
}
|
||||
|
||||
if (typeof args.maxPendingOperations === 'number' && args.maxPendingOperations >= 0) {
|
||||
await waitForBankQueue(client, args.maxPendingOperations);
|
||||
}
|
||||
|
||||
try {
|
||||
const metadata: Record<string, string> = {
|
||||
source: 'openclaw-backfill',
|
||||
file_path: entry.filePath,
|
||||
agent_id: entry.agentId,
|
||||
session_id: entry.sessionId,
|
||||
retained_at: new Date().toISOString(),
|
||||
};
|
||||
if (entry.startedAt) {
|
||||
metadata.session_started_at = entry.startedAt;
|
||||
}
|
||||
await client.retain({
|
||||
content: entry.transcript,
|
||||
document_id: entry.documentId,
|
||||
metadata,
|
||||
});
|
||||
checkpoint.entries[checkpointKey(entry)] = {
|
||||
status: 'enqueued',
|
||||
bankId: entry.bankId,
|
||||
filePath: entry.filePath,
|
||||
sessionId: entry.sessionId,
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
bankRuntime.touchedEntryKeys.push(checkpointKey(entry));
|
||||
if (!bankRuntime.missionApplied && pluginConfig.bankMission) {
|
||||
await client.setBankMission(pluginConfig.bankMission);
|
||||
bankRuntime.missionApplied = true;
|
||||
}
|
||||
saveCheckpoint(args.checkpointPath, checkpoint);
|
||||
console.log(`${entry.agentId}\t${entry.bankId}\t${entry.sessionId}\tenqueued`);
|
||||
imported += 1;
|
||||
} catch (error) {
|
||||
checkpoint.entries[checkpointKey(entry)] = {
|
||||
status: 'failed',
|
||||
bankId: entry.bankId,
|
||||
filePath: entry.filePath,
|
||||
sessionId: entry.sessionId,
|
||||
updatedAt: new Date().toISOString(),
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
};
|
||||
saveCheckpoint(args.checkpointPath, checkpoint);
|
||||
failed += 1;
|
||||
console.error(`${entry.agentId}\t${entry.bankId}\t${entry.sessionId}\tfailed\t${error instanceof Error ? error.message : String(error)}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (args.waitUntilDrained && clientsByBankId.size > 0) {
|
||||
const finalStatsByBank = await waitForBanksToDrain(
|
||||
new Map(Array.from(clientsByBankId.entries()).map(([bankId, value]) => [bankId, value.client])),
|
||||
);
|
||||
const touchedEntriesByBank = new Map(Array.from(clientsByBankId.entries()).map(([bankId, value]) => [bankId, value.touchedEntryKeys]));
|
||||
const initialFailedByBank = new Map(Array.from(clientsByBankId.entries()).map(([bankId, value]) => [bankId, value.initialFailedOperations]));
|
||||
const finalization = applyDrainResults(checkpoint, touchedEntriesByBank, finalStatsByBank, initialFailedByBank);
|
||||
finalized = finalization.completed;
|
||||
for (const warning of finalization.warnings) {
|
||||
console.warn(warning);
|
||||
}
|
||||
saveCheckpoint(args.checkpointPath, checkpoint);
|
||||
}
|
||||
} finally {
|
||||
await runtime.stop();
|
||||
}
|
||||
|
||||
const summary = {
|
||||
profile: args.profile,
|
||||
api_url: runtime.apiUrl,
|
||||
discovered_sessions: discoveredSessions,
|
||||
planned_sessions: plannedEntries.length,
|
||||
imported_sessions: imported,
|
||||
finalized_sessions: finalized,
|
||||
failed_sessions: failed,
|
||||
skipped_empty: skippedEmpty,
|
||||
bank_strategy: args.bankStrategy,
|
||||
checkpoint_path: args.checkpointPath,
|
||||
};
|
||||
console.log(args.json ? JSON.stringify(summary, null, 2) : JSON.stringify(summary));
|
||||
}
|
||||
|
||||
function canonicalizeExecutionPath(path: string): string {
|
||||
const resolved = resolve(path);
|
||||
try {
|
||||
return realpathSync(resolved);
|
||||
} catch {
|
||||
return resolved;
|
||||
}
|
||||
}
|
||||
|
||||
export function isDirectExecution(entrypoint: string | undefined = process.argv[1], moduleUrl: string = import.meta.url): boolean {
|
||||
if (!entrypoint) {
|
||||
return false;
|
||||
}
|
||||
return canonicalizeExecutionPath(entrypoint) === canonicalizeExecutionPath(fileURLToPath(moduleUrl));
|
||||
}
|
||||
|
||||
if (isDirectExecution()) {
|
||||
runCli().catch((error) => {
|
||||
console.error(error instanceof Error ? error.message : String(error));
|
||||
process.exit(1);
|
||||
});
|
||||
}
|
||||
|
|
@ -1,6 +1,10 @@
|
|||
import { describe, it, expect } from 'vitest';
|
||||
import { afterEach, describe, it, expect, vi } from 'vitest';
|
||||
import { HindsightClient } from './client.js';
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe('HindsightClient', () => {
|
||||
it('should create instance with model', () => {
|
||||
const client = new HindsightClient({ llmModel: 'gpt-4' });
|
||||
|
|
@ -24,4 +28,36 @@ describe('HindsightClient', () => {
|
|||
});
|
||||
expect(client).toBeInstanceOf(HindsightClient);
|
||||
});
|
||||
|
||||
it('should ensure bank mission in HTTP mode', async () => {
|
||||
const fetchMock = vi.fn().mockResolvedValue({
|
||||
ok: true,
|
||||
text: async () => '',
|
||||
});
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
const client = new HindsightClient({
|
||||
apiUrl: 'https://api.example.com/',
|
||||
apiToken: 'bearer-token',
|
||||
});
|
||||
client.setBankId('demo');
|
||||
await expect(client.ensureBankMission('mission')).resolves.toBeUndefined();
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
'https://api.example.com/v1/default/banks/demo',
|
||||
expect.objectContaining({ method: 'PUT' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should throw when strict bank mission setup fails in HTTP mode', async () => {
|
||||
const fetchMock = vi.fn().mockResolvedValue({
|
||||
ok: false,
|
||||
status: 500,
|
||||
text: async () => 'boom',
|
||||
});
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
const client = new HindsightClient({
|
||||
apiUrl: 'https://api.example.com/',
|
||||
});
|
||||
client.setBankId('demo');
|
||||
await expect(client.ensureBankMission('mission')).rejects.toThrow('Failed to set bank mission');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import type {
|
|||
RetainResponse,
|
||||
RecallRequest,
|
||||
RecallResponse,
|
||||
BankStats,
|
||||
} from './types.js';
|
||||
import * as log from './logger.js';
|
||||
|
||||
|
|
@ -94,6 +95,16 @@ export class HindsightClient {
|
|||
return this.setBankMissionSubprocess(mission);
|
||||
}
|
||||
|
||||
async ensureBankMission(mission: string): Promise<void> {
|
||||
if (!mission || mission.trim().length === 0) {
|
||||
return;
|
||||
}
|
||||
if (this.httpMode) {
|
||||
return this.ensureBankMissionHttp(mission);
|
||||
}
|
||||
return this.ensureBankMissionSubprocess(mission);
|
||||
}
|
||||
|
||||
private async setBankMissionHttp(mission: string): Promise<void> {
|
||||
try {
|
||||
const url = `${this.apiUrl}/v1/default/banks/${encodeURIComponent(this.bankId)}`;
|
||||
|
|
@ -125,6 +136,28 @@ export class HindsightClient {
|
|||
}
|
||||
}
|
||||
|
||||
private async ensureBankMissionHttp(mission: string): Promise<void> {
|
||||
const url = `${this.apiUrl}/v1/default/banks/${encodeURIComponent(this.bankId)}`;
|
||||
const res = await fetch(url, {
|
||||
method: 'PUT',
|
||||
headers: this.httpHeaders(),
|
||||
body: JSON.stringify({ mission }),
|
||||
signal: AbortSignal.timeout(DEFAULT_TIMEOUT_MS),
|
||||
});
|
||||
if (!res.ok) {
|
||||
const body = await res.text().catch(() => '');
|
||||
throw new Error(`Failed to set bank mission (HTTP ${res.status}): ${body}`);
|
||||
}
|
||||
log.verbose('bank mission ensured via HTTP');
|
||||
}
|
||||
|
||||
private async ensureBankMissionSubprocess(mission: string): Promise<void> {
|
||||
const [cmd, ...baseArgs] = this.getEmbedCommand();
|
||||
const args = [...baseArgs, '--profile', 'openclaw', 'bank', 'mission', this.bankId, sanitize(mission)];
|
||||
const { stdout } = await execFileAsync(cmd, args, { maxBuffer: MAX_BUFFER });
|
||||
log.verbose(`bank mission ensured: ${stdout.trim()}`);
|
||||
}
|
||||
|
||||
// --- retain ---
|
||||
|
||||
async retain(request: RetainRequest): Promise<RetainResponse> {
|
||||
|
|
@ -259,4 +292,21 @@ export class HindsightClient {
|
|||
throw new Error(`Failed to recall memories: ${error}`, { cause: error });
|
||||
}
|
||||
}
|
||||
|
||||
async getBankStats(): Promise<BankStats> {
|
||||
if (!this.httpMode) {
|
||||
throw new Error('Bank stats are only available in HTTP mode');
|
||||
}
|
||||
const url = `${this.apiUrl}/v1/default/banks/${encodeURIComponent(this.bankId)}/stats`;
|
||||
const res = await fetch(url, {
|
||||
method: 'GET',
|
||||
headers: this.httpHeaders(),
|
||||
signal: AbortSignal.timeout(DEFAULT_TIMEOUT_MS),
|
||||
});
|
||||
if (!res.ok) {
|
||||
const text = await res.text().catch(() => '');
|
||||
throw new Error(`Failed to get bank stats (HTTP ${res.status}): ${text}`);
|
||||
}
|
||||
return res.json() as Promise<BankStats>;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -611,7 +611,7 @@ const PROVIDER_DETECTION = [
|
|||
{ name: 'claude-code', keyEnv: '' },
|
||||
];
|
||||
|
||||
function detectLLMConfig(pluginConfig?: PluginConfig): {
|
||||
export function detectLLMConfig(pluginConfig?: PluginConfig): {
|
||||
provider?: string;
|
||||
apiKey?: string;
|
||||
model?: string;
|
||||
|
|
@ -735,7 +735,7 @@ function detectLLMConfig(pluginConfig?: PluginConfig): {
|
|||
* Detect external Hindsight API configuration.
|
||||
* Priority: env vars > plugin config
|
||||
*/
|
||||
function detectExternalApi(pluginConfig?: PluginConfig): {
|
||||
export function detectExternalApi(pluginConfig?: PluginConfig): {
|
||||
apiUrl: string | null;
|
||||
apiToken: string | null;
|
||||
} {
|
||||
|
|
@ -747,7 +747,7 @@ function detectExternalApi(pluginConfig?: PluginConfig): {
|
|||
/**
|
||||
* Build HindsightClientOptions from LLM config, plugin config, and external API settings.
|
||||
*/
|
||||
function buildClientOptions(
|
||||
export function buildClientOptions(
|
||||
llmConfig: { provider?: string; apiKey?: string; model?: string },
|
||||
pluginCfg: PluginConfig,
|
||||
externalApi: { apiUrl: string | null; apiToken: string | null },
|
||||
|
|
|
|||
|
|
@ -129,6 +129,22 @@ export interface RecallResponse {
|
|||
chunks: unknown | null;
|
||||
}
|
||||
|
||||
export interface BankStats {
|
||||
bank_id: string;
|
||||
total_nodes: number;
|
||||
total_links: number;
|
||||
total_documents: number;
|
||||
pending_operations: number;
|
||||
failed_operations: number;
|
||||
pending_consolidation: number;
|
||||
last_consolidated_at: string | null;
|
||||
total_observations: number;
|
||||
nodes_by_fact_type?: Record<string, number>;
|
||||
links_by_link_type?: Record<string, number>;
|
||||
links_by_fact_type?: Record<string, number>;
|
||||
links_breakdown?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface MemoryResult {
|
||||
id: string;
|
||||
text: string;
|
||||
|
|
|
|||
Loading…
Reference in a new issue