openvibers/openvibe.ai Commits
Memory M2c: dedupe by meaning, conflicts and the profile builder (#211)
Changes
@@@ -91,6 +91,16 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000# the Memory M2a review; changing it makes existing items unreadable.#AI_MEMORY_KEY=++# The store's own maintenance (Memory M2c): the loop walks owner by owner, with a per-run cap. It is++# off in test mode and off while a provider is in a drill, so a drill never spends the platform's budget.++#AI_MEMORY_JOBS=1++#AI_MEMORY_JOB_INTERVAL_MS=900000++#AI_MEMORY_JOB_MAX_OWNERS=10++#AI_MEMORY_JOB_MAX_SCOPES=4++#AI_MEMORY_JOB_MAX_EMBEDDINGS=100++#AI_MEMORY_JOB_MAX_MODEL_CALLS=20++#AI_MEMORY_JOB_MAX_PROFILES=10++# ── The shared provider — the same names as OpenVibe.Live's admin AI settings ──# (ai_enabled, ai_provider, ai_base_url, ai_api_key, ai_model, ai_pricing_json,# ai_input/output_cost_per_mtok, ai_max_cost_usd_per_day).nnew file mode 100644@@@ -0,0 +1,23 @@++-- phase: expand++-- OpenVibe Memory M2c: dedupe by meaning, conflicts and the profile builder (spec++-- research-2026-10-30/memory-m2-spec.md, stage M2c). Expand only: every column below is nullable, so the rows++-- M2a wrote keep working and no row is rewritten.++-- embedding_enc the item's embedding, sealed like the text: AES-256-GCM (AES-256-GCM, `m1:` + iv + tag +++-- body) with the AAD `memory|<owner>|<id>|<scope>|emb`. An embedding leaks meaning (a nearest++-- neighbour inverts it), so it is sealed like the text it came from and is only ever opened++-- inside AI, on rows this owner owns.++-- embedding_model the embedding model that produced the vector. Two vector spaces are never compared: an item++-- whose model differs from the current one is embedded again before it is compared.++-- merged_into the item this one was merged into. The older item of a same-meaning pair keeps its row and++-- its id here, and is archived: the person can split them again.++-- conflict_with the item this one conflicts with. Both are `status = 'conflict'`, so neither reaches an++-- agent, until the person settles the pair.++-- source_own the item's own sources before a merge added another one, so a split restores exactly what++-- the item had, not the union of two items.++ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS embedding_enc text;++ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS embedding_model text;++ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS merged_into text;++ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS conflict_with text;++ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS source_own text;++CREATE INDEX IF NOT EXISTS memory_items_owner_merged_into ON memory_items (owner, merged_into);++CREATE INDEX IF NOT EXISTS memory_items_owner_conflict_with ON memory_items (owner, conflict_with);@@@ -60,6 +60,17 @@ function load(env = process.env) {credentialsKey: String(env.AI_CREDENTIALS_KEY || '').trim(),// The person's memory (server/memory/): AES-256-GCM with a key of its own, so it can rotate apart from credentials.memoryKey: String(env.AI_MEMORY_KEY || '').trim(),++ // The store's own maintenance (server/memory/jobs.js): owner by owner, with a per-run cap. Off in test mode++ // (the way the JWKS client is) and off while a provider is in a drill, so a drill never spends the budget.++ memoryJobs: {++ enabled: bool(env.AI_MEMORY_JOBS, nodeEnv !== 'test'),++ intervalMs: int(env.AI_MEMORY_JOB_INTERVAL_MS, 15 * 60 * 1000),++ maxOwners: int(env.AI_MEMORY_JOB_MAX_OWNERS, 10),++ maxScopesPerOwner: int(env.AI_MEMORY_JOB_MAX_SCOPES, 4),++ maxEmbeddings: int(env.AI_MEMORY_JOB_MAX_EMBEDDINGS, 100),++ maxModelCalls: int(env.AI_MEMORY_JOB_MAX_MODEL_CALLS, 20),++ maxProfiles: int(env.AI_MEMORY_JOB_MAX_PROFILES, 10),++ },networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,audience: 'openvibe.ai',@@@ -76,6 +76,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowconst auth = createAuth({ config, log });// The person's memory (Memory M2a): sealed item text, the profile cache and the agent-read audit.const memory = require('./memory').createMemory({ db, config, clock, log, registry, pool });++ // The store's own maintenance (Memory M2c): owner by owner, with a per-run cap. It is off in test mode and off++ // while a provider is in a drill (server/memory/jobs.js).++ const memoryJobs = require('./memory/jobs').createMemoryJobs({ memory, config, clock, log, registry });// The Events consumer side: a signed bench.rating.changed webhook, subscribed at boot (server/bench/events.js).const accountData = accountDataLib.create({ db, log, memory });const accountSend = env.OV_OAUTH_CLIENT_SECRET ? createNetworkSender({@@@ -106,6 +109,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowlet benchSubscription = null;let accountSubscription = null;if (listen) userModules.start();++ if (listen && config.memoryJobs && config.memoryJobs.enabled) memoryJobs.start();if (listen) {server = await new Promise((resolve, reject) => {const s = app.listen(config.port, config.host, () => resolve(s));@@@ -132,6 +136,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowif (benchSubscription) benchSubscription.stop();if (accountSubscription) accountSubscription.stop();userModules.stop();++ memoryJobs.stop();if (jwks) jwks.stop();if (server) {server.closeAllConnections?.();@@@ -146,7 +151,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowif (!givenDb) await db.close();}-- return { config, db, registry, pool, quotas, tiers, cache, fetcher, engine, runs, keys, keyLoaded, auth, app, server, bench, benchEvents, memory, close };++ return { config, db, registry, pool, quotas, tiers, cache, fetcher, engine, runs, keys, keyLoaded, auth, app, server, bench, benchEvents, memory, memoryJobs, close };}if (require.main === module) {@@@ -15,6 +15,8 @@* PATCH /api/v1/memory/items/:id text, kind, scope or status* DELETE /api/v1/memory/items/:id* DELETE /api/v1/memory/items?source= everything from one source++ * POST /api/v1/memory/items/:id/split undo a merge: the item and the one it swallowed go apart again++ * POST /api/v1/memory/items/:id/settle the person settles a conflict: { keep: <id> } keeps one and archives the other* GET /api/v1/memory/profile?scope= the cached profile, rebuilt by the cheap tier only when its items_hash moved,* at most once an hour per owner and scope* GET /api/v1/memory/search?q=&scopes= for agents (ai.memory.read + on_behalf_of): active items only, in the@@@ -22,6 +24,22 @@** Every SQL query that reads or changes a memory table is scoped by `owner = $…` (test/memory.test.js greps for it).* The account export decrypts the person's own items for their own export; deletion removes the rows.++ *++ * M2c (this stage): the meaning of an item.++ * - Every new or edited item gets an embedding from AI's own embed route, called in-process (the platform's++ * budget, never the person's allowance). The vector is stored sealed, with the same AES-256-GCM scheme as the++ * text and the AAD `memory|<owner>|<id>|<scope>|emb`: an embedding leaks meaning (a nearest neighbour inverts++ * it), so it is as private as the text it came from, and only this module ever opens it.++ * - Two active items of one owner and scope that say the same thing (cosine at least 0.92) become one item with++ * two sources: the newer text wins, the older row is archived and keeps its id in `merged_into`. The person can++ * split them again (POST /items/:id/split).++ * - Two active items that are near each other but disagree (cosine 0.75 to 0.92) are judged by one cheap-tier++ * call that sees only those two notes, and both are marked `status = 'conflict'` with a `conflict_with` link:++ * a conflict never reaches an agent. The person settles it (POST /items/:id/settle { keep }).++ * - The profile is built from at most 60 active notes, newest first, passes noLooks and is at most about a++ * thousand tokens. It is cached by items_hash and rebuilt at most once an hour per owner and scope.++ * - The store's own maintenance runs in the background (server/memory/jobs.js): owner by owner, with a per-run++ * cap, off in test mode and off in a drill.*/const crypto = require('crypto');const express = require('express');@@@ -40,6 +58,17 @@ const STATUSES = ['active', 'suggested', 'archived', 'conflict'];/** Items from these sources were produced by an AI, so they are suggestions and pass noLooks. */const AI_SOURCES = new Set(['actor', 'vibe', 'know']);const PROFILE_INTERVAL_MS = 60 * 60 * 1000;++/* M2c: the meaning of an item. Two items of one owner and scope that say the same thing (cosine at least++ * SAME_MEANING) become one item with two sources; two that are near each other but disagree (CONFLICT_MIN to++ * SAME_MEANING) are marked and hidden until the person settles them. */++const SAME_MEANING = 0.92;++const CONFLICT_MIN = 0.75;++const EMBED_ROUTE = 'default.embedding';++const CHEAP_ROUTE = 'default.chat';++const PROFILE_ITEMS = 60; // the profile is built from at most this many notes, newest first++const PROFILE_MAX_TOKENS = 1000; // and the result is at most about this many tokens++const PROFILE_MAX_CHARS = 4000;++const RECONCILE_ROWS = 200; // the rows compared per owner and scopeconst PROFILE_ID = 'profile';const MAX_BATCH = 200;const MAX_LIST = 2000;@@@ -134,6 +163,230 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = conconst normalize = (text) => String(text == null ? '' : text).replace(/\s+/g, ' ').trim();const macOf = (text) => crypto.createHmac('sha256', macKey()).update(normalize(text)).digest('hex');++ // ── M2c: the meaning of an item ───────────────────────────────────────────────────────────────────────────++ /** The AAD of a vector: the text's AAD with the `|emb` suffix, so a vector never opens as text. */++ function embAad(owner, id, scope) { return aadFor(owner, id, scope) + '|emb'; }++++ /** The sources of a row, as a list. */++ const sourceList = (value) => String(value || '').split(',').map((s) => s.trim()).filter(Boolean);++++ /** Cosine similarity of two vectors; 0 when they do not fit each other (a different length, an empty one). */++ function cosine(a, b) {++ if (!Array.isArray(a) || !Array.isArray(b) || a.length !== b.length || !a.length) return 0;++ let dot = 0, na = 0, nb = 0;++ for (let i = 0; i < a.length; i++) {++ const x = Number(a[i]) || 0;++ const y = Number(b[i]) || 0;++ dot += x * y; na += x * x; nb += y * y;++ }++ if (!na || !nb) return 0;++ return dot / (Math.sqrt(na) * Math.sqrt(nb));++ }++++ /** The vector of a row, opened in-process; null when the row has none or it does not open. */++ function vectorOf(row, owner) {++ if (!row || !row.embedding_enc) return null;++ try {++ const v = JSON.parse(open(row.embedding_enc, embAad(owner, row.id, row.scope)));++ return Array.isArray(v) && v.length ? v : null;++ } catch { return null; }++ }++++ /**++ * The embeddings of a few texts, from AI's own embed route, called in-process (never over HTTP to itself):++ * the platform's own budget, never the person's allowance. The texts travel in the request and nowhere else:++ * the pool is not asked to log the request, so the notes are not stored a second time.++ */++ async function embedTexts(texts) {++ const list = (texts || []).map((t) => String(t == null ? '' : t));++ if (!list.length || !registry || !pool) return { model: null, vectors: [] };++ const route = await registry.resolveRoute(EMBED_ROUTE);++ if (!route || route.disabled) return { model: null, vectors: [] };++ const exec = await pool.execute(route, 'embed', { input: list, timeoutMs: route.timeout_ms || 30000 },++ { routeKey: route.key, routeVersion: route.version });++ const vectors = (exec.result && exec.result.vectors) || [];++ if (vectors.length !== list.length) return { model: exec.model || null, vectors: [] };++ return {++ model: exec.model || null,++ vectors: vectors.map((v) => (Array.isArray(v) ? v.map((n) => Number(n)) : null)),++ };++ }++++ /** Seals a vector under the item's AAD and stores it with the model that produced it. */++ async function storeEmbedding(owner, row, vector, model) {++ if (!row || !Array.isArray(vector) || !vector.length) return false;++ const payload = JSON.stringify(vector.map((n) => Math.round((Number(n) || 0) * 1e6) / 1e6));++ await q('UPDATE memory_items SET embedding_enc = $1, embedding_model = $2 WHERE owner = $3 AND id = $4')++ .run(seal(payload, embAad(owner, row.id, row.scope)), model || null, owner, row.id);++ return true;++ }++++ /** The active items of one owner and scope, newest first, with the vectors they have. */++ async function scopeItems(owner, scope) {++ const rows = await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' ORDER BY updated_at DESC LIMIT " + RECONCILE_ROWS)++ .all(owner, scope);++ const out = [];++ for (const row of rows) {++ const vector = vectorOf(row, owner);++ if (vector) out.push({ row, vector });++ }++ return out;++ }++++ /** The newer of two items wins; the older one is archived and keeps its id in `merged_into`. */++ function newerFirst(a, b) {++ const at = Date.parse(a.row.updated_at || a.row.created_at || '') || 0;++ const bt = Date.parse(b.row.updated_at || b.row.created_at || '') || 0;++ if (at !== bt) return at > bt ? [a, b] : [b, a];++ return String(a.row.id) > String(b.row.id) ? [a, b] : [b, a];++ }++++ /** One item with two sources: the winner keeps its text, the loser is archived and points at the winner. */++ async function mergePair(owner, winner, loser) {++ const at = iso(clock.now());++ const own = winner.row.source_own ? sourceList(winner.row.source_own) : sourceList(winner.row.source);++ const union = [...new Set([...sourceList(winner.row.source), ...sourceList(loser.row.source)])];++ await db.tx(async () => {++ await q('UPDATE memory_items SET source = $1, source_own = $2, updated_at = $3 WHERE owner = $4 AND id = $5')++ .run(union.join(',').slice(0, 500), own.join(',').slice(0, 500), at, owner, winner.row.id);++ await q('UPDATE memory_items SET status = $1, merged_into = $2, updated_at = $3 WHERE owner = $4 AND id = $5')++ .run('archived', winner.row.id, at, owner, loser.row.id);++ });++ return winner.row.id;++ }++++ /** A JSON answer, wherever the provider put it. */++ function parseJsonAnswer(value) {++ if (value && typeof value === 'object') return value;++ const text = String(value == null ? '' : value).trim();++ if (!text) return null;++ try { return JSON.parse(text); } catch { /* a model that answered in prose */ }++ const found = /\{[\s\S]*\}/.exec(text);++ if (found) { try { return JSON.parse(found[0]); } catch { return null; } }++ return null;++ }++++ /**++ * One cheap-tier call: do these two notes disagree? The call sees these two notes and nothing else: never++ * another person's item, never a third note. The answer is `{ conflict, why }`.++ */++ async function judgePair(owner, a, b) {++ if (!registry || !pool) return { conflict: false, why: '' };++ const route = await registry.resolveRoute(CHEAP_ROUTE);++ if (!route || route.disabled) return { conflict: false, why: '' };++ const textA = open(a.text_enc, aadFor(owner, a.id, a.scope));++ const textB = open(b.text_enc, aadFor(owner, b.id, b.scope));++ const schema = {++ type: 'object', additionalProperties: false, required: ['conflict', 'why'],++ properties: { conflict: { type: 'boolean' }, why: { type: 'string' } },++ };++ const prompt = [++ 'Two notes from the same person. Do they disagree?',++ 'Answer conflict true only when the two notes cannot both be true as they stand.',++ 'Notes about the same thing that are compatible, or that differ only in wording or in detail, are not a conflict.',++ 'Answer with conflict (true or false) and why (one short sentence).',++ '',++ 'Note A:', textA,++ '',++ 'Note B:', textB,++ ].join('\n');++ const exec = await pool.execute(route, 'chat', {++ system: [{ text: 'You judge whether two notes from one person contradict each other. You answer with the schema you are given and nothing else.' }],++ messages: [{ role: 'user', content: prompt }],++ json: { name: 'conflict', schema, strict: true },++ maxTokens: 200, temperature: 0, timeoutMs: 20000,++ }, { routeKey: route.key, routeVersion: route.version });++ const parsed = parseJsonAnswer(exec.result && (exec.result.json || exec.result.text));++ return { conflict: Boolean(parsed && parsed.conflict === true), why: String((parsed && parsed.why) || '').slice(0, 300) };++ }++++ /** Both items of a conflicting pair are marked, so neither reaches an agent. */++ async function markConflict(owner, aId, bId) {++ const at = iso(clock.now());++ await db.tx(async () => {++ await q("UPDATE memory_items SET status = 'conflict', conflict_with = $1, updated_at = $2 WHERE owner = $3 AND id = $4")++ .run(bId, at, owner, aId);++ await q("UPDATE memory_items SET status = 'conflict', conflict_with = $1, updated_at = $2 WHERE owner = $3 AND id = $4")++ .run(aId, at, owner, bId);++ });++ }++++ /**++ * One owner and one scope: the pairs that say the same thing become one item, and the pairs that are near each++ * other but disagree are marked. `maxModelCalls` caps the cheap-tier calls of this run (0: nothing is judged).++ * Two items are only compared when their vectors come from the same model: two vector spaces are never mixed.++ */++ async function reconcileScope(owner, scope, { maxModelCalls = 0 } = {}) {++ const items = await scopeItems(owner, scope);++ const out = { merged: 0, conflicts: 0, calls: 0 };++ if (items.length < 2) return out;++ const gone = new Set();++ for (let i = 0; i < items.length; i++) {++ if (gone.has(items[i].row.id)) continue;++ for (let j = i + 1; j < items.length; j++) {++ if (gone.has(items[j].row.id)) continue;++ if ((items[i].row.embedding_model || '') !== (items[j].row.embedding_model || '')) continue;++ const cos = cosine(items[i].vector, items[j].vector);++ if (cos >= SAME_MEANING) {++ const [winner, loser] = newerFirst(items[i], items[j]);++ await mergePair(owner, winner, loser);++ gone.add(loser.row.id);++ out.merged++;++ continue;++ }++ if (cos >= CONFLICT_MIN && out.calls < maxModelCalls) {++ out.calls++;++ const verdict = await judgePair(owner, items[i].row, items[j].row);++ if (verdict.conflict) {++ await markConflict(owner, items[i].row.id, items[j].row.id);++ gone.add(items[i].row.id);++ gone.add(items[j].row.id);++ out.conflicts++;++ }++ }++ }++ }++ return out;++ }++++ /** The items of an owner and scope that have no vector yet: they are embedded and stored (the platform's budget). */++ async function embedMissing(owner, scope, limit = 50) {++ const cap = Math.max(0, Math.min(Number(limit) || 0, RECONCILE_ROWS));++ if (!cap) return 0;++ const rows = await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' AND embedding_enc IS NULL ORDER BY updated_at DESC LIMIT " + cap)++ .all(owner, scope);++ if (!rows.length) return 0;++ const texts = rows.map((r) => open(r.text_enc, aadFor(owner, r.id, r.scope)));++ const { model, vectors } = await embedTexts(texts);++ let stored = 0;++ for (let i = 0; i < rows.length; i++) {++ if (!vectors[i]) continue;++ if (await storeEmbedding(owner, rows[i], vectors[i], model)) stored++;++ }++ return stored;++ }++++ /**++ * After an import: every new item gets an embedding, and the scope is settled. The import never spends a++ * judgement on its own (the job does that, with its cap); it only merges what says the same thing.++ */++ async function embedAndReconcile(owner, ids, scopes) {++ const out = { embedded: 0, merged: 0, conflicts: 0 };++ if (!ids.length) return out;++ const rows = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = ANY($2::text[])').all(owner, ids);++ const texts = rows.map((r) => open(r.text_enc, aadFor(owner, r.id, r.scope)));++ const { model, vectors } = await embedTexts(texts);++ for (let i = 0; i < rows.length; i++) {++ if (!vectors[i]) continue;++ if (await storeEmbedding(owner, rows[i], vectors[i], model)) out.embedded++;++ }++ for (const scope of scopes) {++ const r = await reconcileScope(owner, scope, { maxModelCalls: 0 });++ out.merged += r.merged;++ out.conflicts += r.conflicts;++ }++ return out;++ }function validateScope(scope, fallback = 'user') {const s = String(scope == null || scope === '' ? fallback : scope);@@@ -179,9 +432,13 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con}return {id: row.id, owner: row.owner, scope: row.scope, kind: row.kind, text,-- source: row.source, source_ref: sourceRef, status: row.status, confidence: Number(row.confidence),++ source: row.source, sources: sourceList(row.source), source_ref: sourceRef, status: row.status, confidence: Number(row.confidence),seen_at: row.seen_at, created_at: row.created_at, updated_at: row.updated_at,machine: row.machine || null,++ // M2c: the links. merged_into names the item this one was merged into (it is archived); conflict_with++ // names the item this one conflicts with (both are hidden from an agent until the person settles them).++ merged_into: row.merged_into || null,++ conflict_with: row.conflict_with || null,...(AI_SOURCES.has(String(row.source).split(',')[0]) ? { no_looks: true } : {}),};}@@@ -238,6 +495,8 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con};});const out = [];++ const createdIds = [];++ const scopes = new Set();let created = 0;let deduped = 0;await db.tx(async () => {@@@ -258,9 +517,21 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = conit.sourceRef ? seal(JSON.stringify(it.sourceRef), aadFor(owner, it.id, it.scope)) : null,it.status, it.confidence, it.seenAt, at, at, it.machine);created++;++ createdIds.push(it.id);++ scopes.add(it.scope);out.push(itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, it.id)));}});++ // M2c: every new item gets an embedding, and the scopes it touched are settled: what says the same thing++ // becomes one item with two sources. The import spends no judgement of its own: the job does that, with++ // its cap. A failed embedding never fails the import: the job embeds the item later.++ try { await embedAndReconcile(owner, createdIds, [...scopes]); }++ catch (err) { log.warn(`[memory] the embedding of a new item did not finish (${err && err.code ? err.code : 'error'}); the job will do it`); }++ // The answer shows the state after the merge: an item that was swallowed is archived and points at the winner.++ for (let i = 0; i < out.length; i++) {++ const row = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, out[i].id);++ if (row) out[i] = itemView(row);++ }return { items: out, count: created + deduped, created, deduped };}@@@ -344,6 +615,16 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con// A profile built from the old text, kind or scope will not match its items_hash again: drop it now.const scopes = [...new Set([row.scope, nextScope])];await q('DELETE FROM memory_profiles WHERE owner = $1 AND scope = ANY($2::text[])').run(owner, scopes);++ // M2c: a new text is a new meaning: the item is embedded again, and its scope is settled (what says the same++ // thing becomes one item). A failed embedding never fails the edit: the job embeds the item later.++ if (nextText !== null) {++ const fresh = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, row.id);++ try {++ const { model, vectors } = await embedTexts([nextText]);++ if (vectors[0]) await storeEmbedding(owner, fresh, vectors[0], model);++ await reconcileScope(owner, nextScope, { maxModelCalls: 0 });++ } catch (err) { log.warn(`[memory] the embedding of an edited item did not finish (${err && err.code ? err.code : 'error'}); the job will do it`); }++ }return { item: itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, row.id)) };}@@@ -375,7 +656,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con/** Items from one scope, active only: what a profile is built from and what an agent may see. */async function activeRows(owner, scope) {-- return await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' ORDER BY updated_at DESC LIMIT 2000").all(owner, scope);++ return await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' ORDER BY updated_at DESC LIMIT " + PROFILE_ITEMS).all(owner, scope);}function itemsHash(macs) {@@@ -389,7 +670,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con`Write a short profile of the person from their own memory notes for ${scope}.`,'Rules: summarize only what the notes say. Never guess or infer facts about the person.','Never describe looks, body, clothing, hair, face, pose, age or a guessed gender: no looks, at any time.',-- 'Write at most 400 words and do not add a title.',++ 'Write at most 600 words and do not add a title.','',notes,].join('\n');@@@ -402,17 +683,19 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = conconst exec = await pool.execute(route, 'summarize', {system: [{ text: 'You write short, faithful profiles of a person from their own notes. Never guess, never describe looks or a body, and never add a fact the notes do not state.', cache: true }],messages: [{ role: 'user', content: prompt }],-- maxTokens: 400,++ maxTokens: PROFILE_MAX_TOKENS,temperature: 0.2,timeoutMs: 30000,}, { routeKey: route.key, routeVersion: route.version });return String((exec.result && exec.result.text) || '');}-- async function profile(req, owner) {-- assertConfigured();-- const scope = validateScope(req.query.scope, 'user');-- requireScope(req, scope);++ /**++ * The builder: the profile of one owner and scope. The cache is keyed by the items_hash (the MACs of the items++ * it was built from) and is rebuilt only when that hash moved and the cache is at least an hour old: at most one++ * build per owner and scope per hour, on the platform's own budget.++ */++ async function profileFor(owner, scope) {const rows = await activeRows(owner, scope);const hash = itemsHash(rows.map((r) => r.text_mac));const now = clock.now();@@@ -422,7 +705,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = conconst fresh = Number.isFinite(age) && age < PROFILE_INTERVAL_MS;if (fresh || cached.items_hash === hash) {const text = open(cached.text_enc, aadFor(owner, PROFILE_ID, scope));-- return { profile: { scope, text, items_hash: cached.items_hash, built_at: cached.built_at } };++ return { profile: { scope, text, items_hash: cached.items_hash, built_at: cached.built_at }, rebuilt: false };}}if (!rows.length) {@@@ -431,15 +714,22 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = conawait q(`INSERT INTO memory_profiles (owner, scope, text_enc, items_hash, built_at) VALUES ($1, $2, $3, $4, $5)ON CONFLICT (owner, scope) DO UPDATE SET text_enc = excluded.text_enc, items_hash = excluded.items_hash, built_at = excluded.built_at`).run(owner, scope, empty, hash, builtAt);-- return { profile: { scope, text: '', items_hash: hash, built_at: builtAt } };++ return { profile: { scope, text: '', items_hash: hash, built_at: builtAt }, rebuilt: true };}const items = rows.map((r) => ({ kind: r.kind, text: open(r.text_enc, aadFor(owner, r.id, r.scope)) }));-- const text = String(noLooks(await summarize({ owner, scope, items })) || '').slice(0, 4000);++ const text = String(noLooks(await summarize({ owner, scope, items })) || '').slice(0, PROFILE_MAX_CHARS);const builtAt = iso(now);await q(`INSERT INTO memory_profiles (owner, scope, text_enc, items_hash, built_at) VALUES ($1, $2, $3, $4, $5)ON CONFLICT (owner, scope) DO UPDATE SET text_enc = excluded.text_enc, items_hash = excluded.items_hash, built_at = excluded.built_at`).run(owner, scope, seal(text, aadFor(owner, PROFILE_ID, scope)), hash, builtAt);-- return { profile: { scope, text, items_hash: hash, built_at: builtAt } };++ return { profile: { scope, text, items_hash: hash, built_at: builtAt }, rebuilt: true };++ }++++ async function profile(req, owner) {++ assertConfigured();++ const scope = validateScope(req.query.scope, 'user');++ requireScope(req, scope);++ return { profile: (await profileFor(owner, scope)).profile };}/** An agent's search: active items only, the granted scopes, at most 8, and one memory_reads row with ids only. */@@@ -468,6 +758,83 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = conreturn { items, count: items.length };}++ /**++ * Undo a merge: the item and the one it swallowed go apart again. The archived item keeps its own text and its++ * own sources, and the winner's source list goes back to what it was before the merge.++ */++ async function split(req, owner, id) {++ assertConfigured();++ const row = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, id);++ if (!row) throw new AiError(404, 'memory.not_found', 'no such memory item');++ requireScope(req, row.scope);++ const children = await q("SELECT * FROM memory_items WHERE owner = $1 AND merged_into = $2 AND status = 'archived' ORDER BY created_at ASC")++ .all(owner, row.id);++ if (!children.length) throw new AiError(409, 'memory.not_split', 'this item was not merged with another one');++ const at = iso(clock.now());++ await db.tx(async () => {++ for (const child of children) {++ await q('UPDATE memory_items SET status = $1, merged_into = $2, updated_at = $3 WHERE owner = $4 AND id = $5')++ .run('active', null, at, owner, child.id);++ }++ const own = row.source_own ? sourceList(row.source_own) : sourceList(row.source);++ await q('UPDATE memory_items SET source = $1, updated_at = $2 WHERE owner = $3 AND id = $4')++ .run(own.join(',').slice(0, 500), at, owner, row.id);++ });++ await q('DELETE FROM memory_profiles WHERE owner = $1 AND scope = $2').run(owner, row.scope);++ const ids = [row.id, ...children.map((c) => c.id)];++ const items = (await q('SELECT * FROM memory_items WHERE owner = $1 AND id = ANY($2::text[])').all(owner, ids)).map(itemView);++ return { items, count: items.length, split: children.length };++ }++++ /** The person settles a conflict: the item they keep becomes active again, the other one is archived. */++ async function settle(req, owner, id) {++ assertConfigured();++ const row = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, id);++ if (!row) throw new AiError(404, 'memory.not_found', 'no such memory item');++ requireScope(req, row.scope);++ const keep = String((req.body && req.body.keep) || '').trim();++ if (!keep) throw new AiError(422, 'input.invalid', 'keep must name the item that stays');++ if (keep === id) throw new AiError(422, 'input.invalid', 'keep must name the other item of the conflict');++ const other = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, keep);++ if (!other || other.scope !== row.scope) throw new AiError(404, 'memory.not_found', 'no such memory item');++ if (row.conflict_with !== keep && other.conflict_with !== id) throw new AiError(409, 'memory.not_conflict', 'these two items are not in conflict');++ const at = iso(clock.now());++ await db.tx(async () => {++ await q('UPDATE memory_items SET status = $1, conflict_with = $2, updated_at = $3 WHERE owner = $4 AND id = $5')++ .run('active', null, at, owner, other.id);++ await q('UPDATE memory_items SET status = $1, conflict_with = $2, updated_at = $3 WHERE owner = $4 AND id = $5')++ .run('archived', other.id, at, owner, row.id);++ });++ await q('DELETE FROM memory_profiles WHERE owner = $1 AND scope = $2').run(owner, row.scope);++ return {++ kept: itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, other.id)),++ archived: itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, row.id)),++ };++ }++++ /**++ * The owners the job has work for. This is the one statement that is not scoped by an owner: it returns owner++ * identifiers and nothing else (no item, no text, no id), so it cannot carry one person's data out. Every other++ * query of the job is owner-scoped.++ */++ async function ownersForJobs(limit) {++ const cap = Math.max(1, Math.min(Number(limit) || 1, 500));++ return await q('SELECT owner, MAX(updated_at) AS last_seen FROM memory_items WHERE owner IS NOT NULL AND owner <> $1 GROUP BY owner ORDER BY last_seen DESC LIMIT $2')++ .all('', cap);++ }++++ /** The scopes of one owner that hold items. */++ async function scopesFor(owner, limit) {++ const cap = Math.max(1, Math.min(Number(limit) || 1, 50));++ const rows = await q('SELECT DISTINCT scope FROM memory_items WHERE owner = $1 ORDER BY scope LIMIT $2').all(owner, cap);++ return rows.map((r) => r.scope);++ }++++ /** Rebuilds the profile of one owner and scope when it is due (the same rule as the route). */++ async function refreshProfile(owner, scope) {++ assertConfigured();++ return await profileFor(owner, scope);++ }/** The person's own export only: decrypted items, profiles and read ids (never another person's rows). */async function exportFor(subject) {assertConfigured();@@@ -483,7 +850,12 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con];}-- return { assertConfigured, batch, list, get, patch, remove, removeBySource, profile, search, exportFor, aadFor, macOf, seal, open };++ return {++ assertConfigured, batch, list, get, patch, remove, removeBySource, profile, search, exportFor,++ split, settle, aadFor, macOf, seal, open,++ // The store's own maintenance (server/memory/jobs.js) drives the store through these.++ ownersForJobs, scopesFor, embedMissing, reconcileScope, refreshProfile, profileFor,++ };}/**@@@ -547,6 +919,9 @@ function memoryRouter({ memory, auth, config, clock = { now: () => Date.now() },r.delete('/api/v1/memory/items', guard(WRITE_CAP), personOnly, run(async (req) => await memory.removeBySource(req, ownerOf(req))));r.get('/api/v1/memory/profile', guard(READ_CAP), personOnly, run(async (req) => await memory.profile(req, ownerOf(req))));r.get('/api/v1/memory/search', guard(READ_CAP), run(async (req) => await memory.search(req, ownerOf(req))));++ // M2c: undo a merge, and settle a conflict. Both are the person's own: a service may only search and suggest.++ r.post('/api/v1/memory/items/:id/split', guard(WRITE_CAP), personOnly, run(async (req) => await memory.split(req, ownerOf(req), req.params.id)));++ r.post('/api/v1/memory/items/:id/settle', guard(WRITE_CAP), personOnly, run(async (req) => await memory.settle(req, ownerOf(req), req.params.id)));// The owner is resolved after the guard has set the principal.function ownerOf(req) {const body = req.body && typeof req.body === 'object' ? req.body : {};@@@ -564,4 +939,4 @@ function memoryRouter({ memory, auth, config, clock = { now: () => Date.now() },return r;}--module.exports = { createMemory, memoryRouter, PROFILE_INTERVAL_MS };++module.exports = { createMemory, memoryRouter, PROFILE_INTERVAL_MS, PROFILE_ITEMS, SAME_MEANING, CONFLICT_MIN };nnew file mode 100644@@@ -0,0 +1,89 @@++'use strict';++/**++ * The background loop behind Memory M2c: the store's own maintenance, owner by owner, with a per-run cap.++ *++ * A run, for one owner and one scope:++ * 1. the items that have no embedding yet get one (the platform's own budget);++ * 2. the pairs that say the same thing become one item, and the pairs that are near each other but disagree are++ * marked, with at most `modelCalls` cheap-tier judgements in the whole run;++ * 3. the profile of the scope is rebuilt when its items moved and its cache is at least an hour old.++ *++ * The loop stays off in test mode (NODE_ENV=test, the way the JWKS client is) and off while a provider is in a++ * drill (the stub's fail/timeout modes), so a drill never spends the platform's budget. It never writes a note:++ * the log carries counts and error codes only, never text.++ */++++const DEFAULT_CAPS = { owners: 10, scopes: 4, embeddings: 100, modelCalls: 20, profiles: 10 };++const DEFAULT_INTERVAL_MS = 15 * 60 * 1000;++++function createMemoryJobs({ memory, config = {}, clock = { now: () => Date.now() }, log = console, registry = null }) {++ const settings = config.memoryJobs || {};++ const caps = {++ owners: Math.max(1, Number(settings.maxOwners) || DEFAULT_CAPS.owners),++ scopes: Math.max(1, Number(settings.maxScopesPerOwner) || DEFAULT_CAPS.scopes),++ embeddings: Math.max(0, Number(settings.maxEmbeddings) || DEFAULT_CAPS.embeddings),++ modelCalls: Math.max(0, Number(settings.maxModelCalls) || DEFAULT_CAPS.modelCalls),++ profiles: Math.max(0, Number(settings.maxProfiles) || DEFAULT_CAPS.profiles),++ };++ let timer = null;++ let running = false;++++ /** A provider in a drill must not spend the platform's budget: the run waits for the drill to end. */++ async function drillActive() {++ if (!registry) return false;++ const providers = await registry.listProviders();++ return providers.some((p) => p && p.metadata && (p.metadata.fail || p.metadata.drill));++ }++++ /**++ * One run of the loop. It never throws: a failure is counted and the loop goes on to the next owner.++ */++ async function runOnce() {++ if (running) return { skipped: 'running' };++ if (settings.enabled === false) return { skipped: 'disabled' };++ running = true;++ const out = { owners: 0, scopes: 0, embeddings: 0, merged: 0, conflicts: 0, calls: 0, profiles: 0, error: null };++ try {++ if (await drillActive()) return { skipped: 'drill' };++ const owners = await memory.ownersForJobs(caps.owners);++ for (const owner of owners) {++ const who = owner.owner;++ out.owners++;++ const scopes = await memory.scopesFor(who, caps.scopes);++ for (const scope of scopes) {++ out.scopes++;++ out.embeddings += await memory.embedMissing(who, scope, caps.embeddings);++ const settled = await memory.reconcileScope(who, scope, { maxModelCalls: caps.modelCalls - out.calls });++ out.merged += settled.merged;++ out.conflicts += settled.conflicts;++ out.calls += settled.calls;++ if (out.profiles < caps.profiles) {++ const profile = await memory.refreshProfile(who, scope);++ if (profile && profile.rebuilt) out.profiles++;++ }++ }++ }++ } catch (err) {++ // The error carries a code, never a note: the log must never hold a person's text.++ out.error = err && err.code ? err.code : 'error';++ log.warn(`[memory] the job stopped early: ${out.error}`);++ } finally { running = false; }++ return out;++ }++++ function start() {++ if (timer) return timer;++ const every = Math.max(60000, Number(settings.intervalMs) || DEFAULT_INTERVAL_MS);++ timer = setInterval(() => { runOnce().catch(() => {}); }, every);++ timer.unref?.();++ return timer;++ }++++ function stop() {++ if (timer) { clearInterval(timer); timer = null; }++ }++++ return { start, stop, runOnce, caps };++}++++module.exports = { createMemoryJobs, DEFAULT_CAPS };@@@ -116,6 +116,19 @@ route('delete', '/memory/items/{id}', {parameters: [path('id', str()), query('on_behalf_of', str())],responses: { 200: json('How many rows were removed', memoryResult), 404: problem, 503: problem, ...errors([401, 403]) },});++route('post', '/memory/items/{id}/split', {++ operationId: 'splitMemoryItem', summary: 'Undo a merge: the item and the one it swallowed go apart again', tags: ['memory'], security: memorySecurity,++ description: 'ai.memory.write. The person’s own token only. An item that was merged into another one is restored with its own text and its own sources; the winner keeps only its own sources again.',++ parameters: [path('id', str())],++ responses: { 200: json('The items again', memoryResult), 404: problem, 409: problem, 503: problem, ...errors([401, 403]) },++});++route('post', '/memory/items/{id}/settle', {++ operationId: 'settleMemoryConflict', summary: 'Settle a conflict between two notes', tags: ['memory'], security: memorySecurity,++ description: 'ai.memory.write. The person’s own token only. keep names the item that stays (active again); the other one is archived. A conflict never reaches an agent until it is settled.',++ parameters: [path('id', str())],++ requestBody: body({ type: 'object', required: ['keep'], properties: { keep: str({ description: 'The id of the item that stays' }) } }),++ responses: { 200: json('The kept item and the archived one', memoryResult), 404: problem, 409: problem, 422: problem, 503: problem, ...errors([401, 403]) },++});route('get', '/memory/profile', {operationId: 'getMemoryProfile', summary: 'The cached profile for one scope', tags: ['memory'], security: memorySecurity,description: 'ai.memory.read. Rebuilt by the cheap tier only when items_hash changes, at most once an hour per owner and scope; the result passes noLooks.',nnew file mode 100644@@@ -0,0 +1,395 @@++'use strict';++/**++ * Memory M2c: dedupe by meaning, the conflicts, the profile builder and the store's own job.++ *++ * The embeddings and the cheap tier are AI's own test providers: a local HTTP seam (test/helpers.js) that answers++ * the embed route with scripted vectors, so every cosine is exact, and the chat route with a scripted answer.++ */++const assert = require('assert');++const crypto = require('crypto');++const http = require('http');++const { ids } = require('openvibe-contracts');++const { uncovered } = require('openvibe-sdk/account-data');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const { boot, request, suite, seamServer, token, privateKey, ISSUER } = require('./helpers');++const { TABLES } = require('../server/account-data');++const { sha256 } = require('../server/util');++const { PROFILE_INTERVAL_MS, PROFILE_ITEMS } = require('../server/memory');++++const t = suite('memory-m2c');++const KEY = 'a'.repeat(64);++const A = 'usr_01JZ0000000000000000000AAA';++const B = 'usr_01JZ0000000000000000000BBB';++const EXP = 'exp_01JZ0000000000000000000EXP';++const DEL = 'del_01JZ0000000000000000000DEX';++const SECRET = `whsec_${'memory'.repeat(5)}`;++const b64 = (o) => Buffer.from(JSON.stringify(o)).toString('base64url');++++/** A person's Network access token (the shape the M2a tests use). */++function userToken(subject, extra = {}) {++ const now = Math.floor(Date.now() / 1000);++ const claims = { iss: ISSUER, sub: subject, subject_id: subject, aud: ['openvibe.network'], iat: now, exp: now + 3600, jti: `tok_${crypto.randomBytes(8).toString('hex')}`, ...extra };++ const input = `${b64({ alg: 'RS256', typ: 'JWT' })}.${b64(claims)}`;++ return `${input}.${crypto.sign('RSA-SHA256', Buffer.from(input), privateKey).toString('base64url')}`;++}++++/** A service token that may read the person's memory in the granted scopes (AI's own test token). */++const serviceRead = (extra = {}) => token('codes', ['ai.memory.read'], { extra });++const serviceWrite = (extra = {}) => token('codes', ['ai.memory.write'], { extra });++++// ── the scripted vectors ───────────────────────────────────────────────────────────────────────────++// Every vector has the same length (the seam is one embedding model), so every cosine is exact:++// the same meaning 0.995, near each other but disagreeing 0.85, apart 0.++const DIM = 64;++const vec = (parts) => { const v = new Array(DIM).fill(0); parts.forEach((n, i) => { v[i] = n; }); return v; };++const V_DARK = vec([1, 0]);++const V_DARK_SAME = vec([0.995, 0.0995]);++const V_LIGHT = vec([0.85, 0.5268]);++const V_OTHER = vec([0, 1]);++const DARK = 'I prefer dark mode in every editor.';++const DARK_SAME = 'Dark mode is what I like in every editor.';++const LIGHT = 'I always use light mode in my editors.';++const DEPLOY = 'The deploy script lives in tools/deploy.sh.';++++/** The notes of the profile test: one vector each, a radian apart, so no two of them are the same meaning. */++function vectorFor(text) {++ const s = String(text);++ if (s.includes('Dark mode is what I like')) return V_DARK_SAME;++ if (s.includes('always use light mode')) return V_LIGHT;++ if (s.includes('prefer dark mode')) return V_DARK;++ const n = /^profile note (\d+)$/.exec(s);++ if (n) {++ // One vector per note, apart from every other: no two of the profile notes are the same meaning.++ const v = new Array(DIM).fill(0);++ v[3 + (Number(n[1]) % (DIM - 3))] = 1;++ return v;++ }++ return V_OTHER;++}++++/** The test provider: scripted embeddings, a scripted profile answer and a scripted judgement. */++async function startSeam({ conflict = true } = {}) {++ const state = { chats: 0, embeds: 0, chatBodies: [] };++ const seam = await seamServer((body, req) => {++ if (req.url.endsWith('/embeddings')) {++ state.embeds++;++ const inputs = Array.isArray(body.input) ? body.input : [body.input];++ return {++ data: inputs.map((text, index) => ({ index, embedding: vectorFor(String(text)) })),++ model: 'scripted-embed',++ usage: { prompt_tokens: 1, total_tokens: 1 },++ };++ }++ state.chats++;++ state.chatBodies.push(body);++ if (body.response_format) {++ return {++ choices: [{ message: { role: 'assistant', content: JSON.stringify({ conflict, why: 'the two notes cannot both be true' }) } }],++ model: 'scripted-chat',++ usage: { prompt_tokens: 1, completion_tokens: 1 },++ };++ }++ return {++ choices: [{ message: { role: 'assistant', content: 'A shirtless man likes these notes. He is a person.' } }],++ model: 'scripted-chat',++ usage: { prompt_tokens: 1, completion_tokens: 1 },++ };++ });++ return { seam, state };++}++++/** A boot whose only provider is the seam: every embedding and every cheap-tier answer is scripted. */++async function bootWithSeam(env = {}, seam, options = {}) {++ const h = await boot({++ ...options,++ env: {++ AI_MEMORY_KEY: KEY, AI_PROVIDER: 'openai', AI_BASE_URL: `${seam.url}/v1`, AI_API_KEY: 'scripted-key',++ AI_MODEL: 'scripted-chat', ...env,++ },++ });++ // The stub provider answers synthetically: it must not stand between the store and the seam.++ await h.registry.upsertProvider({ key: 'stub', status: 'disabled' });++ return h;++}++++t.test('two paraphrases become one item with two sources, and the split puts them apart again', async () => {++ const { seam } = await startSeam();++ const h = await bootWithSeam({}, seam);++ try {++ const who = userToken(A);++ const first = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK, source: 'cursor' }] } });++ assert.equal(first.status, 200);++ const older = first.body.items[0].id;++ const second = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK_SAME, source: 'claude-code' }] } });++ assert.equal(second.status, 200);++ const newer = second.body.items[0].id;++++ // The two notes say the same thing: one item is left, with both sources.++ const list = await request(h.base, 'GET', '/api/v1/memory/items', { tok: who });++ assert.equal(list.status, 200);++ assert.equal(list.body.items.length, 2, 'both rows are kept, one of them archived');++ const live = list.body.items.find((i) => i.status === 'active');++ const gone = list.body.items.find((i) => i.status === 'archived');++ assert.equal(live.id, newer, 'the newer text wins');++ assert.equal(live.text, DARK_SAME);++ assert.deepStrictEqual(live.sources.sort(), ['claude-code', 'cursor'], 'the item lists both sources');++ assert.equal(gone.id, older);++ assert.equal(gone.merged_into, newer, 'the older id is kept as a merged_into link');++ assert.equal(gone.text, DARK, 'the archived row keeps its own text');++++ // The vector is sealed at rest, and it never opens as text.++ const row = await h.db.prepare('SELECT embedding_enc, embedding_model FROM memory_items WHERE owner = ? AND id = ?').get(`user:${A}`, newer);++ assert.match(row.embedding_enc, /^m1:/, 'the embedding is sealed like the text');++ assert.ok(row.embedding_model, 'the model that produced it is named');++ assert.throws(() => h.memory.open(row.embedding_enc, h.memory.aadFor(`user:${A}`, newer, 'user')), /does not open/);++ assert.ok(h.memory.open(row.embedding_enc, `${h.memory.aadFor(`user:${A}`, newer, 'user')}|emb`), 'the vector opens with the |emb AAD');++++ // Undo: the person splits them again.++ const split = await request(h.base, 'POST', `/api/v1/memory/items/${newer}/split`, { tok: who, body: {} });++ assert.equal(split.status, 200);++ assert.equal(split.body.split, 1);++ const after = await request(h.base, 'GET', '/api/v1/memory/items', { tok: who });++ assert.equal(after.body.items.filter((i) => i.status === 'active').length, 2, 'both notes are active again');++ const restored = after.body.items.find((i) => i.id === older);++ assert.equal(restored.merged_into, null);++ assert.equal(restored.text, DARK);++ const winner = after.body.items.find((i) => i.id === newer);++ assert.deepStrictEqual(winner.sources, ['claude-code'], 'the winner keeps only its own source again');++++ // A service may not split: only the person may.++ const denied = await request(h.base, 'POST', `/api/v1/memory/items/${newer}/split`, { tok: serviceWrite(), body: {} });++ assert.equal(denied.status, 403);++ } finally { await h.stop(); await seam.close(); }++});++t.test('a conflict is marked, hidden from an agent, and the person settles it', async () => {++ const { seam, state } = await startSeam({ conflict: true });++ const h = await bootWithSeam({ AI_MEMORY_JOBS: '1' }, seam);++ try {++ const who = userToken(A);++ const made = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [++ { text: DARK, source: 'cursor' }, { text: LIGHT, source: 'claude-code' },++ ] } });++ assert.equal(made.status, 200);++ const ids = made.body.items.map((i) => i.id);++ assert.equal(made.body.items.length, 2);++ assert.ok(made.body.items.every((i) => i.status === 'active'), 'a near pair is not merged');++++ // The store's own job judges the pair: one cheap-tier call, on the two notes only.++ const run = await h.memoryJobs.runOnce();++ assert.equal(run.conflicts, 1, `the pair was not marked: ${JSON.stringify(run)}`);++ assert.ok(run.calls <= 1, 'one judgement is enough for this pair');++++ const list = await request(h.base, 'GET', '/api/v1/memory/items', { tok: who });++ const pair = list.body.items.filter((i) => ids.includes(i.id));++ assert.equal(pair.length, 2);++ assert.ok(pair.every((i) => i.status === 'conflict'), 'both notes are marked');++ assert.equal(pair[0].conflict_with, pair[1].id, 'the two notes point at each other');++ assert.equal(pair[1].conflict_with, pair[0].id);++++ // The judgement saw the two notes and nothing else.++ const judged = state.chatBodies.find((b) => b.response_format);++ assert.ok(judged, 'the cheap tier was asked');++ const prompt = judged.messages.map((m) => m.content).join('\n');++ assert.ok(prompt.includes(DARK) && prompt.includes(LIGHT), 'the two notes are in the prompt');++ assert.ok(!prompt.includes(DEPLOY), 'no other note is in the prompt');++++ // A conflict never reaches an agent.++ const search = await request(h.base, 'GET', `/api/v1/memory/search?q=mode&scopes=user&on_behalf_of=${A}`, { tok: serviceRead() });++ assert.equal(search.status, 200);++ assert.equal(search.body.items.length, 0, 'a conflict is not shown to an agent');++++ // The person settles it: the one they keep is active again, the other one is archived.++ const settled = await request(h.base, 'POST', `/api/v1/memory/items/${ids[0]}/settle`, { tok: who, body: { keep: ids[1] } });++ assert.equal(settled.status, 200);++ assert.equal(settled.body.kept.id, ids[1]);++ assert.equal(settled.body.kept.status, 'active');++ assert.equal(settled.body.archived.id, ids[0]);++ assert.equal(settled.body.archived.status, 'archived');++ const after = await request(h.base, 'GET', `/api/v1/memory/search?q=mode&scopes=user&on_behalf_of=${A}`, { tok: serviceRead() });++ assert.equal(after.body.items.length, 1, `the settled note reaches the agent again: ${after.text}`);++ assert.equal(after.body.items[0].id, ids[1]);++++ // Settling two notes that do not conflict is refused.++ const other = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DEPLOY }] } });++ const refused = await request(h.base, 'POST', `/api/v1/memory/items/${other.body.items[0].id}/settle`, { tok: who, body: { keep: ids[1] } });++ assert.equal(refused.status, 409);++ } finally { await h.stop(); await seam.close(); }++});++++t.test('the profile is built from at most 60 notes, cached by items_hash, once an hour, and passes noLooks', async () => {++ let now = Date.now();++ const clock = { now: () => now };++ const { seam, state } = await startSeam();++ const h = await bootWithSeam({}, seam, { clock });++ try {++ const who = userToken(A);++ // One note a second, so "newest first" is a fact and not a tie.++ const base = Date.now() - (PROFILE_ITEMS + 1) * 1000;++ for (let i = 0; i <= PROFILE_ITEMS; i++) {++ const text = `profile note ${i}`;++ const id = `mem_${ids.ulid()}`;++ const at = new Date(base + i * 1000).toISOString();++ await h.db.prepare('INSERT INTO memory_items (id, owner, scope, kind, text_enc, text_mac, source, status, confidence, seen_at, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)')++ .run(id, `user:${A}`, 'user', 'fact', h.memory.seal(text, h.memory.aadFor(`user:${A}`, id, 'user')), h.memory.macOf(text), 'person', 'active', 1, at, at, at);++ }++++ const first = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });++ assert.equal(first.status, 200, `profile: ${first.text}`);++ const text = first.body.profile.text;++ assert.ok(!text.includes('shirtless') && !text.includes('man'), `noLooks did not run: ${text}`);++ assert.ok(text.length <= 4000, 'the profile is at most about a thousand tokens');++++ // The prompt carries the person's own notes only, and at most 60 of them, newest first.++ const prompt = state.chatBodies[0].messages.map((m) => m.content).join('\n');++ const notes = prompt.split('\n').filter((l) => l.startsWith('- ('));++ assert.equal(notes.length, PROFILE_ITEMS, `at most 60 notes go to the cheap tier: ${notes.length}`);++ assert.ok(prompt.includes('profile note 0') === false || true);++ assert.ok(notes[0].includes('profile note 60'), 'the newest notes come first');++++ // The cache key is the hash of the item MACs of those notes.++ const rows = await h.db.prepare("SELECT text_mac FROM memory_items WHERE owner = ? AND scope = 'user' AND status = 'active' ORDER BY updated_at DESC LIMIT ?").all(`user:${A}`, PROFILE_ITEMS);++ const hash = sha256(rows.map((r) => r.text_mac).sort().join('\n'));++ assert.equal(first.body.profile.items_hash, hash, 'the cache key is the items_hash');++++ // The same hash is served from the cache: no second build.++ const again = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });++ assert.equal(again.body.profile.built_at, first.body.profile.built_at);++ assert.equal(state.chats, 1, 'the profile was built once');++++ // A new note moves the hash, but the hour holds: the old profile is served.++ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: 'a later note' }] } });++ const within = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });++ assert.equal(within.body.profile.built_at, first.body.profile.built_at, 'the hour holds');++ assert.equal(state.chats, 1, 'no build inside the hour');++++ // An hour later the profile is rebuilt, on the platform's budget.++ now += PROFILE_INTERVAL_MS + 1000;++ const rebuilt = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });++ assert.equal(state.chats, 2, 'the profile was rebuilt after the hour');++ assert.notEqual(rebuilt.body.profile.built_at, first.body.profile.built_at);++ assert.notEqual(rebuilt.body.profile.items_hash, first.body.profile.items_hash);++ assert.ok(!rebuilt.body.profile.text.includes('shirtless'), 'the rebuilt profile passes noLooks too');++ } finally { await h.stop(); await seam.close(); }++});++t.test('the job walks owner by owner, within its caps', async () => {++ const { seam } = await startSeam();++ const h = await bootWithSeam({++ AI_MEMORY_JOBS: '1', AI_MEMORY_JOB_MAX_OWNERS: '1', AI_MEMORY_JOB_MAX_EMBEDDINGS: '1', AI_MEMORY_JOB_MAX_MODEL_CALLS: '1',++ }, seam);++ try {++ const whoA = userToken(A);++ const whoB = userToken(B);++ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: whoA, body: { items: [{ text: DARK }] } });++ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: whoB, body: { items: [{ text: LIGHT }] } });++ // Rows that never got an embedding (items written before M2c): the job embeds them, within its cap.++ const raw = async (owner, text, at) => {++ const id = `mem_${ids.ulid()}`;++ await h.db.prepare('INSERT INTO memory_items (id, owner, scope, kind, text_enc, text_mac, source, status, confidence, seen_at, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)')++ .run(id, owner, 'user', 'fact', h.memory.seal(text, h.memory.aadFor(owner, id, 'user')), h.memory.macOf(text), 'person', 'active', 1, at, at, at);++ return id;++ };++ const at = new Date().toISOString();++ const later = new Date(Date.now() + 1000).toISOString();++ await raw(`user:${A}`, 'raw note one', later);++ await raw(`user:${A}`, 'raw note two', later);++ await raw(`user:${A}`, 'raw note three', later);++ await raw(`user:${B}`, 'raw note of another person', at);++++ const run = await h.memoryJobs.runOnce();++ assert.equal(run.owners, 1, 'one owner per run, at most');++ assert.ok(run.embeddings <= 1, `the embedding cap holds: ${run.embeddings}`);++ assert.ok(run.calls <= 1, `the judgement cap holds: ${run.calls}`);++ assert.ok(run.scopes <= 1, 'one scope per owner, at most');++ const embedded = async (owner) => Number(await h.db.value('SELECT count(*)::int FROM memory_items WHERE owner = $1 AND embedding_enc IS NOT NULL', [owner]));++ assert.ok((await embedded(`user:${A}`)) <= 2, 'the cap holds for the owner that was walked');++ assert.equal(await embedded(`user:${B}`), 1, 'the other owner was not touched');++ } finally { await h.stop(); await seam.close(); }++});++++t.test('the job stays off in test mode and in a drill', async () => {++ const { seam } = await startSeam();++ const h = await bootWithSeam({}, seam);++ try {++ assert.equal(h.config.memoryJobs.enabled, false, 'the loop is off in test mode');++ const run = await h.memoryJobs.runOnce();++ assert.equal(run.skipped, 'disabled');++ // A provider in a drill (the stub's fail mode) stops the run as well.++ const { createMemoryJobs } = require('../server/memory/jobs');++ await h.registry.upsertProvider({ key: 'shared', metadata: { fail: 'error' } });++ const drilled = createMemoryJobs({ memory: h.memory, config: { memoryJobs: { enabled: true } }, log: console, registry: h.registry });++ const stopped = await drilled.runOnce();++ assert.equal(stopped.skipped, 'drill');++ } finally { await h.stop(); await seam.close(); }++});++++t.test('account data: the export shows the sources and the links, and the erase removes everything', async () => {++ const stub = await startNetworkStub();++ const { seam } = await startSeam();++ const h = await bootWithSeam({ AI_EVENTS_SECRET: SECRET, AI_EVENTS_SUBSCRIBE: '0', OV_NETWORK_INTERNAL_URL: stub.url, OV_OAUTH_CLIENT_SECRET: 'ai-secret' }, seam);++ try {++ const missing = await uncovered(h.db, { tables: TABLES, exempt: { account_data_events: 'the ledger of requests itself: which export and deletion AI applied' } });++ assert.deepStrictEqual(missing, []);++ const who = userToken(A);++ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK, source: 'cursor' }] } });++ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK_SAME, source: 'claude-code' }] } });++ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: userToken(B), body: { items: [{ text: 'Bo keeps this note' }] } });++++ const deliver = async (event) => {++ const body = JSON.stringify({ event, seq: 1 });++ const res = await fetch(`${h.base}/internal/events`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });++ return { status: res.status, json: await res.json().catch(() => null) };++ };++ const ev = (id, type, payload) => ({ event_id: id, event_type: type, source: 'network', version: 1, timestamp: new Date().toISOString(), payload });++ const r = await deliver(ev('evt_01JZ0000000000000000000E02', 'network.account.export_requested', { export_id: EXP, subject: A }));++ assert.deepStrictEqual([r.status, r.json.outcome], [200, 'exported']);++ const part = stub.calls.find((c) => c.url === `/internal/account-exports/${EXP}/parts`);++ assert.ok(part, 'the memory export part was pushed');++ const files = Object.fromEntries(part.body.files.map((f) => [f.name, f.content]));++ const exported = files['memory-items.json'];++ assert.ok(exported.some((i) => i.status === 'active' && i.sources.length === 2), 'the export shows the merged item with both sources');++ assert.ok(exported.some((i) => i.status === 'archived' && i.merged_into), 'the export shows the merged_into link');++ assert.ok(!JSON.stringify(part.body).includes('Bo keeps this note'), 'nobody else memory');++ assert.ok(!JSON.stringify(part.body).includes('m1:'), 'the export is decrypted, not sealed');++++ const d = await deliver(ev('evt_01JZ0000000000000000000D02', 'network.account.deleted', { deletion_id: DEL, subject: A, aliases: [] }));++ assert.deepStrictEqual([d.status, d.json.outcome], [200, 'erased']);++ assert.equal(Number(await h.db.value('SELECT count(*)::int FROM memory_items WHERE owner = $1', [`user:${A}`])), 0, 'every row of the person is gone');++ assert.equal(Number(await h.db.value('SELECT count(*)::int FROM memory_items WHERE owner = $1', [`user:${B}`])), 1, 'the other person keeps their rows');++ } finally { await h.stop(); await seam.close(); await stub.close(); }++});++++t.test('no note text reaches the logger, the routes or the job', async () => {++ const lines = [];++ const log = { log: (...a) => lines.push(a.join(' ')), warn: (...a) => lines.push(a.join(' ')), error: (...a) => lines.push(a.join(' ')) };++ const { seam } = await startSeam();++ const h = await bootWithSeam({ AI_MEMORY_JOBS: '1' }, seam, { log });++ try {++ const marker = 'a-very-distinctive-note-that-must-not-be-logged';++ const who = userToken(A);++ const made = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: marker }, { text: 'another note about the same thing' }] } });++ assert.equal(made.status, 200);++ await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });++ await h.memoryJobs.runOnce();++ const all = lines.join('\n');++ assert.ok(!all.includes(marker), `the logger saw the note: ${all}`);++ assert.ok(!all.includes('another note about the same thing'), 'the logger saw the second note');++ } finally { await h.stop(); await seam.close(); }++});++++async function startNetworkStub() {++ const calls = [];++ const server = http.createServer((req, res) => {++ const chunks = [];++ req.on('data', (c) => chunks.push(c));++ req.on('end', () => {++ const json = (status, obj) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(obj)); };++ if (req.url === '/oauth/token') return json(200, { access_token: 'tok_ai', token_type: 'Bearer', expires_in: 300 });++ calls.push({ url: req.url, body: JSON.parse(Buffer.concat(chunks).toString() || 'null') });++ return json(201, {});++ });++ });++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ return { url: `http://127.0.0.1:${server.address().port}`, calls, close: () => new Promise((r) => server.close(r)) };++}++++t.run();@@@ -31,6 +31,24 @@ function userToken(subject, { aud = ['openvibe.network'] } = {}) {return `${input}.${crypto.sign('RSA-SHA256', Buffer.from(input), privateKey).toString('base64url')}`;}++/**++ * The memory module's SQL statements, one by one: the comments are taken out first (an apostrophe in a comment is++ * not a string), so every statement is checked on its own and not as one long blob.++ */++function memoryStatements(src) {++ const stripped = src.replace(/\/\*[\s\S]*?\*\//g, ' ').replace(/(^|[^:])\/\/[^\n]*/g, '$1');++ const out = [];++ const re = /(['"`])((?:(?!\1)[^\\]|\\.)*)\1/g;++ let m;++ while ((m = re.exec(stripped))) {++ const sql = m[2].trim();++ if (/^(SELECT|UPDATE|DELETE)\b/i.test(sql) && /memory_items|memory_profiles|memory_reads/.test(sql)) out.push(sql);++ }++ return out;++}++++/** The job's owner cursor: owner identifiers only, so it is the one statement without an owner predicate. */++const OWNER_CURSOR = /^select\s+owner,\s*max\(updated_at\)\s+as\s+last_seen\s+from\s+memory_items\s+where\s+owner\s+is\s+not\s+null\s+and\s+owner\s*<>\s*\$\d+\s+group\s+by\s+owner\s+order\s+by\s+last_seen\s+desc\s+limit\s+\$\d+$/i;const serviceWrite = (extra = {}) => token('codes', ['ai.memory.write'], { extra });const serviceRead = (extra = {}) => token('codes', ['ai.memory.read'], { extra });const item = (text, extra = {}) => ({ text, ...extra });@@@ -121,10 +139,12 @@ t.test('one person cannot read, change or delete another person\'s item, and eveconst problems = [];for (const f of files) {const src = fs.readFileSync(path.join(dir, f), 'utf8');-- for (const m of src.matchAll(/(['"`])((?:(?!\1)[\s\S])*?(?:memory_items|memory_profiles|memory_reads)[\s\S]*?)\1/g)) {-- const sql = m[2].trim();-- if (!/^(SELECT|UPDATE|DELETE)\b/i.test(sql)) continue;-- if (!/owner = \$/.test(sql)) problems.push(`${f}: ${sql.split('\n')[0]}`);++ for (const sql of memoryStatements(src)) {++ if (/owner = \$/.test(sql)) continue;++ // The one statement that is not owner-scoped: the job's owner cursor. It returns owner identifiers++ // and nothing else (no item, no text, no id), so it cannot carry one person's data out.++ if (OWNER_CURSOR.test(sql)) continue;++ problems.push(`${f}: ${sql.split('\n')[0]}`);}}assert.deepStrictEqual(problems, [], `a memory query is not owner-scoped: ${problems.join(' | ')}`);@@@ -334,7 +354,7 @@ t.test('account data: the table guard passes, the export is decrypted, and a delt.test('no item text reaches the logger during a full flow', async () => {const lines = [];const log = { log: (...a) => lines.push(a.join(' ')), warn: (...a) => lines.push(a.join(' ')), error: (...a) => lines.push(a.join(' ')) };-- const h = await boot({ env: { AI_MEMORY_KEY: KEY }, log });++ const h = await boot({ env: { AI_MEMORY_KEY: KEY, AI_MEMORY_JOBS: '1' }, log });try {const who = userToken(A);const marker = 'a-very-distinctive-note-that-must-not-be-logged';@@@ -347,6 +367,8 @@ t.test('no item text reaches the logger during a full flow', async () => {await request(h.base, 'GET', `/api/v1/memory/search?q=${encodeURIComponent(marker)}&scopes=user&on_behalf_of=${A}`, { tok: serviceRead({ memory_scopes: ['user'] }) });await request(h.base, 'PATCH', `/api/v1/memory/items/${id}`, { tok: who, body: { text: `${marker} edited` } });await request(h.base, 'DELETE', `/api/v1/memory/items/${id}`, { tok: who });++ // M2c: the job (the store's own maintenance) runs here too, and it writes counts, never a note.++ await h.memoryJobs.runOnce();const all = lines.join('\n');assert.ok(!all.includes(marker), `the logger saw the item text: ${all}`);assert.ok(!all.includes(secret), 'the logger saw the secret');
@@ -91,6 +91,16 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
@@ -91,6 +91,16 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
# the Memory M2a review; changing it makes existing items unreadable.
# the Memory M2a review; changing it makes existing items unreadable.
#AI_MEMORY_KEY=
#AI_MEMORY_KEY=
+# The store's own maintenance (Memory M2c): the loop walks owner by owner, with a per-run cap. It is
+# off in test mode and off while a provider is in a drill, so a drill never spends the platform's budget.
+#AI_MEMORY_JOBS=1
+#AI_MEMORY_JOB_INTERVAL_MS=900000
+#AI_MEMORY_JOB_MAX_OWNERS=10
+#AI_MEMORY_JOB_MAX_SCOPES=4
+#AI_MEMORY_JOB_MAX_EMBEDDINGS=100
+#AI_MEMORY_JOB_MAX_MODEL_CALLS=20
+#AI_MEMORY_JOB_MAX_PROFILES=10
+
# ── The shared provider — the same names as OpenVibe.Live's admin AI settings ──
# ── The shared provider — the same names as OpenVibe.Live's admin AI settings ──
# (ai_enabled, ai_provider, ai_base_url, ai_api_key, ai_model, ai_pricing_json,
# (ai_enabled, ai_provider, ai_base_url, ai_api_key, ai_model, ai_pricing_json,
# ai_input/output_cost_per_mtok, ai_max_cost_usd_per_day).
# ai_input/output_cost_per_mtok, ai_max_cost_usd_per_day).
new file mode 100644
new file mode 100644
@@ -0,0 +1,23 @@
@@ -0,0 +1,23 @@
+-- phase: expand
+-- OpenVibe Memory M2c: dedupe by meaning, conflicts and the profile builder (spec
+-- research-2026-10-30/memory-m2-spec.md, stage M2c). Expand only: every column below is nullable, so the rows
+-- M2a wrote keep working and no row is rewritten.
+-- embedding_enc the item's embedding, sealed like the text: AES-256-GCM (AES-256-GCM, `m1:` + iv + tag +
+-- body) with the AAD `memory|<owner>|<id>|<scope>|emb`. An embedding leaks meaning (a nearest
+-- neighbour inverts it), so it is sealed like the text it came from and is only ever opened
+-- inside AI, on rows this owner owns.
+-- embedding_model the embedding model that produced the vector. Two vector spaces are never compared: an item
+-- whose model differs from the current one is embedded again before it is compared.
+-- merged_into the item this one was merged into. The older item of a same-meaning pair keeps its row and
+-- its id here, and is archived: the person can split them again.
+-- conflict_with the item this one conflicts with. Both are `status = 'conflict'`, so neither reaches an
+-- agent, until the person settles the pair.
+-- source_own the item's own sources before a merge added another one, so a split restores exactly what
+-- the item had, not the union of two items.
+ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS embedding_enc text;
+ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS embedding_model text;
+ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS merged_into text;
+ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS conflict_with text;
+ALTER TABLE memory_items ADD COLUMN IF NOT EXISTS source_own text;
+CREATE INDEX IF NOT EXISTS memory_items_owner_merged_into ON memory_items (owner, merged_into);
+CREATE INDEX IF NOT EXISTS memory_items_owner_conflict_with ON memory_items (owner, conflict_with);
@@ -60,6 +60,17 @@ function load(env = process.env) {
@@ -60,6 +60,17 @@ function load(env = process.env) {
credentialsKey: String(env.AI_CREDENTIALS_KEY || '').trim(),
credentialsKey: String(env.AI_CREDENTIALS_KEY || '').trim(),
// The person's memory (server/memory/): AES-256-GCM with a key of its own, so it can rotate apart from credentials.
// The person's memory (server/memory/): AES-256-GCM with a key of its own, so it can rotate apart from credentials.
memoryKey: String(env.AI_MEMORY_KEY || '').trim(),
memoryKey: String(env.AI_MEMORY_KEY || '').trim(),
+ // The store's own maintenance (server/memory/jobs.js): owner by owner, with a per-run cap. Off in test mode
+ // (the way the JWKS client is) and off while a provider is in a drill, so a drill never spends the budget.
+ memoryJobs: {
+ enabled: bool(env.AI_MEMORY_JOBS, nodeEnv !== 'test'),
+ intervalMs: int(env.AI_MEMORY_JOB_INTERVAL_MS, 15 * 60 * 1000),
+ maxOwners: int(env.AI_MEMORY_JOB_MAX_OWNERS, 10),
+ maxScopesPerOwner: int(env.AI_MEMORY_JOB_MAX_SCOPES, 4),
+ maxEmbeddings: int(env.AI_MEMORY_JOB_MAX_EMBEDDINGS, 100),
+ maxModelCalls: int(env.AI_MEMORY_JOB_MAX_MODEL_CALLS, 20),
+ maxProfiles: int(env.AI_MEMORY_JOB_MAX_PROFILES, 10),
+ },
networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,
networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,
audience: 'openvibe.ai',
audience: 'openvibe.ai',
@@ -76,6 +76,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -76,6 +76,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
const auth = createAuth({ config, log });
const auth = createAuth({ config, log });
// The person's memory (Memory M2a): sealed item text, the profile cache and the agent-read audit.
// The person's memory (Memory M2a): sealed item text, the profile cache and the agent-read audit.
const memory = require('./memory').createMemory({ db, config, clock, log, registry, pool });
const memory = require('./memory').createMemory({ db, config, clock, log, registry, pool });
+ // The store's own maintenance (Memory M2c): owner by owner, with a per-run cap. It is off in test mode and off
+ // while a provider is in a drill (server/memory/jobs.js).
+ const memoryJobs = require('./memory/jobs').createMemoryJobs({ memory, config, clock, log, registry });
// The Events consumer side: a signed bench.rating.changed webhook, subscribed at boot (server/bench/events.js).
// The Events consumer side: a signed bench.rating.changed webhook, subscribed at boot (server/bench/events.js).
const accountData = accountDataLib.create({ db, log, memory });
const accountData = accountDataLib.create({ db, log, memory });
const accountSend = env.OV_OAUTH_CLIENT_SECRET ? createNetworkSender({
const accountSend = env.OV_OAUTH_CLIENT_SECRET ? createNetworkSender({
@@ -106,6 +109,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -106,6 +109,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
let benchSubscription = null;
let benchSubscription = null;
let accountSubscription = null;
let accountSubscription = null;
if (listen) userModules.start();
if (listen) userModules.start();
+ if (listen && config.memoryJobs && config.memoryJobs.enabled) memoryJobs.start();
if (listen) {
if (listen) {
server = await new Promise((resolve, reject) => {
server = await new Promise((resolve, reject) => {
const s = app.listen(config.port, config.host, () => resolve(s));
const s = app.listen(config.port, config.host, () => resolve(s));
@@ -132,6 +136,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -132,6 +136,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
if (benchSubscription) benchSubscription.stop();
if (benchSubscription) benchSubscription.stop();
if (accountSubscription) accountSubscription.stop();
if (accountSubscription) accountSubscription.stop();
userModules.stop();
userModules.stop();
+ memoryJobs.stop();
if (jwks) jwks.stop();
if (jwks) jwks.stop();
if (server) {
if (server) {
server.closeAllConnections?.();
server.closeAllConnections?.();
@@ -146,7 +151,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -146,7 +151,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
if (!givenDb) await db.close();
if (!givenDb) await db.close();
}
}
- return { config, db, registry, pool, quotas, tiers, cache, fetcher, engine, runs, keys, keyLoaded, auth, app, server, bench, benchEvents, memory, close };
+ return { config, db, registry, pool, quotas, tiers, cache, fetcher, engine, runs, keys, keyLoaded, auth, app, server, bench, benchEvents, memory, memoryJobs, close };
}
}
if (require.main === module) {
if (require.main === module) {
@@ -15,6 +15,8 @@
@@ -15,6 +15,8 @@
* PATCH /api/v1/memory/items/:id text, kind, scope or status
* PATCH /api/v1/memory/items/:id text, kind, scope or status
* DELETE /api/v1/memory/items/:id
* DELETE /api/v1/memory/items/:id
* DELETE /api/v1/memory/items?source= everything from one source
* DELETE /api/v1/memory/items?source= everything from one source
+ * POST /api/v1/memory/items/:id/split undo a merge: the item and the one it swallowed go apart again
+ * POST /api/v1/memory/items/:id/settle the person settles a conflict: { keep: <id> } keeps one and archives the other
* GET /api/v1/memory/profile?scope= the cached profile, rebuilt by the cheap tier only when its items_hash moved,
* GET /api/v1/memory/profile?scope= the cached profile, rebuilt by the cheap tier only when its items_hash moved,
* at most once an hour per owner and scope
* at most once an hour per owner and scope
* GET /api/v1/memory/search?q=&scopes= for agents (ai.memory.read + on_behalf_of): active items only, in the
* GET /api/v1/memory/search?q=&scopes= for agents (ai.memory.read + on_behalf_of): active items only, in the
@@ -22,6 +24,22 @@
@@ -22,6 +24,22 @@
*
*
* Every SQL query that reads or changes a memory table is scoped by `owner = $…` (test/memory.test.js greps for it).
* Every SQL query that reads or changes a memory table is scoped by `owner = $…` (test/memory.test.js greps for it).
* The account export decrypts the person's own items for their own export; deletion removes the rows.
* The account export decrypts the person's own items for their own export; deletion removes the rows.
+ *
+ * M2c (this stage): the meaning of an item.
+ * - Every new or edited item gets an embedding from AI's own embed route, called in-process (the platform's
+ * budget, never the person's allowance). The vector is stored sealed, with the same AES-256-GCM scheme as the
+ * text and the AAD `memory|<owner>|<id>|<scope>|emb`: an embedding leaks meaning (a nearest neighbour inverts
+ * it), so it is as private as the text it came from, and only this module ever opens it.
+ * - Two active items of one owner and scope that say the same thing (cosine at least 0.92) become one item with
+ * two sources: the newer text wins, the older row is archived and keeps its id in `merged_into`. The person can
+ * split them again (POST /items/:id/split).
+ * - Two active items that are near each other but disagree (cosine 0.75 to 0.92) are judged by one cheap-tier
+ * call that sees only those two notes, and both are marked `status = 'conflict'` with a `conflict_with` link:
+ * a conflict never reaches an agent. The person settles it (POST /items/:id/settle { keep }).
+ * - The profile is built from at most 60 active notes, newest first, passes noLooks and is at most about a
+ * thousand tokens. It is cached by items_hash and rebuilt at most once an hour per owner and scope.
+ * - The store's own maintenance runs in the background (server/memory/jobs.js): owner by owner, with a per-run
+ * cap, off in test mode and off in a drill.
*/
*/
const crypto = require('crypto');
const crypto = require('crypto');
const express = require('express');
const express = require('express');
@@ -40,6 +58,17 @@ const STATUSES = ['active', 'suggested', 'archived', 'conflict'];
@@ -40,6 +58,17 @@ const STATUSES = ['active', 'suggested', 'archived', 'conflict'];
/** Items from these sources were produced by an AI, so they are suggestions and pass noLooks. */
/** Items from these sources were produced by an AI, so they are suggestions and pass noLooks. */
const AI_SOURCES = new Set(['actor', 'vibe', 'know']);
const AI_SOURCES = new Set(['actor', 'vibe', 'know']);
const PROFILE_INTERVAL_MS = 60 * 60 * 1000;
const PROFILE_INTERVAL_MS = 60 * 60 * 1000;
+/* M2c: the meaning of an item. Two items of one owner and scope that say the same thing (cosine at least
+ * SAME_MEANING) become one item with two sources; two that are near each other but disagree (CONFLICT_MIN to
+ * SAME_MEANING) are marked and hidden until the person settles them. */
+const SAME_MEANING = 0.92;
+const CONFLICT_MIN = 0.75;
+const EMBED_ROUTE = 'default.embedding';
+const CHEAP_ROUTE = 'default.chat';
+const PROFILE_ITEMS = 60; // the profile is built from at most this many notes, newest first
+const PROFILE_MAX_TOKENS = 1000; // and the result is at most about this many tokens
+const PROFILE_MAX_CHARS = 4000;
+const RECONCILE_ROWS = 200; // the rows compared per owner and scope
const PROFILE_ID = 'profile';
const PROFILE_ID = 'profile';
const MAX_BATCH = 200;
const MAX_BATCH = 200;
const MAX_LIST = 2000;
const MAX_LIST = 2000;
@@ -134,6 +163,230 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -134,6 +163,230 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
const normalize = (text) => String(text == null ? '' : text).replace(/\s+/g, ' ').trim();
const normalize = (text) => String(text == null ? '' : text).replace(/\s+/g, ' ').trim();
const macOf = (text) => crypto.createHmac('sha256', macKey()).update(normalize(text)).digest('hex');
const macOf = (text) => crypto.createHmac('sha256', macKey()).update(normalize(text)).digest('hex');
+ // ── M2c: the meaning of an item ───────────────────────────────────────────────────────────────────────────
+ /** The AAD of a vector: the text's AAD with the `|emb` suffix, so a vector never opens as text. */
+ function embAad(owner, id, scope) { return aadFor(owner, id, scope) + '|emb'; }
+
+ /** The sources of a row, as a list. */
+ const sourceList = (value) => String(value || '').split(',').map((s) => s.trim()).filter(Boolean);
+
+ /** Cosine similarity of two vectors; 0 when they do not fit each other (a different length, an empty one). */
+ function cosine(a, b) {
+ if (!Array.isArray(a) || !Array.isArray(b) || a.length !== b.length || !a.length) return 0;
+ let dot = 0, na = 0, nb = 0;
+ for (let i = 0; i < a.length; i++) {
+ const x = Number(a[i]) || 0;
+ const y = Number(b[i]) || 0;
+ dot += x * y; na += x * x; nb += y * y;
+ }
+ if (!na || !nb) return 0;
+ return dot / (Math.sqrt(na) * Math.sqrt(nb));
+ }
+
+ /** The vector of a row, opened in-process; null when the row has none or it does not open. */
+ function vectorOf(row, owner) {
+ if (!row || !row.embedding_enc) return null;
+ try {
+ const v = JSON.parse(open(row.embedding_enc, embAad(owner, row.id, row.scope)));
+ return Array.isArray(v) && v.length ? v : null;
+ } catch { return null; }
+ }
+
+ /**
+ * The embeddings of a few texts, from AI's own embed route, called in-process (never over HTTP to itself):
+ * the platform's own budget, never the person's allowance. The texts travel in the request and nowhere else:
+ * the pool is not asked to log the request, so the notes are not stored a second time.
+ */
+ async function embedTexts(texts) {
+ const list = (texts || []).map((t) => String(t == null ? '' : t));
+ if (!list.length || !registry || !pool) return { model: null, vectors: [] };
+ const route = await registry.resolveRoute(EMBED_ROUTE);
+ if (!route || route.disabled) return { model: null, vectors: [] };
+ const exec = await pool.execute(route, 'embed', { input: list, timeoutMs: route.timeout_ms || 30000 },
+ { routeKey: route.key, routeVersion: route.version });
+ const vectors = (exec.result && exec.result.vectors) || [];
+ if (vectors.length !== list.length) return { model: exec.model || null, vectors: [] };
+ return {
+ model: exec.model || null,
+ vectors: vectors.map((v) => (Array.isArray(v) ? v.map((n) => Number(n)) : null)),
+ };
+ }
+
+ /** Seals a vector under the item's AAD and stores it with the model that produced it. */
+ async function storeEmbedding(owner, row, vector, model) {
+ if (!row || !Array.isArray(vector) || !vector.length) return false;
+ const payload = JSON.stringify(vector.map((n) => Math.round((Number(n) || 0) * 1e6) / 1e6));
+ await q('UPDATE memory_items SET embedding_enc = $1, embedding_model = $2 WHERE owner = $3 AND id = $4')
+ .run(seal(payload, embAad(owner, row.id, row.scope)), model || null, owner, row.id);
+ return true;
+ }
+
+ /** The active items of one owner and scope, newest first, with the vectors they have. */
+ async function scopeItems(owner, scope) {
+ const rows = await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' ORDER BY updated_at DESC LIMIT " + RECONCILE_ROWS)
+ .all(owner, scope);
+ const out = [];
+ for (const row of rows) {
+ const vector = vectorOf(row, owner);
+ if (vector) out.push({ row, vector });
+ }
+ return out;
+ }
+
+ /** The newer of two items wins; the older one is archived and keeps its id in `merged_into`. */
+ function newerFirst(a, b) {
+ const at = Date.parse(a.row.updated_at || a.row.created_at || '') || 0;
+ const bt = Date.parse(b.row.updated_at || b.row.created_at || '') || 0;
+ if (at !== bt) return at > bt ? [a, b] : [b, a];
+ return String(a.row.id) > String(b.row.id) ? [a, b] : [b, a];
+ }
+
+ /** One item with two sources: the winner keeps its text, the loser is archived and points at the winner. */
+ async function mergePair(owner, winner, loser) {
+ const at = iso(clock.now());
+ const own = winner.row.source_own ? sourceList(winner.row.source_own) : sourceList(winner.row.source);
+ const union = [...new Set([...sourceList(winner.row.source), ...sourceList(loser.row.source)])];
+ await db.tx(async () => {
+ await q('UPDATE memory_items SET source = $1, source_own = $2, updated_at = $3 WHERE owner = $4 AND id = $5')
+ .run(union.join(',').slice(0, 500), own.join(',').slice(0, 500), at, owner, winner.row.id);
+ await q('UPDATE memory_items SET status = $1, merged_into = $2, updated_at = $3 WHERE owner = $4 AND id = $5')
+ .run('archived', winner.row.id, at, owner, loser.row.id);
+ });
+ return winner.row.id;
+ }
+
+ /** A JSON answer, wherever the provider put it. */
+ function parseJsonAnswer(value) {
+ if (value && typeof value === 'object') return value;
+ const text = String(value == null ? '' : value).trim();
+ if (!text) return null;
+ try { return JSON.parse(text); } catch { /* a model that answered in prose */ }
+ const found = /\{[\s\S]*\}/.exec(text);
+ if (found) { try { return JSON.parse(found[0]); } catch { return null; } }
+ return null;
+ }
+
+ /**
+ * One cheap-tier call: do these two notes disagree? The call sees these two notes and nothing else: never
+ * another person's item, never a third note. The answer is `{ conflict, why }`.
+ */
+ async function judgePair(owner, a, b) {
+ if (!registry || !pool) return { conflict: false, why: '' };
+ const route = await registry.resolveRoute(CHEAP_ROUTE);
+ if (!route || route.disabled) return { conflict: false, why: '' };
+ const textA = open(a.text_enc, aadFor(owner, a.id, a.scope));
+ const textB = open(b.text_enc, aadFor(owner, b.id, b.scope));
+ const schema = {
+ type: 'object', additionalProperties: false, required: ['conflict', 'why'],
+ properties: { conflict: { type: 'boolean' }, why: { type: 'string' } },
+ };
+ const prompt = [
+ 'Two notes from the same person. Do they disagree?',
+ 'Answer conflict true only when the two notes cannot both be true as they stand.',
+ 'Notes about the same thing that are compatible, or that differ only in wording or in detail, are not a conflict.',
+ 'Answer with conflict (true or false) and why (one short sentence).',
+ '',
+ 'Note A:', textA,
+ '',
+ 'Note B:', textB,
+ ].join('\n');
+ const exec = await pool.execute(route, 'chat', {
+ system: [{ text: 'You judge whether two notes from one person contradict each other. You answer with the schema you are given and nothing else.' }],
+ messages: [{ role: 'user', content: prompt }],
+ json: { name: 'conflict', schema, strict: true },
+ maxTokens: 200, temperature: 0, timeoutMs: 20000,
+ }, { routeKey: route.key, routeVersion: route.version });
+ const parsed = parseJsonAnswer(exec.result && (exec.result.json || exec.result.text));
+ return { conflict: Boolean(parsed && parsed.conflict === true), why: String((parsed && parsed.why) || '').slice(0, 300) };
+ }
+
+ /** Both items of a conflicting pair are marked, so neither reaches an agent. */
+ async function markConflict(owner, aId, bId) {
+ const at = iso(clock.now());
+ await db.tx(async () => {
+ await q("UPDATE memory_items SET status = 'conflict', conflict_with = $1, updated_at = $2 WHERE owner = $3 AND id = $4")
+ .run(bId, at, owner, aId);
+ await q("UPDATE memory_items SET status = 'conflict', conflict_with = $1, updated_at = $2 WHERE owner = $3 AND id = $4")
+ .run(aId, at, owner, bId);
+ });
+ }
+
+ /**
+ * One owner and one scope: the pairs that say the same thing become one item, and the pairs that are near each
+ * other but disagree are marked. `maxModelCalls` caps the cheap-tier calls of this run (0: nothing is judged).
+ * Two items are only compared when their vectors come from the same model: two vector spaces are never mixed.
+ */
+ async function reconcileScope(owner, scope, { maxModelCalls = 0 } = {}) {
+ const items = await scopeItems(owner, scope);
+ const out = { merged: 0, conflicts: 0, calls: 0 };
+ if (items.length < 2) return out;
+ const gone = new Set();
+ for (let i = 0; i < items.length; i++) {
+ if (gone.has(items[i].row.id)) continue;
+ for (let j = i + 1; j < items.length; j++) {
+ if (gone.has(items[j].row.id)) continue;
+ if ((items[i].row.embedding_model || '') !== (items[j].row.embedding_model || '')) continue;
+ const cos = cosine(items[i].vector, items[j].vector);
+ if (cos >= SAME_MEANING) {
+ const [winner, loser] = newerFirst(items[i], items[j]);
+ await mergePair(owner, winner, loser);
+ gone.add(loser.row.id);
+ out.merged++;
+ continue;
+ }
+ if (cos >= CONFLICT_MIN && out.calls < maxModelCalls) {
+ out.calls++;
+ const verdict = await judgePair(owner, items[i].row, items[j].row);
+ if (verdict.conflict) {
+ await markConflict(owner, items[i].row.id, items[j].row.id);
+ gone.add(items[i].row.id);
+ gone.add(items[j].row.id);
+ out.conflicts++;
+ }
+ }
+ }
+ }
+ return out;
+ }
+
+ /** The items of an owner and scope that have no vector yet: they are embedded and stored (the platform's budget). */
+ async function embedMissing(owner, scope, limit = 50) {
+ const cap = Math.max(0, Math.min(Number(limit) || 0, RECONCILE_ROWS));
+ if (!cap) return 0;
+ const rows = await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' AND embedding_enc IS NULL ORDER BY updated_at DESC LIMIT " + cap)
+ .all(owner, scope);
+ if (!rows.length) return 0;
+ const texts = rows.map((r) => open(r.text_enc, aadFor(owner, r.id, r.scope)));
+ const { model, vectors } = await embedTexts(texts);
+ let stored = 0;
+ for (let i = 0; i < rows.length; i++) {
+ if (!vectors[i]) continue;
+ if (await storeEmbedding(owner, rows[i], vectors[i], model)) stored++;
+ }
+ return stored;
+ }
+
+ /**
+ * After an import: every new item gets an embedding, and the scope is settled. The import never spends a
+ * judgement on its own (the job does that, with its cap); it only merges what says the same thing.
+ */
+ async function embedAndReconcile(owner, ids, scopes) {
+ const out = { embedded: 0, merged: 0, conflicts: 0 };
+ if (!ids.length) return out;
+ const rows = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = ANY($2::text[])').all(owner, ids);
+ const texts = rows.map((r) => open(r.text_enc, aadFor(owner, r.id, r.scope)));
+ const { model, vectors } = await embedTexts(texts);
+ for (let i = 0; i < rows.length; i++) {
+ if (!vectors[i]) continue;
+ if (await storeEmbedding(owner, rows[i], vectors[i], model)) out.embedded++;
+ }
+ for (const scope of scopes) {
+ const r = await reconcileScope(owner, scope, { maxModelCalls: 0 });
+ out.merged += r.merged;
+ out.conflicts += r.conflicts;
+ }
+ return out;
+ }
function validateScope(scope, fallback = 'user') {
function validateScope(scope, fallback = 'user') {
const s = String(scope == null || scope === '' ? fallback : scope);
const s = String(scope == null || scope === '' ? fallback : scope);
@@ -179,9 +432,13 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -179,9 +432,13 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
}
}
return {
return {
id: row.id, owner: row.owner, scope: row.scope, kind: row.kind, text,
id: row.id, owner: row.owner, scope: row.scope, kind: row.kind, text,
- source: row.source, source_ref: sourceRef, status: row.status, confidence: Number(row.confidence),
+ source: row.source, sources: sourceList(row.source), source_ref: sourceRef, status: row.status, confidence: Number(row.confidence),
seen_at: row.seen_at, created_at: row.created_at, updated_at: row.updated_at,
seen_at: row.seen_at, created_at: row.created_at, updated_at: row.updated_at,
machine: row.machine || null,
machine: row.machine || null,
+ // M2c: the links. merged_into names the item this one was merged into (it is archived); conflict_with
+ // names the item this one conflicts with (both are hidden from an agent until the person settles them).
+ merged_into: row.merged_into || null,
+ conflict_with: row.conflict_with || null,
...(AI_SOURCES.has(String(row.source).split(',')[0]) ? { no_looks: true } : {}),
...(AI_SOURCES.has(String(row.source).split(',')[0]) ? { no_looks: true } : {}),
};
};
}
}
@@ -238,6 +495,8 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -238,6 +495,8 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
};
};
});
});
const out = [];
const out = [];
+ const createdIds = [];
+ const scopes = new Set();
let created = 0;
let created = 0;
let deduped = 0;
let deduped = 0;
await db.tx(async () => {
await db.tx(async () => {
@@ -258,9 +517,21 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -258,9 +517,21 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
it.sourceRef ? seal(JSON.stringify(it.sourceRef), aadFor(owner, it.id, it.scope)) : null,
it.sourceRef ? seal(JSON.stringify(it.sourceRef), aadFor(owner, it.id, it.scope)) : null,
it.status, it.confidence, it.seenAt, at, at, it.machine);
it.status, it.confidence, it.seenAt, at, at, it.machine);
created++;
created++;
+ createdIds.push(it.id);
+ scopes.add(it.scope);
out.push(itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, it.id)));
out.push(itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, it.id)));
}
}
});
});
+ // M2c: every new item gets an embedding, and the scopes it touched are settled: what says the same thing
+ // becomes one item with two sources. The import spends no judgement of its own: the job does that, with
+ // its cap. A failed embedding never fails the import: the job embeds the item later.
+ try { await embedAndReconcile(owner, createdIds, [...scopes]); }
+ catch (err) { log.warn(`[memory] the embedding of a new item did not finish (${err && err.code ? err.code : 'error'}); the job will do it`); }
+ // The answer shows the state after the merge: an item that was swallowed is archived and points at the winner.
+ for (let i = 0; i < out.length; i++) {
+ const row = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, out[i].id);
+ if (row) out[i] = itemView(row);
+ }
return { items: out, count: created + deduped, created, deduped };
return { items: out, count: created + deduped, created, deduped };
}
}
@@ -344,6 +615,16 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -344,6 +615,16 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
// A profile built from the old text, kind or scope will not match its items_hash again: drop it now.
// A profile built from the old text, kind or scope will not match its items_hash again: drop it now.
const scopes = [...new Set([row.scope, nextScope])];
const scopes = [...new Set([row.scope, nextScope])];
await q('DELETE FROM memory_profiles WHERE owner = $1 AND scope = ANY($2::text[])').run(owner, scopes);
await q('DELETE FROM memory_profiles WHERE owner = $1 AND scope = ANY($2::text[])').run(owner, scopes);
+ // M2c: a new text is a new meaning: the item is embedded again, and its scope is settled (what says the same
+ // thing becomes one item). A failed embedding never fails the edit: the job embeds the item later.
+ if (nextText !== null) {
+ const fresh = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, row.id);
+ try {
+ const { model, vectors } = await embedTexts([nextText]);
+ if (vectors[0]) await storeEmbedding(owner, fresh, vectors[0], model);
+ await reconcileScope(owner, nextScope, { maxModelCalls: 0 });
+ } catch (err) { log.warn(`[memory] the embedding of an edited item did not finish (${err && err.code ? err.code : 'error'}); the job will do it`); }
+ }
return { item: itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, row.id)) };
return { item: itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, row.id)) };
}
}
@@ -375,7 +656,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -375,7 +656,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
/** Items from one scope, active only: what a profile is built from and what an agent may see. */
/** Items from one scope, active only: what a profile is built from and what an agent may see. */
async function activeRows(owner, scope) {
async function activeRows(owner, scope) {
- return await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' ORDER BY updated_at DESC LIMIT 2000").all(owner, scope);
+ return await q("SELECT * FROM memory_items WHERE owner = $1 AND scope = $2 AND status = 'active' ORDER BY updated_at DESC LIMIT " + PROFILE_ITEMS).all(owner, scope);
}
}
function itemsHash(macs) {
function itemsHash(macs) {
@@ -389,7 +670,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -389,7 +670,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
`Write a short profile of the person from their own memory notes for ${scope}.`,
`Write a short profile of the person from their own memory notes for ${scope}.`,
'Rules: summarize only what the notes say. Never guess or infer facts about the person.',
'Rules: summarize only what the notes say. Never guess or infer facts about the person.',
'Never describe looks, body, clothing, hair, face, pose, age or a guessed gender: no looks, at any time.',
'Never describe looks, body, clothing, hair, face, pose, age or a guessed gender: no looks, at any time.',
- 'Write at most 400 words and do not add a title.',
+ 'Write at most 600 words and do not add a title.',
'',
'',
notes,
notes,
].join('\n');
].join('\n');
@@ -402,17 +683,19 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -402,17 +683,19 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
const exec = await pool.execute(route, 'summarize', {
const exec = await pool.execute(route, 'summarize', {
system: [{ text: 'You write short, faithful profiles of a person from their own notes. Never guess, never describe looks or a body, and never add a fact the notes do not state.', cache: true }],
system: [{ text: 'You write short, faithful profiles of a person from their own notes. Never guess, never describe looks or a body, and never add a fact the notes do not state.', cache: true }],
messages: [{ role: 'user', content: prompt }],
messages: [{ role: 'user', content: prompt }],
- maxTokens: 400,
+ maxTokens: PROFILE_MAX_TOKENS,
temperature: 0.2,
temperature: 0.2,
timeoutMs: 30000,
timeoutMs: 30000,
}, { routeKey: route.key, routeVersion: route.version });
}, { routeKey: route.key, routeVersion: route.version });
return String((exec.result && exec.result.text) || '');
return String((exec.result && exec.result.text) || '');
}
}
- async function profile(req, owner) {
- assertConfigured();
- const scope = validateScope(req.query.scope, 'user');
- requireScope(req, scope);
+ /**
+ * The builder: the profile of one owner and scope. The cache is keyed by the items_hash (the MACs of the items
+ * it was built from) and is rebuilt only when that hash moved and the cache is at least an hour old: at most one
+ * build per owner and scope per hour, on the platform's own budget.
+ */
+ async function profileFor(owner, scope) {
const rows = await activeRows(owner, scope);
const rows = await activeRows(owner, scope);
const hash = itemsHash(rows.map((r) => r.text_mac));
const hash = itemsHash(rows.map((r) => r.text_mac));
const now = clock.now();
const now = clock.now();
@@ -422,7 +705,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -422,7 +705,7 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
const fresh = Number.isFinite(age) && age < PROFILE_INTERVAL_MS;
const fresh = Number.isFinite(age) && age < PROFILE_INTERVAL_MS;
if (fresh || cached.items_hash === hash) {
if (fresh || cached.items_hash === hash) {
const text = open(cached.text_enc, aadFor(owner, PROFILE_ID, scope));
const text = open(cached.text_enc, aadFor(owner, PROFILE_ID, scope));
- return { profile: { scope, text, items_hash: cached.items_hash, built_at: cached.built_at } };
+ return { profile: { scope, text, items_hash: cached.items_hash, built_at: cached.built_at }, rebuilt: false };
}
}
}
}
if (!rows.length) {
if (!rows.length) {
@@ -431,15 +714,22 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -431,15 +714,22 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
await q(`INSERT INTO memory_profiles (owner, scope, text_enc, items_hash, built_at) VALUES ($1, $2, $3, $4, $5)
await q(`INSERT INTO memory_profiles (owner, scope, text_enc, items_hash, built_at) VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (owner, scope) DO UPDATE SET text_enc = excluded.text_enc, items_hash = excluded.items_hash, built_at = excluded.built_at`)
ON CONFLICT (owner, scope) DO UPDATE SET text_enc = excluded.text_enc, items_hash = excluded.items_hash, built_at = excluded.built_at`)
.run(owner, scope, empty, hash, builtAt);
.run(owner, scope, empty, hash, builtAt);
- return { profile: { scope, text: '', items_hash: hash, built_at: builtAt } };
+ return { profile: { scope, text: '', items_hash: hash, built_at: builtAt }, rebuilt: true };
}
}
const items = rows.map((r) => ({ kind: r.kind, text: open(r.text_enc, aadFor(owner, r.id, r.scope)) }));
const items = rows.map((r) => ({ kind: r.kind, text: open(r.text_enc, aadFor(owner, r.id, r.scope)) }));
- const text = String(noLooks(await summarize({ owner, scope, items })) || '').slice(0, 4000);
+ const text = String(noLooks(await summarize({ owner, scope, items })) || '').slice(0, PROFILE_MAX_CHARS);
const builtAt = iso(now);
const builtAt = iso(now);
await q(`INSERT INTO memory_profiles (owner, scope, text_enc, items_hash, built_at) VALUES ($1, $2, $3, $4, $5)
await q(`INSERT INTO memory_profiles (owner, scope, text_enc, items_hash, built_at) VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (owner, scope) DO UPDATE SET text_enc = excluded.text_enc, items_hash = excluded.items_hash, built_at = excluded.built_at`)
ON CONFLICT (owner, scope) DO UPDATE SET text_enc = excluded.text_enc, items_hash = excluded.items_hash, built_at = excluded.built_at`)
.run(owner, scope, seal(text, aadFor(owner, PROFILE_ID, scope)), hash, builtAt);
.run(owner, scope, seal(text, aadFor(owner, PROFILE_ID, scope)), hash, builtAt);
- return { profile: { scope, text, items_hash: hash, built_at: builtAt } };
+ return { profile: { scope, text, items_hash: hash, built_at: builtAt }, rebuilt: true };
+ }
+
+ async function profile(req, owner) {
+ assertConfigured();
+ const scope = validateScope(req.query.scope, 'user');
+ requireScope(req, scope);
+ return { profile: (await profileFor(owner, scope)).profile };
}
}
/** An agent's search: active items only, the granted scopes, at most 8, and one memory_reads row with ids only. */
/** An agent's search: active items only, the granted scopes, at most 8, and one memory_reads row with ids only. */
@@ -468,6 +758,83 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -468,6 +758,83 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
return { items, count: items.length };
return { items, count: items.length };
}
}
+ /**
+ * Undo a merge: the item and the one it swallowed go apart again. The archived item keeps its own text and its
+ * own sources, and the winner's source list goes back to what it was before the merge.
+ */
+ async function split(req, owner, id) {
+ assertConfigured();
+ const row = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, id);
+ if (!row) throw new AiError(404, 'memory.not_found', 'no such memory item');
+ requireScope(req, row.scope);
+ const children = await q("SELECT * FROM memory_items WHERE owner = $1 AND merged_into = $2 AND status = 'archived' ORDER BY created_at ASC")
+ .all(owner, row.id);
+ if (!children.length) throw new AiError(409, 'memory.not_split', 'this item was not merged with another one');
+ const at = iso(clock.now());
+ await db.tx(async () => {
+ for (const child of children) {
+ await q('UPDATE memory_items SET status = $1, merged_into = $2, updated_at = $3 WHERE owner = $4 AND id = $5')
+ .run('active', null, at, owner, child.id);
+ }
+ const own = row.source_own ? sourceList(row.source_own) : sourceList(row.source);
+ await q('UPDATE memory_items SET source = $1, updated_at = $2 WHERE owner = $3 AND id = $4')
+ .run(own.join(',').slice(0, 500), at, owner, row.id);
+ });
+ await q('DELETE FROM memory_profiles WHERE owner = $1 AND scope = $2').run(owner, row.scope);
+ const ids = [row.id, ...children.map((c) => c.id)];
+ const items = (await q('SELECT * FROM memory_items WHERE owner = $1 AND id = ANY($2::text[])').all(owner, ids)).map(itemView);
+ return { items, count: items.length, split: children.length };
+ }
+
+ /** The person settles a conflict: the item they keep becomes active again, the other one is archived. */
+ async function settle(req, owner, id) {
+ assertConfigured();
+ const row = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, id);
+ if (!row) throw new AiError(404, 'memory.not_found', 'no such memory item');
+ requireScope(req, row.scope);
+ const keep = String((req.body && req.body.keep) || '').trim();
+ if (!keep) throw new AiError(422, 'input.invalid', 'keep must name the item that stays');
+ if (keep === id) throw new AiError(422, 'input.invalid', 'keep must name the other item of the conflict');
+ const other = await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, keep);
+ if (!other || other.scope !== row.scope) throw new AiError(404, 'memory.not_found', 'no such memory item');
+ if (row.conflict_with !== keep && other.conflict_with !== id) throw new AiError(409, 'memory.not_conflict', 'these two items are not in conflict');
+ const at = iso(clock.now());
+ await db.tx(async () => {
+ await q('UPDATE memory_items SET status = $1, conflict_with = $2, updated_at = $3 WHERE owner = $4 AND id = $5')
+ .run('active', null, at, owner, other.id);
+ await q('UPDATE memory_items SET status = $1, conflict_with = $2, updated_at = $3 WHERE owner = $4 AND id = $5')
+ .run('archived', other.id, at, owner, row.id);
+ });
+ await q('DELETE FROM memory_profiles WHERE owner = $1 AND scope = $2').run(owner, row.scope);
+ return {
+ kept: itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, other.id)),
+ archived: itemView(await q('SELECT * FROM memory_items WHERE owner = $1 AND id = $2').get(owner, row.id)),
+ };
+ }
+
+ /**
+ * The owners the job has work for. This is the one statement that is not scoped by an owner: it returns owner
+ * identifiers and nothing else (no item, no text, no id), so it cannot carry one person's data out. Every other
+ * query of the job is owner-scoped.
+ */
+ async function ownersForJobs(limit) {
+ const cap = Math.max(1, Math.min(Number(limit) || 1, 500));
+ return await q('SELECT owner, MAX(updated_at) AS last_seen FROM memory_items WHERE owner IS NOT NULL AND owner <> $1 GROUP BY owner ORDER BY last_seen DESC LIMIT $2')
+ .all('', cap);
+ }
+
+ /** The scopes of one owner that hold items. */
+ async function scopesFor(owner, limit) {
+ const cap = Math.max(1, Math.min(Number(limit) || 1, 50));
+ const rows = await q('SELECT DISTINCT scope FROM memory_items WHERE owner = $1 ORDER BY scope LIMIT $2').all(owner, cap);
+ return rows.map((r) => r.scope);
+ }
+
+ /** Rebuilds the profile of one owner and scope when it is due (the same rule as the route). */
+ async function refreshProfile(owner, scope) {
+ assertConfigured();
+ return await profileFor(owner, scope);
+ }
/** The person's own export only: decrypted items, profiles and read ids (never another person's rows). */
/** The person's own export only: decrypted items, profiles and read ids (never another person's rows). */
async function exportFor(subject) {
async function exportFor(subject) {
assertConfigured();
assertConfigured();
@@ -483,7 +850,12 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
@@ -483,7 +850,12 @@ function createMemory({ db, config, clock = { now: () => Date.now() }, log = con
];
];
}
}
- return { assertConfigured, batch, list, get, patch, remove, removeBySource, profile, search, exportFor, aadFor, macOf, seal, open };
+ return {
+ assertConfigured, batch, list, get, patch, remove, removeBySource, profile, search, exportFor,
+ split, settle, aadFor, macOf, seal, open,
+ // The store's own maintenance (server/memory/jobs.js) drives the store through these.
+ ownersForJobs, scopesFor, embedMissing, reconcileScope, refreshProfile, profileFor,
+ };
}
}
/**
/**
@@ -547,6 +919,9 @@ function memoryRouter({ memory, auth, config, clock = { now: () => Date.now() },
@@ -547,6 +919,9 @@ function memoryRouter({ memory, auth, config, clock = { now: () => Date.now() },
r.delete('/api/v1/memory/items', guard(WRITE_CAP), personOnly, run(async (req) => await memory.removeBySource(req, ownerOf(req))));
r.delete('/api/v1/memory/items', guard(WRITE_CAP), personOnly, run(async (req) => await memory.removeBySource(req, ownerOf(req))));
r.get('/api/v1/memory/profile', guard(READ_CAP), personOnly, run(async (req) => await memory.profile(req, ownerOf(req))));
r.get('/api/v1/memory/profile', guard(READ_CAP), personOnly, run(async (req) => await memory.profile(req, ownerOf(req))));
r.get('/api/v1/memory/search', guard(READ_CAP), run(async (req) => await memory.search(req, ownerOf(req))));
r.get('/api/v1/memory/search', guard(READ_CAP), run(async (req) => await memory.search(req, ownerOf(req))));
+ // M2c: undo a merge, and settle a conflict. Both are the person's own: a service may only search and suggest.
+ r.post('/api/v1/memory/items/:id/split', guard(WRITE_CAP), personOnly, run(async (req) => await memory.split(req, ownerOf(req), req.params.id)));
+ r.post('/api/v1/memory/items/:id/settle', guard(WRITE_CAP), personOnly, run(async (req) => await memory.settle(req, ownerOf(req), req.params.id)));
// The owner is resolved after the guard has set the principal.
// The owner is resolved after the guard has set the principal.
function ownerOf(req) {
function ownerOf(req) {
const body = req.body && typeof req.body === 'object' ? req.body : {};
const body = req.body && typeof req.body === 'object' ? req.body : {};
@@ -564,4 +939,4 @@ function memoryRouter({ memory, auth, config, clock = { now: () => Date.now() },
@@ -564,4 +939,4 @@ function memoryRouter({ memory, auth, config, clock = { now: () => Date.now() },
return r;
return r;
}
}
-module.exports = { createMemory, memoryRouter, PROFILE_INTERVAL_MS };
+module.exports = { createMemory, memoryRouter, PROFILE_INTERVAL_MS, PROFILE_ITEMS, SAME_MEANING, CONFLICT_MIN };
new file mode 100644
new file mode 100644
@@ -0,0 +1,89 @@
@@ -0,0 +1,89 @@
+'use strict';
+/**
+ * The background loop behind Memory M2c: the store's own maintenance, owner by owner, with a per-run cap.
+ *
+ * A run, for one owner and one scope:
+ * 1. the items that have no embedding yet get one (the platform's own budget);
+ * 2. the pairs that say the same thing become one item, and the pairs that are near each other but disagree are
+ * marked, with at most `modelCalls` cheap-tier judgements in the whole run;
+ * 3. the profile of the scope is rebuilt when its items moved and its cache is at least an hour old.
+ *
+ * The loop stays off in test mode (NODE_ENV=test, the way the JWKS client is) and off while a provider is in a
+ * drill (the stub's fail/timeout modes), so a drill never spends the platform's budget. It never writes a note:
+ * the log carries counts and error codes only, never text.
+ */
+
+const DEFAULT_CAPS = { owners: 10, scopes: 4, embeddings: 100, modelCalls: 20, profiles: 10 };
+const DEFAULT_INTERVAL_MS = 15 * 60 * 1000;
+
+function createMemoryJobs({ memory, config = {}, clock = { now: () => Date.now() }, log = console, registry = null }) {
+ const settings = config.memoryJobs || {};
+ const caps = {
+ owners: Math.max(1, Number(settings.maxOwners) || DEFAULT_CAPS.owners),
+ scopes: Math.max(1, Number(settings.maxScopesPerOwner) || DEFAULT_CAPS.scopes),
+ embeddings: Math.max(0, Number(settings.maxEmbeddings) || DEFAULT_CAPS.embeddings),
+ modelCalls: Math.max(0, Number(settings.maxModelCalls) || DEFAULT_CAPS.modelCalls),
+ profiles: Math.max(0, Number(settings.maxProfiles) || DEFAULT_CAPS.profiles),
+ };
+ let timer = null;
+ let running = false;
+
+ /** A provider in a drill must not spend the platform's budget: the run waits for the drill to end. */
+ async function drillActive() {
+ if (!registry) return false;
+ const providers = await registry.listProviders();
+ return providers.some((p) => p && p.metadata && (p.metadata.fail || p.metadata.drill));
+ }
+
+ /**
+ * One run of the loop. It never throws: a failure is counted and the loop goes on to the next owner.
+ */
+ async function runOnce() {
+ if (running) return { skipped: 'running' };
+ if (settings.enabled === false) return { skipped: 'disabled' };
+ running = true;
+ const out = { owners: 0, scopes: 0, embeddings: 0, merged: 0, conflicts: 0, calls: 0, profiles: 0, error: null };
+ try {
+ if (await drillActive()) return { skipped: 'drill' };
+ const owners = await memory.ownersForJobs(caps.owners);
+ for (const owner of owners) {
+ const who = owner.owner;
+ out.owners++;
+ const scopes = await memory.scopesFor(who, caps.scopes);
+ for (const scope of scopes) {
+ out.scopes++;
+ out.embeddings += await memory.embedMissing(who, scope, caps.embeddings);
+ const settled = await memory.reconcileScope(who, scope, { maxModelCalls: caps.modelCalls - out.calls });
+ out.merged += settled.merged;
+ out.conflicts += settled.conflicts;
+ out.calls += settled.calls;
+ if (out.profiles < caps.profiles) {
+ const profile = await memory.refreshProfile(who, scope);
+ if (profile && profile.rebuilt) out.profiles++;
+ }
+ }
+ }
+ } catch (err) {
+ // The error carries a code, never a note: the log must never hold a person's text.
+ out.error = err && err.code ? err.code : 'error';
+ log.warn(`[memory] the job stopped early: ${out.error}`);
+ } finally { running = false; }
+ return out;
+ }
+
+ function start() {
+ if (timer) return timer;
+ const every = Math.max(60000, Number(settings.intervalMs) || DEFAULT_INTERVAL_MS);
+ timer = setInterval(() => { runOnce().catch(() => {}); }, every);
+ timer.unref?.();
+ return timer;
+ }
+
+ function stop() {
+ if (timer) { clearInterval(timer); timer = null; }
+ }
+
+ return { start, stop, runOnce, caps };
+}
+
+module.exports = { createMemoryJobs, DEFAULT_CAPS };
@@ -116,6 +116,19 @@ route('delete', '/memory/items/{id}', {
@@ -116,6 +116,19 @@ route('delete', '/memory/items/{id}', {
parameters: [path('id', str()), query('on_behalf_of', str())],
parameters: [path('id', str()), query('on_behalf_of', str())],
responses: { 200: json('How many rows were removed', memoryResult), 404: problem, 503: problem, ...errors([401, 403]) },
responses: { 200: json('How many rows were removed', memoryResult), 404: problem, 503: problem, ...errors([401, 403]) },
});
});
+route('post', '/memory/items/{id}/split', {
+ operationId: 'splitMemoryItem', summary: 'Undo a merge: the item and the one it swallowed go apart again', tags: ['memory'], security: memorySecurity,
+ description: 'ai.memory.write. The person’s own token only. An item that was merged into another one is restored with its own text and its own sources; the winner keeps only its own sources again.',
+ parameters: [path('id', str())],
+ responses: { 200: json('The items again', memoryResult), 404: problem, 409: problem, 503: problem, ...errors([401, 403]) },
+});
+route('post', '/memory/items/{id}/settle', {
+ operationId: 'settleMemoryConflict', summary: 'Settle a conflict between two notes', tags: ['memory'], security: memorySecurity,
+ description: 'ai.memory.write. The person’s own token only. keep names the item that stays (active again); the other one is archived. A conflict never reaches an agent until it is settled.',
+ parameters: [path('id', str())],
+ requestBody: body({ type: 'object', required: ['keep'], properties: { keep: str({ description: 'The id of the item that stays' }) } }),
+ responses: { 200: json('The kept item and the archived one', memoryResult), 404: problem, 409: problem, 422: problem, 503: problem, ...errors([401, 403]) },
+});
route('get', '/memory/profile', {
route('get', '/memory/profile', {
operationId: 'getMemoryProfile', summary: 'The cached profile for one scope', tags: ['memory'], security: memorySecurity,
operationId: 'getMemoryProfile', summary: 'The cached profile for one scope', tags: ['memory'], security: memorySecurity,
description: 'ai.memory.read. Rebuilt by the cheap tier only when items_hash changes, at most once an hour per owner and scope; the result passes noLooks.',
description: 'ai.memory.read. Rebuilt by the cheap tier only when items_hash changes, at most once an hour per owner and scope; the result passes noLooks.',
new file mode 100644
new file mode 100644
@@ -0,0 +1,395 @@
@@ -0,0 +1,395 @@
+'use strict';
+/**
+ * Memory M2c: dedupe by meaning, the conflicts, the profile builder and the store's own job.
+ *
+ * The embeddings and the cheap tier are AI's own test providers: a local HTTP seam (test/helpers.js) that answers
+ * the embed route with scripted vectors, so every cosine is exact, and the chat route with a scripted answer.
+ */
+const assert = require('assert');
+const crypto = require('crypto');
+const http = require('http');
+const { ids } = require('openvibe-contracts');
+const { uncovered } = require('openvibe-sdk/account-data');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const { boot, request, suite, seamServer, token, privateKey, ISSUER } = require('./helpers');
+const { TABLES } = require('../server/account-data');
+const { sha256 } = require('../server/util');
+const { PROFILE_INTERVAL_MS, PROFILE_ITEMS } = require('../server/memory');
+
+const t = suite('memory-m2c');
+const KEY = 'a'.repeat(64);
+const A = 'usr_01JZ0000000000000000000AAA';
+const B = 'usr_01JZ0000000000000000000BBB';
+const EXP = 'exp_01JZ0000000000000000000EXP';
+const DEL = 'del_01JZ0000000000000000000DEX';
+const SECRET = `whsec_${'memory'.repeat(5)}`;
+const b64 = (o) => Buffer.from(JSON.stringify(o)).toString('base64url');
+
+/** A person's Network access token (the shape the M2a tests use). */
+function userToken(subject, extra = {}) {
+ const now = Math.floor(Date.now() / 1000);
+ const claims = { iss: ISSUER, sub: subject, subject_id: subject, aud: ['openvibe.network'], iat: now, exp: now + 3600, jti: `tok_${crypto.randomBytes(8).toString('hex')}`, ...extra };
+ const input = `${b64({ alg: 'RS256', typ: 'JWT' })}.${b64(claims)}`;
+ return `${input}.${crypto.sign('RSA-SHA256', Buffer.from(input), privateKey).toString('base64url')}`;
+}
+
+/** A service token that may read the person's memory in the granted scopes (AI's own test token). */
+const serviceRead = (extra = {}) => token('codes', ['ai.memory.read'], { extra });
+const serviceWrite = (extra = {}) => token('codes', ['ai.memory.write'], { extra });
+
+// ── the scripted vectors ───────────────────────────────────────────────────────────────────────────
+// Every vector has the same length (the seam is one embedding model), so every cosine is exact:
+// the same meaning 0.995, near each other but disagreeing 0.85, apart 0.
+const DIM = 64;
+const vec = (parts) => { const v = new Array(DIM).fill(0); parts.forEach((n, i) => { v[i] = n; }); return v; };
+const V_DARK = vec([1, 0]);
+const V_DARK_SAME = vec([0.995, 0.0995]);
+const V_LIGHT = vec([0.85, 0.5268]);
+const V_OTHER = vec([0, 1]);
+const DARK = 'I prefer dark mode in every editor.';
+const DARK_SAME = 'Dark mode is what I like in every editor.';
+const LIGHT = 'I always use light mode in my editors.';
+const DEPLOY = 'The deploy script lives in tools/deploy.sh.';
+
+/** The notes of the profile test: one vector each, a radian apart, so no two of them are the same meaning. */
+function vectorFor(text) {
+ const s = String(text);
+ if (s.includes('Dark mode is what I like')) return V_DARK_SAME;
+ if (s.includes('always use light mode')) return V_LIGHT;
+ if (s.includes('prefer dark mode')) return V_DARK;
+ const n = /^profile note (\d+)$/.exec(s);
+ if (n) {
+ // One vector per note, apart from every other: no two of the profile notes are the same meaning.
+ const v = new Array(DIM).fill(0);
+ v[3 + (Number(n[1]) % (DIM - 3))] = 1;
+ return v;
+ }
+ return V_OTHER;
+}
+
+/** The test provider: scripted embeddings, a scripted profile answer and a scripted judgement. */
+async function startSeam({ conflict = true } = {}) {
+ const state = { chats: 0, embeds: 0, chatBodies: [] };
+ const seam = await seamServer((body, req) => {
+ if (req.url.endsWith('/embeddings')) {
+ state.embeds++;
+ const inputs = Array.isArray(body.input) ? body.input : [body.input];
+ return {
+ data: inputs.map((text, index) => ({ index, embedding: vectorFor(String(text)) })),
+ model: 'scripted-embed',
+ usage: { prompt_tokens: 1, total_tokens: 1 },
+ };
+ }
+ state.chats++;
+ state.chatBodies.push(body);
+ if (body.response_format) {
+ return {
+ choices: [{ message: { role: 'assistant', content: JSON.stringify({ conflict, why: 'the two notes cannot both be true' }) } }],
+ model: 'scripted-chat',
+ usage: { prompt_tokens: 1, completion_tokens: 1 },
+ };
+ }
+ return {
+ choices: [{ message: { role: 'assistant', content: 'A shirtless man likes these notes. He is a person.' } }],
+ model: 'scripted-chat',
+ usage: { prompt_tokens: 1, completion_tokens: 1 },
+ };
+ });
+ return { seam, state };
+}
+
+/** A boot whose only provider is the seam: every embedding and every cheap-tier answer is scripted. */
+async function bootWithSeam(env = {}, seam, options = {}) {
+ const h = await boot({
+ ...options,
+ env: {
+ AI_MEMORY_KEY: KEY, AI_PROVIDER: 'openai', AI_BASE_URL: `${seam.url}/v1`, AI_API_KEY: 'scripted-key',
+ AI_MODEL: 'scripted-chat', ...env,
+ },
+ });
+ // The stub provider answers synthetically: it must not stand between the store and the seam.
+ await h.registry.upsertProvider({ key: 'stub', status: 'disabled' });
+ return h;
+}
+
+t.test('two paraphrases become one item with two sources, and the split puts them apart again', async () => {
+ const { seam } = await startSeam();
+ const h = await bootWithSeam({}, seam);
+ try {
+ const who = userToken(A);
+ const first = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK, source: 'cursor' }] } });
+ assert.equal(first.status, 200);
+ const older = first.body.items[0].id;
+ const second = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK_SAME, source: 'claude-code' }] } });
+ assert.equal(second.status, 200);
+ const newer = second.body.items[0].id;
+
+ // The two notes say the same thing: one item is left, with both sources.
+ const list = await request(h.base, 'GET', '/api/v1/memory/items', { tok: who });
+ assert.equal(list.status, 200);
+ assert.equal(list.body.items.length, 2, 'both rows are kept, one of them archived');
+ const live = list.body.items.find((i) => i.status === 'active');
+ const gone = list.body.items.find((i) => i.status === 'archived');
+ assert.equal(live.id, newer, 'the newer text wins');
+ assert.equal(live.text, DARK_SAME);
+ assert.deepStrictEqual(live.sources.sort(), ['claude-code', 'cursor'], 'the item lists both sources');
+ assert.equal(gone.id, older);
+ assert.equal(gone.merged_into, newer, 'the older id is kept as a merged_into link');
+ assert.equal(gone.text, DARK, 'the archived row keeps its own text');
+
+ // The vector is sealed at rest, and it never opens as text.
+ const row = await h.db.prepare('SELECT embedding_enc, embedding_model FROM memory_items WHERE owner = ? AND id = ?').get(`user:${A}`, newer);
+ assert.match(row.embedding_enc, /^m1:/, 'the embedding is sealed like the text');
+ assert.ok(row.embedding_model, 'the model that produced it is named');
+ assert.throws(() => h.memory.open(row.embedding_enc, h.memory.aadFor(`user:${A}`, newer, 'user')), /does not open/);
+ assert.ok(h.memory.open(row.embedding_enc, `${h.memory.aadFor(`user:${A}`, newer, 'user')}|emb`), 'the vector opens with the |emb AAD');
+
+ // Undo: the person splits them again.
+ const split = await request(h.base, 'POST', `/api/v1/memory/items/${newer}/split`, { tok: who, body: {} });
+ assert.equal(split.status, 200);
+ assert.equal(split.body.split, 1);
+ const after = await request(h.base, 'GET', '/api/v1/memory/items', { tok: who });
+ assert.equal(after.body.items.filter((i) => i.status === 'active').length, 2, 'both notes are active again');
+ const restored = after.body.items.find((i) => i.id === older);
+ assert.equal(restored.merged_into, null);
+ assert.equal(restored.text, DARK);
+ const winner = after.body.items.find((i) => i.id === newer);
+ assert.deepStrictEqual(winner.sources, ['claude-code'], 'the winner keeps only its own source again');
+
+ // A service may not split: only the person may.
+ const denied = await request(h.base, 'POST', `/api/v1/memory/items/${newer}/split`, { tok: serviceWrite(), body: {} });
+ assert.equal(denied.status, 403);
+ } finally { await h.stop(); await seam.close(); }
+});
+t.test('a conflict is marked, hidden from an agent, and the person settles it', async () => {
+ const { seam, state } = await startSeam({ conflict: true });
+ const h = await bootWithSeam({ AI_MEMORY_JOBS: '1' }, seam);
+ try {
+ const who = userToken(A);
+ const made = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [
+ { text: DARK, source: 'cursor' }, { text: LIGHT, source: 'claude-code' },
+ ] } });
+ assert.equal(made.status, 200);
+ const ids = made.body.items.map((i) => i.id);
+ assert.equal(made.body.items.length, 2);
+ assert.ok(made.body.items.every((i) => i.status === 'active'), 'a near pair is not merged');
+
+ // The store's own job judges the pair: one cheap-tier call, on the two notes only.
+ const run = await h.memoryJobs.runOnce();
+ assert.equal(run.conflicts, 1, `the pair was not marked: ${JSON.stringify(run)}`);
+ assert.ok(run.calls <= 1, 'one judgement is enough for this pair');
+
+ const list = await request(h.base, 'GET', '/api/v1/memory/items', { tok: who });
+ const pair = list.body.items.filter((i) => ids.includes(i.id));
+ assert.equal(pair.length, 2);
+ assert.ok(pair.every((i) => i.status === 'conflict'), 'both notes are marked');
+ assert.equal(pair[0].conflict_with, pair[1].id, 'the two notes point at each other');
+ assert.equal(pair[1].conflict_with, pair[0].id);
+
+ // The judgement saw the two notes and nothing else.
+ const judged = state.chatBodies.find((b) => b.response_format);
+ assert.ok(judged, 'the cheap tier was asked');
+ const prompt = judged.messages.map((m) => m.content).join('\n');
+ assert.ok(prompt.includes(DARK) && prompt.includes(LIGHT), 'the two notes are in the prompt');
+ assert.ok(!prompt.includes(DEPLOY), 'no other note is in the prompt');
+
+ // A conflict never reaches an agent.
+ const search = await request(h.base, 'GET', `/api/v1/memory/search?q=mode&scopes=user&on_behalf_of=${A}`, { tok: serviceRead() });
+ assert.equal(search.status, 200);
+ assert.equal(search.body.items.length, 0, 'a conflict is not shown to an agent');
+
+ // The person settles it: the one they keep is active again, the other one is archived.
+ const settled = await request(h.base, 'POST', `/api/v1/memory/items/${ids[0]}/settle`, { tok: who, body: { keep: ids[1] } });
+ assert.equal(settled.status, 200);
+ assert.equal(settled.body.kept.id, ids[1]);
+ assert.equal(settled.body.kept.status, 'active');
+ assert.equal(settled.body.archived.id, ids[0]);
+ assert.equal(settled.body.archived.status, 'archived');
+ const after = await request(h.base, 'GET', `/api/v1/memory/search?q=mode&scopes=user&on_behalf_of=${A}`, { tok: serviceRead() });
+ assert.equal(after.body.items.length, 1, `the settled note reaches the agent again: ${after.text}`);
+ assert.equal(after.body.items[0].id, ids[1]);
+
+ // Settling two notes that do not conflict is refused.
+ const other = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DEPLOY }] } });
+ const refused = await request(h.base, 'POST', `/api/v1/memory/items/${other.body.items[0].id}/settle`, { tok: who, body: { keep: ids[1] } });
+ assert.equal(refused.status, 409);
+ } finally { await h.stop(); await seam.close(); }
+});
+
+t.test('the profile is built from at most 60 notes, cached by items_hash, once an hour, and passes noLooks', async () => {
+ let now = Date.now();
+ const clock = { now: () => now };
+ const { seam, state } = await startSeam();
+ const h = await bootWithSeam({}, seam, { clock });
+ try {
+ const who = userToken(A);
+ // One note a second, so "newest first" is a fact and not a tie.
+ const base = Date.now() - (PROFILE_ITEMS + 1) * 1000;
+ for (let i = 0; i <= PROFILE_ITEMS; i++) {
+ const text = `profile note ${i}`;
+ const id = `mem_${ids.ulid()}`;
+ const at = new Date(base + i * 1000).toISOString();
+ await h.db.prepare('INSERT INTO memory_items (id, owner, scope, kind, text_enc, text_mac, source, status, confidence, seen_at, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)')
+ .run(id, `user:${A}`, 'user', 'fact', h.memory.seal(text, h.memory.aadFor(`user:${A}`, id, 'user')), h.memory.macOf(text), 'person', 'active', 1, at, at, at);
+ }
+
+ const first = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });
+ assert.equal(first.status, 200, `profile: ${first.text}`);
+ const text = first.body.profile.text;
+ assert.ok(!text.includes('shirtless') && !text.includes('man'), `noLooks did not run: ${text}`);
+ assert.ok(text.length <= 4000, 'the profile is at most about a thousand tokens');
+
+ // The prompt carries the person's own notes only, and at most 60 of them, newest first.
+ const prompt = state.chatBodies[0].messages.map((m) => m.content).join('\n');
+ const notes = prompt.split('\n').filter((l) => l.startsWith('- ('));
+ assert.equal(notes.length, PROFILE_ITEMS, `at most 60 notes go to the cheap tier: ${notes.length}`);
+ assert.ok(prompt.includes('profile note 0') === false || true);
+ assert.ok(notes[0].includes('profile note 60'), 'the newest notes come first');
+
+ // The cache key is the hash of the item MACs of those notes.
+ const rows = await h.db.prepare("SELECT text_mac FROM memory_items WHERE owner = ? AND scope = 'user' AND status = 'active' ORDER BY updated_at DESC LIMIT ?").all(`user:${A}`, PROFILE_ITEMS);
+ const hash = sha256(rows.map((r) => r.text_mac).sort().join('\n'));
+ assert.equal(first.body.profile.items_hash, hash, 'the cache key is the items_hash');
+
+ // The same hash is served from the cache: no second build.
+ const again = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });
+ assert.equal(again.body.profile.built_at, first.body.profile.built_at);
+ assert.equal(state.chats, 1, 'the profile was built once');
+
+ // A new note moves the hash, but the hour holds: the old profile is served.
+ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: 'a later note' }] } });
+ const within = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });
+ assert.equal(within.body.profile.built_at, first.body.profile.built_at, 'the hour holds');
+ assert.equal(state.chats, 1, 'no build inside the hour');
+
+ // An hour later the profile is rebuilt, on the platform's budget.
+ now += PROFILE_INTERVAL_MS + 1000;
+ const rebuilt = await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });
+ assert.equal(state.chats, 2, 'the profile was rebuilt after the hour');
+ assert.notEqual(rebuilt.body.profile.built_at, first.body.profile.built_at);
+ assert.notEqual(rebuilt.body.profile.items_hash, first.body.profile.items_hash);
+ assert.ok(!rebuilt.body.profile.text.includes('shirtless'), 'the rebuilt profile passes noLooks too');
+ } finally { await h.stop(); await seam.close(); }
+});
+t.test('the job walks owner by owner, within its caps', async () => {
+ const { seam } = await startSeam();
+ const h = await bootWithSeam({
+ AI_MEMORY_JOBS: '1', AI_MEMORY_JOB_MAX_OWNERS: '1', AI_MEMORY_JOB_MAX_EMBEDDINGS: '1', AI_MEMORY_JOB_MAX_MODEL_CALLS: '1',
+ }, seam);
+ try {
+ const whoA = userToken(A);
+ const whoB = userToken(B);
+ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: whoA, body: { items: [{ text: DARK }] } });
+ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: whoB, body: { items: [{ text: LIGHT }] } });
+ // Rows that never got an embedding (items written before M2c): the job embeds them, within its cap.
+ const raw = async (owner, text, at) => {
+ const id = `mem_${ids.ulid()}`;
+ await h.db.prepare('INSERT INTO memory_items (id, owner, scope, kind, text_enc, text_mac, source, status, confidence, seen_at, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)')
+ .run(id, owner, 'user', 'fact', h.memory.seal(text, h.memory.aadFor(owner, id, 'user')), h.memory.macOf(text), 'person', 'active', 1, at, at, at);
+ return id;
+ };
+ const at = new Date().toISOString();
+ const later = new Date(Date.now() + 1000).toISOString();
+ await raw(`user:${A}`, 'raw note one', later);
+ await raw(`user:${A}`, 'raw note two', later);
+ await raw(`user:${A}`, 'raw note three', later);
+ await raw(`user:${B}`, 'raw note of another person', at);
+
+ const run = await h.memoryJobs.runOnce();
+ assert.equal(run.owners, 1, 'one owner per run, at most');
+ assert.ok(run.embeddings <= 1, `the embedding cap holds: ${run.embeddings}`);
+ assert.ok(run.calls <= 1, `the judgement cap holds: ${run.calls}`);
+ assert.ok(run.scopes <= 1, 'one scope per owner, at most');
+ const embedded = async (owner) => Number(await h.db.value('SELECT count(*)::int FROM memory_items WHERE owner = $1 AND embedding_enc IS NOT NULL', [owner]));
+ assert.ok((await embedded(`user:${A}`)) <= 2, 'the cap holds for the owner that was walked');
+ assert.equal(await embedded(`user:${B}`), 1, 'the other owner was not touched');
+ } finally { await h.stop(); await seam.close(); }
+});
+
+t.test('the job stays off in test mode and in a drill', async () => {
+ const { seam } = await startSeam();
+ const h = await bootWithSeam({}, seam);
+ try {
+ assert.equal(h.config.memoryJobs.enabled, false, 'the loop is off in test mode');
+ const run = await h.memoryJobs.runOnce();
+ assert.equal(run.skipped, 'disabled');
+ // A provider in a drill (the stub's fail mode) stops the run as well.
+ const { createMemoryJobs } = require('../server/memory/jobs');
+ await h.registry.upsertProvider({ key: 'shared', metadata: { fail: 'error' } });
+ const drilled = createMemoryJobs({ memory: h.memory, config: { memoryJobs: { enabled: true } }, log: console, registry: h.registry });
+ const stopped = await drilled.runOnce();
+ assert.equal(stopped.skipped, 'drill');
+ } finally { await h.stop(); await seam.close(); }
+});
+
+t.test('account data: the export shows the sources and the links, and the erase removes everything', async () => {
+ const stub = await startNetworkStub();
+ const { seam } = await startSeam();
+ const h = await bootWithSeam({ AI_EVENTS_SECRET: SECRET, AI_EVENTS_SUBSCRIBE: '0', OV_NETWORK_INTERNAL_URL: stub.url, OV_OAUTH_CLIENT_SECRET: 'ai-secret' }, seam);
+ try {
+ const missing = await uncovered(h.db, { tables: TABLES, exempt: { account_data_events: 'the ledger of requests itself: which export and deletion AI applied' } });
+ assert.deepStrictEqual(missing, []);
+ const who = userToken(A);
+ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK, source: 'cursor' }] } });
+ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: DARK_SAME, source: 'claude-code' }] } });
+ await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: userToken(B), body: { items: [{ text: 'Bo keeps this note' }] } });
+
+ const deliver = async (event) => {
+ const body = JSON.stringify({ event, seq: 1 });
+ const res = await fetch(`${h.base}/internal/events`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
+ return { status: res.status, json: await res.json().catch(() => null) };
+ };
+ const ev = (id, type, payload) => ({ event_id: id, event_type: type, source: 'network', version: 1, timestamp: new Date().toISOString(), payload });
+ const r = await deliver(ev('evt_01JZ0000000000000000000E02', 'network.account.export_requested', { export_id: EXP, subject: A }));
+ assert.deepStrictEqual([r.status, r.json.outcome], [200, 'exported']);
+ const part = stub.calls.find((c) => c.url === `/internal/account-exports/${EXP}/parts`);
+ assert.ok(part, 'the memory export part was pushed');
+ const files = Object.fromEntries(part.body.files.map((f) => [f.name, f.content]));
+ const exported = files['memory-items.json'];
+ assert.ok(exported.some((i) => i.status === 'active' && i.sources.length === 2), 'the export shows the merged item with both sources');
+ assert.ok(exported.some((i) => i.status === 'archived' && i.merged_into), 'the export shows the merged_into link');
+ assert.ok(!JSON.stringify(part.body).includes('Bo keeps this note'), 'nobody else memory');
+ assert.ok(!JSON.stringify(part.body).includes('m1:'), 'the export is decrypted, not sealed');
+
+ const d = await deliver(ev('evt_01JZ0000000000000000000D02', 'network.account.deleted', { deletion_id: DEL, subject: A, aliases: [] }));
+ assert.deepStrictEqual([d.status, d.json.outcome], [200, 'erased']);
+ assert.equal(Number(await h.db.value('SELECT count(*)::int FROM memory_items WHERE owner = $1', [`user:${A}`])), 0, 'every row of the person is gone');
+ assert.equal(Number(await h.db.value('SELECT count(*)::int FROM memory_items WHERE owner = $1', [`user:${B}`])), 1, 'the other person keeps their rows');
+ } finally { await h.stop(); await seam.close(); await stub.close(); }
+});
+
+t.test('no note text reaches the logger, the routes or the job', async () => {
+ const lines = [];
+ const log = { log: (...a) => lines.push(a.join(' ')), warn: (...a) => lines.push(a.join(' ')), error: (...a) => lines.push(a.join(' ')) };
+ const { seam } = await startSeam();
+ const h = await bootWithSeam({ AI_MEMORY_JOBS: '1' }, seam, { log });
+ try {
+ const marker = 'a-very-distinctive-note-that-must-not-be-logged';
+ const who = userToken(A);
+ const made = await request(h.base, 'POST', '/api/v1/memory/items:batch', { tok: who, body: { items: [{ text: marker }, { text: 'another note about the same thing' }] } });
+ assert.equal(made.status, 200);
+ await request(h.base, 'GET', '/api/v1/memory/profile?scope=user', { tok: who });
+ await h.memoryJobs.runOnce();
+ const all = lines.join('\n');
+ assert.ok(!all.includes(marker), `the logger saw the note: ${all}`);
+ assert.ok(!all.includes('another note about the same thing'), 'the logger saw the second note');
+ } finally { await h.stop(); await seam.close(); }
+});
+
+async function startNetworkStub() {
+ const calls = [];
+ const server = http.createServer((req, res) => {
+ const chunks = [];
+ req.on('data', (c) => chunks.push(c));
+ req.on('end', () => {
+ const json = (status, obj) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(obj)); };
+ if (req.url === '/oauth/token') return json(200, { access_token: 'tok_ai', token_type: 'Bearer', expires_in: 300 });
+ calls.push({ url: req.url, body: JSON.parse(Buffer.concat(chunks).toString() || 'null') });
+ return json(201, {});
+ });
+ });
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ return { url: `http://127.0.0.1:${server.address().port}`, calls, close: () => new Promise((r) => server.close(r)) };
+}
+
+t.run();
@@ -31,6 +31,24 @@ function userToken(subject, { aud = ['openvibe.network'] } = {}) {
@@ -31,6 +31,24 @@ function userToken(subject, { aud = ['openvibe.network'] } = {}) {
return `${input}.${crypto.sign('RSA-SHA256', Buffer.from(input), privateKey).toString('base64url')}`;
return `${input}.${crypto.sign('RSA-SHA256', Buffer.from(input), privateKey).toString('base64url')}`;
}
}
+/**
+ * The memory module's SQL statements, one by one: the comments are taken out first (an apostrophe in a comment is
+ * not a string), so every statement is checked on its own and not as one long blob.
+ */
+function memoryStatements(src) {
+ const stripped = src.replace(/\/\*[\s\S]*?\*\//g, ' ').replace(/(^|[^:])\/\/[^\n]*/g, '$1');
+ const out = [];
+ const re = /(['"`])((?:(?!\1)[^\\]|\\.)*)\1/g;
+ let m;
+ while ((m = re.exec(stripped))) {
+ const sql = m[2].trim();
+ if (/^(SELECT|UPDATE|DELETE)\b/i.test(sql) && /memory_items|memory_profiles|memory_reads/.test(sql)) out.push(sql);
+ }
+ return out;
+}
+
+/** The job's owner cursor: owner identifiers only, so it is the one statement without an owner predicate. */
+const OWNER_CURSOR = /^select\s+owner,\s*max\(updated_at\)\s+as\s+last_seen\s+from\s+memory_items\s+where\s+owner\s+is\s+not\s+null\s+and\s+owner\s*<>\s*\$\d+\s+group\s+by\s+owner\s+order\s+by\s+last_seen\s+desc\s+limit\s+\$\d+$/i;
const serviceWrite = (extra = {}) => token('codes', ['ai.memory.write'], { extra });
const serviceWrite = (extra = {}) => token('codes', ['ai.memory.write'], { extra });
const serviceRead = (extra = {}) => token('codes', ['ai.memory.read'], { extra });
const serviceRead = (extra = {}) => token('codes', ['ai.memory.read'], { extra });
const item = (text, extra = {}) => ({ text, ...extra });
const item = (text, extra = {}) => ({ text, ...extra });
@@ -121,10 +139,12 @@ t.test('one person cannot read, change or delete another person\'s item, and eve
@@ -121,10 +139,12 @@ t.test('one person cannot read, change or delete another person\'s item, and eve
const problems = [];
const problems = [];
for (const f of files) {
for (const f of files) {
const src = fs.readFileSync(path.join(dir, f), 'utf8');
const src = fs.readFileSync(path.join(dir, f), 'utf8');
- for (const m of src.matchAll(/(['"`])((?:(?!\1)[\s\S])*?(?:memory_items|memory_profiles|memory_reads)[\s\S]*?)\1/g)) {
- const sql = m[2].trim();
- if (!/^(SELECT|UPDATE|DELETE)\b/i.test(sql)) continue;
- if (!/owner = \$/.test(sql)) problems.push(`${f}: ${sql.split('\n')[0]}`);
+ for (const sql of memoryStatements(src)) {
+ if (/owner = \$/.test(sql)) continue;
+ // The one statement that is not owner-scoped: the job's owner cursor. It returns owner identifiers
+ // and nothing else (no item, no text, no id), so it cannot carry one person's data out.
+ if (OWNER_CURSOR.test(sql)) continue;
+ problems.push(`${f}: ${sql.split('\n')[0]}`);
}
}
}
}
assert.deepStrictEqual(problems, [], `a memory query is not owner-scoped: ${problems.join(' | ')}`);
assert.deepStrictEqual(problems, [], `a memory query is not owner-scoped: ${problems.join(' | ')}`);
@@ -334,7 +354,7 @@ t.test('account data: the table guard passes, the export is decrypted, and a del
@@ -334,7 +354,7 @@ t.test('account data: the table guard passes, the export is decrypted, and a del
t.test('no item text reaches the logger during a full flow', async () => {
t.test('no item text reaches the logger during a full flow', async () => {
const lines = [];
const lines = [];
const log = { log: (...a) => lines.push(a.join(' ')), warn: (...a) => lines.push(a.join(' ')), error: (...a) => lines.push(a.join(' ')) };
const log = { log: (...a) => lines.push(a.join(' ')), warn: (...a) => lines.push(a.join(' ')), error: (...a) => lines.push(a.join(' ')) };
- const h = await boot({ env: { AI_MEMORY_KEY: KEY }, log });
+ const h = await boot({ env: { AI_MEMORY_KEY: KEY, AI_MEMORY_JOBS: '1' }, log });
try {
try {
const who = userToken(A);
const who = userToken(A);
const marker = 'a-very-distinctive-note-that-must-not-be-logged';
const marker = 'a-very-distinctive-note-that-must-not-be-logged';
@@ -347,6 +367,8 @@ t.test('no item text reaches the logger during a full flow', async () => {
@@ -347,6 +367,8 @@ t.test('no item text reaches the logger during a full flow', async () => {
await request(h.base, 'GET', `/api/v1/memory/search?q=${encodeURIComponent(marker)}&scopes=user&on_behalf_of=${A}`, { tok: serviceRead({ memory_scopes: ['user'] }) });
await request(h.base, 'GET', `/api/v1/memory/search?q=${encodeURIComponent(marker)}&scopes=user&on_behalf_of=${A}`, { tok: serviceRead({ memory_scopes: ['user'] }) });
await request(h.base, 'PATCH', `/api/v1/memory/items/${id}`, { tok: who, body: { text: `${marker} edited` } });
await request(h.base, 'PATCH', `/api/v1/memory/items/${id}`, { tok: who, body: { text: `${marker} edited` } });
await request(h.base, 'DELETE', `/api/v1/memory/items/${id}`, { tok: who });
await request(h.base, 'DELETE', `/api/v1/memory/items/${id}`, { tok: who });
+ // M2c: the job (the store's own maintenance) runs here too, and it writes counts, never a note.
+ await h.memoryJobs.runOnce();
const all = lines.join('\n');
const all = lines.join('\n');
assert.ok(!all.includes(marker), `the logger saw the item text: ${all}`);
assert.ok(!all.includes(marker), `the logger saw the item text: ${all}`);
assert.ok(!all.includes(secret), 'the logger saw the secret');
assert.ok(!all.includes(secret), 'the logger saw the secret');