Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions packages/openclaw-memory/scripts/e2e.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import { fileURLToPath } from 'node:url';
import { ownerBrowser } from '@repo/browser-testing/owner-browser';
import metadata from '../package.json';
import { CALLBACK_URL, PLUGIN_ID } from '../src/contract';
import { evidenceFromMessages } from '../src/learning-evidence';
import { LearningStore } from '../src/learning-store';
import { OPENCLAW_INSTALL_COMMAND, openclawSetupPrompt } from '../src/setup-prompt';
import { reservePort, startApp } from './e2e-app';
Expand Down Expand Up @@ -102,6 +103,35 @@ async function configuration() {
);
}

const completedEvidence = evidenceFromMessages([
{ role: 'user', content: 'This conversation has already been learned.', timestamp: 1 },
]);
async function verifyLearningProgress(input: { seed?: boolean } = {}) {
const state = await Bun.file(connectionFile).json();
const store = new LearningStore({
directory: dirname(connectionFile),
config: state.config,
connectionId: state.learningId,
});
try {
const pending = store.status().pending;
store.capture({
source: 'completed-conversation',
evidence: completedEvidence,
now: Date.now(),
});
if (input.seed) {
const job = store.next({ agentId: 'main', now: Date.now() })!;
assert.equal(job.source, 'completed-conversation');
store.acknowledge({ sessionKey: job.sessionKey });
store.finish({ id: job.id, now: Date.now() });
}
assert.equal(store.status().pending, pending, 'Completed conversation was queued again');
} finally {
store.close();
}
}

await mkdir(join(directory, 'bin'), { recursive: true });
await writeFile(
join(directory, 'bin/openclaw'),
Expand Down Expand Up @@ -251,6 +281,7 @@ config.models={providers:{fixture:{baseUrl:${JSON.stringify(`${model.origin}/v1`
assert(connected.plugins.entries['active-memory'].config.allowedChatTypes.includes('group'));
assert((await command(['openclaw', 'context-use', 'status'])).includes('"connected": true'));
console.log('Native installation and pasted callback authorization passed.');
await verifyLearningProgress({ seed: true });

await command(['npx', '--yes', packageSpec, 'connect', app.origin]);
assert((await Bun.file(connectionFile).json()).oauth.tokens);
Expand Down Expand Up @@ -285,6 +316,7 @@ config.models={providers:{fixture:{baseUrl:${JSON.stringify(`${model.origin}/v1`
console.log(
'Explicit reconnect registered and authorized a fresh client while retaining restoration history.',
);
await verifyLearningProgress();
await mkdir(join(directory, 'moved'));
const movedApp = await startApp({ repo, directory: join(directory, 'moved'), node });
let movedOwner: Awaited<ReturnType<typeof ownerBrowser>> | undefined;
Expand Down Expand Up @@ -321,6 +353,7 @@ config.models={providers:{fixture:{baseUrl:${JSON.stringify(`${model.origin}/v1`
}),
);
assert.deepEqual((await Bun.file(connectionFile).json()).changes, originalJournal);
await verifyLearningProgress();
const reauthorizedConfig = await configuration();
assert.equal(
reauthorizedConfig.plugins.entries['active-memory'].config.timeoutMs,
Expand Down Expand Up @@ -478,6 +511,7 @@ config.models={providers:{fixture:{baseUrl:${JSON.stringify(`${model.origin}/v1`
clientName: 'OpenClaw reinstallation',
});
await authorize(newRedirect);
await verifyLearningProgress();
// This was the original timeout before reinstallation, so it must not look like a failed commit.
await configure("config.plugins.entries['active-memory'].config.timeoutMs=45000;");
await command(['npx', '--yes', packageSpec, 'connect', app.origin]);
Expand Down
37 changes: 36 additions & 1 deletion packages/openclaw-memory/src/connection.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,13 @@ import {
prepareConfiguration,
restoreConfiguration,
} from './configuration';
import { AUTHORIZATION_SCOPE, PLUGIN_ID, REQUEST_TIMEOUT_MS, serverUrl } from './contract';
import {
AUTHORIZATION_SCOPE,
PLUGIN_ID,
type PluginConfig,
REQUEST_TIMEOUT_MS,
serverUrl,
} from './contract';
import { ConnectionError } from './error';
import { attachmentDirectory, LearningStore, learningDatabase } from './learning-store';
import { authorizationResponse, oauthProvider } from './oauth';
Expand Down Expand Up @@ -62,6 +68,18 @@ async function removeLearningData(input: {
}
}

function preserveLearningProgress(input: {
directory: string;
config: PluginConfig;
connectionId?: string;
}): void {
const { connectionId } = input;
if (connectionId && existsSync(join(input.directory, learningDatabase(connectionId)))) {
// Export receipts from a queue created before durable progress was introduced.
new LearningStore({ ...input, connectionId }).close();
}
}

export async function connect(input: {
directory: string;
instance: string;
Expand All @@ -86,6 +104,11 @@ export async function connect(input: {
await installPackage();
if (input.reauthorize || state.config.serverUrl !== config.serverUrl) {
const previousLearningId = state.learningId;
preserveLearningProgress({
directory: input.directory,
config: state.config,
connectionId: previousLearningId,
});
state.config = config;
state.oauth = {};
state.tools = [];
Expand Down Expand Up @@ -147,6 +170,11 @@ export async function finishAuthorization(input: {
delete state.oauth.verifier;
// A new authorization can belong to another person even when the OAuth client is reused.
const previousLearningId = state.learningId;
preserveLearningProgress({
directory: input.directory,
config: state.config,
connectionId: previousLearningId,
});
state.learningId = randomUUID();
await writeState({ directory: input.directory, state });
if (previousLearningId) {
Expand Down Expand Up @@ -187,6 +215,13 @@ export async function disconnect(directory: string): Promise<{ preserved: string
}
},
});
if (state?.learningId) {
preserveLearningProgress({
directory,
config: state.config,
connectionId: state.learningId,
});
}
await rm(directory, { recursive: true, force: true });
return { preserved };
},
Expand Down
102 changes: 82 additions & 20 deletions packages/openclaw-memory/src/learning-store.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { randomUUID } from 'node:crypto';
import { createHash, randomUUID } from 'node:crypto';
import { chmodSync, mkdirSync } from 'node:fs';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
Expand All @@ -19,6 +19,14 @@ export const attachmentDirectory = (connectionId: string) =>
export const DREAM_INTERVAL_MS = 21_600_000;
export const learningDatabase = (connectionId: string) =>
`learning-${encodeURIComponent(connectionId)}.sqlite`;
const learningProgressDirectory = (directory: string) => `${directory}.progress`;

export function learningProgressDatabase(input: { directory: string; config: PluginConfig }) {
const scope = createHash('sha256')
.update(JSON.stringify([input.config.serverUrl, input.config.agentId]))
.digest('hex');
return join(learningProgressDirectory(input.directory), `${scope}.sqlite`);
}

export type LearningJob = {
id: string;
Expand All @@ -32,7 +40,7 @@ export type LearningJob = {
startedAt: number;
};

/** A temporary delivery queue, not an alternative personal memory store. */
/** Temporary evidence with durable completion receipts, not an alternative personal memory store. */
export class LearningStore {
private readonly db: DatabaseSync;

Expand Down Expand Up @@ -73,19 +81,53 @@ export class LearningStore {
this.db.close();
throw new Error('Context Use learning queue belongs to a different connection.');
}
// Receipts belong to the destination and agent, independently of OAuth credentials
// and the private connection directory removed during uninstall.
try {
mkdirSync(learningProgressDirectory(input.directory), {
recursive: true,
mode: PRIVATE_DIRECTORY_MODE,
});
const progressPath = learningProgressDatabase(input);
this.db.prepare('ATTACH DATABASE ? AS progress').run(progressPath);
chmodSync(progressPath, PRIVATE_FILE_MODE);
this.db.exec(`
CREATE TABLE IF NOT EXISTS progress.completed (
source TEXT NOT NULL, fingerprint TEXT NOT NULL, PRIMARY KEY (source, fingerprint)
);
CREATE TABLE IF NOT EXISTS progress.attachments (
source TEXT NOT NULL, key TEXT NOT NULL, asset TEXT, uploadName TEXT,
PRIMARY KEY (source, key)
);
INSERT OR IGNORE INTO progress.completed
SELECT source, fingerprint FROM evidence WHERE text IS NULL;
INSERT OR IGNORE INTO progress.attachments
SELECT source, key, asset, uploadName FROM attachments
WHERE asset IS NOT NULL OR uploadName IS NOT NULL;
`);
} catch (error) {
this.db.close();
throw error;
}
}

capture(input: { source: string; evidence: LearningEvidence[]; now: number }): void {
this.transaction(() => {
const insert = this.db.prepare(
'INSERT OR IGNORE INTO evidence(source, fingerprint, text, capturedAt) VALUES (?, ?, ?, ?)',
`INSERT OR IGNORE INTO evidence(source, fingerprint, text, capturedAt)
SELECT ?, ?, ?, ? WHERE NOT EXISTS (
SELECT 1 FROM progress.completed WHERE source = ? AND fingerprint = ?
)`,
);
for (const item of input.evidence) {
insert.run(input.source, item.key, item.text, input.now);
insert.run(input.source, item.key, item.text, input.now, input.source, item.key);
if (item.attachment) {
this.db
.prepare(`INSERT OR IGNORE INTO attachments(source, key, evidenceId, data)
SELECT source, fingerprint, id, ? FROM evidence WHERE source = ? AND fingerprint = ?`)
.prepare(`INSERT OR IGNORE INTO attachments(source, key, evidenceId, data, asset, uploadName)
SELECT e.source, e.fingerprint, e.id, ?, p.asset, p.uploadName
FROM evidence e LEFT JOIN progress.attachments p
ON p.source = e.source AND p.key = e.fingerprint
WHERE e.source = ? AND e.fingerprint = ? AND e.text IS NOT NULL`)
.run(JSON.stringify(item.attachment), input.source, item.key);
}
}
Expand Down Expand Up @@ -209,30 +251,46 @@ export class LearningStore {

prepareAttachmentUpload(input: { jobId: string; key: string; name: string }): string {
const source = this.attachmentSource(input);
this.db
.prepare(
'UPDATE attachments SET uploadName = COALESCE(uploadName, ?) WHERE source = ? AND key = ?',
)
.run(input.name, source, input.key);
return (
return this.transaction(() => {
this.db
.prepare('SELECT uploadName FROM attachments WHERE source = ? AND key = ?')
.get(source, input.key) as { uploadName: string }
).uploadName;
.prepare(
'UPDATE attachments SET uploadName = COALESCE(uploadName, ?) WHERE source = ? AND key = ?',
)
.run(input.name, source, input.key);
this.preserveAttachment({ source, key: input.key });
return (
this.db
.prepare('SELECT uploadName FROM attachments WHERE source = ? AND key = ?')
.get(source, input.key) as { uploadName: string }
).uploadName;
});
}

clearAttachmentUploadName(input: { jobId: string; key: string }): void {
const source = this.attachmentSource(input);
this.db
.prepare('UPDATE attachments SET uploadName = NULL WHERE source = ? AND key = ?')
.run(source, input.key);
this.transaction(() => {
this.db
.prepare('UPDATE attachments SET uploadName = NULL WHERE source = ? AND key = ?')
.run(source, input.key);
this.preserveAttachment({ source, key: input.key });
});
}

savedAttachment(input: { jobId: string; key: string; asset: string }): void {
const source = this.attachmentSource(input);
this.transaction(() => {
this.db
.prepare('UPDATE attachments SET asset = ? WHERE source = ? AND key = ?')
.run(input.asset, source, input.key);
this.preserveAttachment({ source, key: input.key });
});
}

private preserveAttachment(input: { source: string; key: string }): void {
this.db
.prepare('UPDATE attachments SET asset = ? WHERE source = ? AND key = ?')
.run(input.asset, source, input.key);
.prepare(`INSERT OR REPLACE INTO progress.attachments
SELECT source, key, asset, uploadName FROM attachments WHERE source = ? AND key = ?`)
.run(input.source, input.key);
}

acknowledge(input: { sessionKey: string; omittedAttachments?: string[] }): void {
Expand Down Expand Up @@ -272,6 +330,10 @@ export class LearningStore {
throw new Error('Learning must be acknowledged before removing its evidence.');
}
if (job.kind === 'learn') {
this.db
.prepare(`INSERT OR IGNORE INTO progress.completed
SELECT source, fingerprint FROM evidence WHERE source = ? AND id <= ?`)
.run(job.source, job.through);
this.db
.prepare('UPDATE evidence SET text = NULL WHERE source = ? AND id <= ?')
.run(job.source, job.through);
Expand Down
3 changes: 3 additions & 0 deletions packages/openclaw-memory/src/setup-skill.ts
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,9 @@ Run ${OPENCLAW_INSTALL_COMMAND} remove. It restores the plugin's configuration
changes, clears local credentials, cancels unfinished learning, discards pending
local evidence, and uninstalls the package. User edits, remote memories, and
conversation history are preserved. It also works when setup was unfinished or disabled.
Completed-learning fingerprints and saved-attachment references remain locally so reinstalling
for the same instance and agent does not repeat completed work. These receipts contain no
conversation text, attachment bytes, or credentials.

Removal runs separately so the requesting turn can finish. End this turn once
removal is queued. Do not poll in this turn, schedule retries, stop the gateway,
Expand Down
5 changes: 3 additions & 2 deletions packages/openclaw-memory/test/learning-attachments.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@ import { attachmentsFromMessages } from '../src/learning-evidence';
import { attachmentDirectory, LearningStore } from '../src/learning-store';

test('staged original bytes survive loss of the host cache and are removed after learning', async () => {
const directory = await mkdtemp(join(tmpdir(), 'learning-attachments-'));
const root = await mkdtemp(join(tmpdir(), 'learning-attachments-'));
const directory = join(root, 'connection');
const connectionId = 'owner';
const config = { agentId: 'main', serverUrl: 'https://example.com/mcp' };
const db = new LearningStore({ directory, config, connectionId });
Expand Down Expand Up @@ -53,6 +54,6 @@ test('staged original bytes survive loss of the host cache and are removed after
await expect(read()).rejects.toThrow();
} finally {
db.close();
await rm(directory, { recursive: true, force: true });
await rm(root, { recursive: true, force: true });
}
});
Loading
Loading