From d9990bbcbf02bc7b6697bdfa31f7510f363b739c Mon Sep 17 00:00:00 2001 From: phil-lipp <52623794+phil-lipp@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:17:59 +0200 Subject: [PATCH 1/5] fix(user-profile): scope cold-start buffer per user and persist learning_paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cold-start buffer was a single instance field on the profile-manager singleton, so items observed for one user during embedding warm-up drained into whichever user's mergeItems ran next — merging User A's preferences into User B's profile. Key the buffer by profileId (Map) and drain only the current profile's bucket; legacy unattributed buffer files are dropped on load. Separately, createProfile/updateProfile rebuilt cleanedData as only {preferences, patterns, workflows}, silently stripping learning_paths on every write and rendering the Learning Paths injection feature dead. Carry the field through. --- .../user-profile/user-profile-manager.ts | 110 +++++++++++++----- 1 file changed, 82 insertions(+), 28 deletions(-) diff --git a/src/services/user-profile/user-profile-manager.ts b/src/services/user-profile/user-profile-manager.ts index a5aff06..cf2b397 100644 --- a/src/services/user-profile/user-profile-manager.ts +++ b/src/services/user-profile/user-profile-manager.ts @@ -70,19 +70,23 @@ function normalizeDescription(text: string): string { } const USER_PROFILES_DB_NAME = "user-profiles.db"; +const COLD_BUFFER_DEFAULT_KEY = "__unattributed__"; export class UserProfileManager { private db: TursoDb | null = null; private dbPath: string; private initPromise: Promise | null = null; - private coldBuffer: { preferences: any[]; patterns: any[]; workflows: any[] }; + // Cold-start buffers are keyed by profileId so items observed for one user never + // drain into another user's merge (cross-user contamination). The unattributed bucket + // (COLD_BUFFER_DEFAULT_KEY) only holds items from merges that ran without a profileId. + private coldBuffers: Map; private coldBufferPath: string; private dedupCheckedCache: Set = new Set(); constructor() { this.dbPath = join(CONFIG.storagePath || "", USER_PROFILES_DB_NAME); this.coldBufferPath = join(CONFIG.storagePath || "", "cold-buffer.json"); - this.coldBuffer = this.loadColdBuffer(); + this.coldBuffers = this.loadColdBuffers(); } reset(): void { @@ -128,35 +132,75 @@ export class UserProfileManager { return this.db; } - private loadColdBuffer(): { preferences: any[]; patterns: any[]; workflows: any[] } { + private emptyColdBuffer(): { preferences: any[]; patterns: any[]; workflows: any[] } { + return { preferences: [], patterns: [], workflows: [] }; + } + + private getColdBuffer(profileId?: string): { + preferences: any[]; + patterns: any[]; + workflows: any[]; + } { + const key = profileId || COLD_BUFFER_DEFAULT_KEY; + let buf = this.coldBuffers.get(key); + if (!buf) { + buf = this.emptyColdBuffer(); + this.coldBuffers.set(key, buf); + } + return buf; + } + + private loadColdBuffers(): Map< + string, + { preferences: any[]; patterns: any[]; workflows: any[] } + > { + const map = new Map(); try { if (existsSync(this.coldBufferPath)) { const raw = readFileSync(this.coldBufferPath, "utf-8"); const data = JSON.parse(raw); - if (data.preferences?.length || data.patterns?.length || data.workflows?.length) { - log("profile cold buffer: loaded from disk", { - prefs: data.preferences?.length || 0, - pats: data.patterns?.length || 0, - wfs: data.workflows?.length || 0, - }); + // Legacy flat format ({ preferences, patterns, workflows }) is cross-user + // contaminated and cannot be attributed to a profile, so it is dropped rather + // than replayed. New format is keyed by profileId. + const isLegacyFlat = + data && + typeof data === "object" && + !Array.isArray(data) && + ("preferences" in data || "patterns" in data || "workflows" in data); + if (data && typeof data === "object" && !Array.isArray(data) && !isLegacyFlat) { + let loaded = 0; + for (const [pid, v] of Object.entries(data)) { + map.set(pid, { + preferences: Array.isArray(v?.preferences) ? v.preferences : [], + patterns: Array.isArray(v?.patterns) ? v.patterns : [], + workflows: Array.isArray(v?.workflows) ? v.workflows : [], + }); + loaded++; + } + if (loaded > 0) { + log("profile cold buffer: loaded from disk", { profiles: loaded }); + } + } else if (isLegacyFlat) { + log("profile cold buffer: dropping legacy unattributed buffer"); } - return { - preferences: Array.isArray(data.preferences) ? data.preferences : [], - patterns: Array.isArray(data.patterns) ? data.patterns : [], - workflows: Array.isArray(data.workflows) ? data.workflows : [], - }; } } catch { - // 文件损坏或不存在，返回空缓冲 + // Corrupt or missing file — start with an empty buffer set. } - return { preferences: [], patterns: [], workflows: [] }; + return map; } - private saveColdBuffer(): void { + private saveColdBuffers(): void { try { - writeFileSync(this.coldBufferPath, JSON.stringify(this.coldBuffer), "utf-8"); + const obj: Record = {}; + for (const [pid, v] of this.coldBuffers.entries()) { + if (v.preferences.length || v.patterns.length || v.workflows.length) { + obj[pid] = v; + } + } + writeFileSync(this.coldBufferPath, JSON.stringify(obj), "utf-8"); } catch { - // 磁盘满或无权限时静默失败 + // Silently ignore disk-full / permission errors. } } @@ -237,6 +281,7 @@ export class UserProfileManager { preferences: safeArray(profileData.preferences), patterns: safeArray(profileData.patterns), workflows: safeArray(profileData.workflows), + ...(profileData.learning_paths ? { learning_paths: profileData.learning_paths } : {}), }; await db.run( @@ -279,6 +324,7 @@ export class UserProfileManager { preferences: safeArray(profileData.preferences), patterns: safeArray(profileData.patterns), workflows: safeArray(profileData.workflows), + ...(profileData.learning_paths ? { learning_paths: profileData.learning_paths } : {}), }; const versionRow = await db.get(`SELECT version FROM user_profiles WHERE id = ?`, [profileId]); @@ -445,6 +491,9 @@ export class UserProfileManager { async deleteProfile(profileId: string): Promise { const db = await this.ready(); await db.run(`DELETE FROM user_profiles WHERE id = ?`, [profileId]); + if (this.coldBuffers.delete(profileId)) { + this.saveColdBuffers(); + } } async getProfileById(profileId: string): Promise { @@ -594,12 +643,16 @@ export class UserProfileManager { let matchCount = 0; let newCount = 0; - if (useEmbedding && (this.coldBuffer as any)[itemType + "s"].length > 0) { - const buffered = [...(this.coldBuffer as any)[itemType + "s"]]; - (this.coldBuffer as any)[itemType + "s"] = []; - this.saveColdBuffer(); + // Scope the cold buffer to this profile so we only drain items observed for the + // same user (see COLD_BUFFER_DEFAULT_KEY for the unattributed case). + const coldBuffer = this.getColdBuffer(profileId); + if (useEmbedding && (coldBuffer as any)[itemType + "s"].length > 0) { + const buffered = [...(coldBuffer as any)[itemType + "s"]]; + (coldBuffer as any)[itemType + "s"] = []; + this.saveColdBuffers(); log("profile cold start: draining buffer", { type: itemType, + profileId: profileId || COLD_BUFFER_DEFAULT_KEY, bufferSize: buffered.length, }); incoming = [...buffered, ...incoming]; @@ -1053,15 +1106,16 @@ export class UserProfileManager { (Array.isArray((newItem as any).evidence) && (newItem as any).evidence.includes("manual-write")); if (!isExplicit) { - (this.coldBuffer as any)[itemType + "s"].push(newItem); - if ((this.coldBuffer as any)[itemType + "s"].length > 50) { - (this.coldBuffer as any)[itemType + "s"].shift(); + (coldBuffer as any)[itemType + "s"].push(newItem); + if ((coldBuffer as any)[itemType + "s"].length > 50) { + (coldBuffer as any)[itemType + "s"].shift(); } - this.saveColdBuffer(); + this.saveColdBuffers(); log("profile cold start: buffered", { type: itemType, + profileId: profileId || COLD_BUFFER_DEFAULT_KEY, cat: newItem.category, - bufferSize: (this.coldBuffer as any)[itemType + "s"].length, + bufferSize: (coldBuffer as any)[itemType + "s"].length, }); continue; } From 93b615d804c40a021eaa8ac42f902b5d2ef4f033 Mon Sep 17 00:00:00 2001 From: phil-lipp <52623794+phil-lipp@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:18:10 +0200 Subject: [PATCH 2/5] fix(user-profile): harden cleanup rebuild and injection against bad data rebuildProfileUsing dereferenced originalItem unconditionally in the merged-group branch, so a keeper id the model hallucinated (present only in its mapping, not in originalById) threw and aborted the entire cleanup run. Skip such groups, and normalize the AI-controlled mapping (kept/merged/removed) to well-formed arrays so a malformed response degrades to a no-op cleanup instead of throwing. getUserProfileContext parsed the stored profileData with no guard, so one corrupt row broke context injection for every request. Wrap in try/catch and return null. --- src/services/user-profile/ai-cleanup.ts | 32 ++++++++++++++++++-- src/services/user-profile/profile-context.ts | 9 +++++- 2 files changed, 38 insertions(+), 3 deletions(-) diff --git a/src/services/user-profile/ai-cleanup.ts b/src/services/user-profile/ai-cleanup.ts index 647f49a..5aaa5a0 100644 --- a/src/services/user-profile/ai-cleanup.ts +++ b/src/services/user-profile/ai-cleanup.ts @@ -131,6 +131,29 @@ interface AIMapping { removed: string[]; } +// The model controls this JSON; it may omit fields or return wrong types. Coerce to a +// well-formed AIMapping so rebuildProfileUsing/generateDiff never call .map/.filter/.includes +// on undefined. A missing or malformed mapping degrades to a no-op cleanup (all originals are +// preserved as "unmentioned") rather than aborting or corrupting the profile. +function normalizeAIMapping(raw: any): AIMapping { + if (!raw || typeof raw !== "object") { + log("AI cleanup: response missing valid mapping; treating as no-op", { + mappingType: typeof raw, + }); + return { kept: [], merged: [], removed: [] }; + } + const isStr = (x: any): x is string => typeof x === "string"; + const kept = Array.isArray(raw.kept) ? raw.kept.filter(isStr) : []; + const merged = Array.isArray(raw.merged) + ? raw.merged + .filter((g: any): g is any[] => Array.isArray(g)) + .map((g: any[]) => g.filter(isStr)) + .filter((g: string[]) => g.length > 0) + : []; + const removed = Array.isArray(raw.removed) ? raw.removed.filter(isStr) : []; + return { kept, merged, removed }; +} + function addIdsToProfile(profile: UserProfileData): IndexedProfile { const items = { preferences: profile.preferences.map((p, i) => ({ ...p, id: `pref_${i}` })), @@ -251,7 +274,7 @@ async function callViaExternalAPI( const parsed = JSON.parse(content); return { profile: parsed as IndexedProfile, - mapping: parsed.mapping as AIMapping, + mapping: normalizeAIMapping(parsed.mapping), }; } @@ -357,7 +380,7 @@ async function callViaOpencodeWithClient( const parsed = JSON.parse(jsonMatch[0]); return { profile: parsed as IndexedProfile, - mapping: parsed.mapping as AIMapping, + mapping: normalizeAIMapping(parsed.mapping), }; } finally { try { @@ -440,6 +463,11 @@ export function rebuildProfileUsing( } if (mergedGroups.some((g) => g[0] === id)) { + // The merge accumulation below dereferences originalItem unconditionally. If the + // model returned a keeper id that only exists in its cleaned output (a hallucinated + // id absent from originalById), skip the group instead of throwing and aborting the + // entire cleanup run. + if (!originalItem) continue; const group = mergedGroups.find((g) => g[0] === id)!; let bestFreq = (originalItem as any).frequency || 0; let bestCentroid = (originalItem as any).centroid; diff --git a/src/services/user-profile/profile-context.ts b/src/services/user-profile/profile-context.ts index fb1db64..2184e00 100644 --- a/src/services/user-profile/profile-context.ts +++ b/src/services/user-profile/profile-context.ts @@ -50,7 +50,14 @@ export async function getUserProfileContext(userId: string): Promise Date: Thu, 13 Aug 2026 13:18:20 +0200 Subject: [PATCH 3/5] fix(user-memory-learning): await profile writes and evolve before serialization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fire-and-forget/unawaited hazards in the learning cycle: - The retry-exhausted branch did not await markMultipleAsUserLearningCaptured, so finally cleared isLearningRunning while the write was in flight; the next cycle re-fetched and re-analyzed the same prompts (token burn), and the rejection was unhandled. Await it. - evolveAndUpdate mutates item.description/centroid in place but was called fire-and-forget, racing the JSON.stringify in updateProfile — the evolved description was included or lost nondeterministically. Make applyValidations async and await the evolve so mutation completes pre-serialization. - performUserProfileLearning had try/finally but no catch; JSON.parse of a corrupt profileData row rejected the promise (unhandled at the fire-and-forget site). Add a catch that logs and returns. --- src/services/user-memory-learning.ts | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/src/services/user-memory-learning.ts b/src/services/user-memory-learning.ts index eebbe62..f567fef 100644 --- a/src/services/user-memory-learning.ts +++ b/src/services/user-memory-learning.ts @@ -215,7 +215,7 @@ Rules: updatedProfileData ); - const validationSummary = applyValidations( + const validationSummary = await applyValidations( updatedProfileData, llmResult, existingProfile.id, @@ -248,7 +248,7 @@ Rules: profileId: existingProfile?.id, userId, }); - userPromptManager.markMultipleAsUserLearningCaptured(prompts.map((p) => p.id)); + await userPromptManager.markMultipleAsUserLearningCaptured(prompts.map((p) => p.id)); return; } @@ -281,6 +281,11 @@ Rules: }) .catch(() => {}); } + } catch (error) { + // Guard against corrupt stored profileData (JSON.parse throws) and any other + // fault: this runs fire-and-forget from the idle timer, so an uncaught rejection + // would surface as an unhandled promise rejection. Log and exit cleanly. + log("user-profile-learning: aborted", { error: String(error) }); } finally { isLearningRunning = false; } @@ -549,12 +554,12 @@ export function createUserProfileToolSchema(existingProfile: boolean) { type AnalysisResult = { raw: UserProfileData; merged: UserProfileData | null }; -function applyValidations( +async function applyValidations( profileData: UserProfileData, llmResult: UserProfileData, profileId: string, prefKeys?: string[] -): string | null { +): Promise { const validations = (llmResult as any).validations as | Array<{ index: number; @@ -611,7 +616,12 @@ function applyValidations( const evidence = (item as any).evidence; if (Array.isArray(evidence) && evidence.length >= 3) { const itemType = profileData.preferences.includes(item) ? "preference" : "pattern"; - userProfileManager.evolveAndUpdate(item, itemType, profileId).catch(() => {}); + // Await so the in-place description/centroid mutation completes before the + // caller serializes updatedProfileData — otherwise the evolved description is + // included or lost nondeterministically. Failures stay non-fatal. + try { + await userProfileManager.evolveAndUpdate(item, itemType, profileId); + } catch {} } } else { results.push(`no_evidence [${v.index}] ${v.reason}`); From 715cef4926a9e3d5443b5bd4f1c8b042bf523008 Mon Sep 17 00:00:00 2001 From: phil-lipp <52623794+phil-lipp@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:18:37 +0200 Subject: [PATCH 4/5] fix(turso): safe shard creation, exact shard-path match, resilient lock cleanup - createShard committed the registry INSERT before initShardDb ran; if init threw (disk full, permissions) the row persisted pointing at an uninitialized file, so the next getWriteShard failed isShardValid and threw 'incompatible or corrupt', blocking all writes to that scope. Initialize the shard DB first (it is idempotent), then insert. - getShardByPath matched db_path with LIKE '%' || filename, so the underscores in shard names (user__shard_N.db) acted as single-char wildcards and could match the wrong row. Anchor on the '/' separator and escape LIKE metacharacters. - readLiveLock called unlinkSync outside its try/catch; a race (already removed) or a Windows open handle threw ENOENT/EPERM out of assertNoTursoMigrationInProgress and falsely blocked writes. Wrap it. --- src/services/turso/operation-lock.ts | 7 ++++++- src/services/turso/shard-manager.ts | 21 +++++++++++++++------ 2 files changed, 21 insertions(+), 7 deletions(-) diff --git a/src/services/turso/operation-lock.ts b/src/services/turso/operation-lock.ts index 1219f8b..35d8224 100644 --- a/src/services/turso/operation-lock.ts +++ b/src/services/turso/operation-lock.ts @@ -30,7 +30,12 @@ function readLiveLock(path: string): LockState | null { } catch { // Corrupt locks are stale and removed below. } - unlinkSync(path); + try { + unlinkSync(path); + } catch { + // Already removed (race) or held open (Windows) — either way the lock is + // gone or unowned; do not let cleanup failure block writes. + } return null; } diff --git a/src/services/turso/shard-manager.ts b/src/services/turso/shard-manager.ts index 867c9ff..14f79b0 100644 --- a/src/services/turso/shard-manager.ts +++ b/src/services/turso/shard-manager.ts @@ -180,6 +180,13 @@ export class TursoShardManager { const storedPath = join(`${scope}s`, basename(fullPath)).replace(/\\/g, "/"); const now = Date.now(); + // Initialize the shard file BEFORE inserting the registry row. If init throws + // (disk full, permissions), the registry stays free of an orphan row that points + // at an uninitialized file — such a row would later fail isShardValid on every + // getWriteShard and brick all writes to this scope. initShardDb is idempotent. + const shardDb = await tursoConnectionManager.getConnection(fullPath); + await this.initShardDb(shardDb); + let result; try { result = await metadataDb.execute( @@ -205,9 +212,6 @@ export class TursoShardManager { throw error; } - const shardDb = await tursoConnectionManager.getConnection(fullPath); - await this.initShardDb(shardDb); - return { id: Number(result.lastInsertRowid), scope, @@ -487,9 +491,14 @@ export class TursoShardManager { async getShardByPath(dbPath: string): Promise { const metadataDb = await this.ensureInitialized(); const fileName = basename(dbPath); - const row = await metadataDb.get(`SELECT * FROM shards WHERE db_path LIKE '%' || ?`, [ - fileName, - ]); + // Stored db_path is always `s/` (see createShard/registerExistingShard), + // so anchor on the "/" separator. Escape LIKE metacharacters in the filename — otherwise + // the "_" in shard names like `user__0.db` would match any character. + const escaped = fileName.replace(/[\\%_]/g, "\\$&"); + const row = await metadataDb.get( + `SELECT * FROM shards WHERE db_path LIKE '%/' || ? ESCAPE '\\'`, + [escaped] + ); if (!row) return null; return this.rowToShardInfo(row); } From e79af2441beda6cef23ce8e5c1cb8015d4766a5a Mon Sep 17 00:00:00 2001 From: phil-lipp <52623794+phil-lipp@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:18:48 +0200 Subject: [PATCH 5/5] test(user-profile): cover per-user cold-buffer isolation Asserts that items buffered for one profile during embedding cold-start never drain into another profile's merge, and that each bucket drains only for its own profile. --- ...user-profile-cold-buffer-isolation.test.ts | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 tests/user-profile-cold-buffer-isolation.test.ts diff --git a/tests/user-profile-cold-buffer-isolation.test.ts b/tests/user-profile-cold-buffer-isolation.test.ts new file mode 100644 index 0000000..0ae83c4 --- /dev/null +++ b/tests/user-profile-cold-buffer-isolation.test.ts @@ -0,0 +1,81 @@ +import { afterEach, beforeEach, describe, expect, it } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { tursoConnectionManager } from "../src/services/turso/connection-manager.js"; + +let tmpDir: string; + +async function makeManager() { + const { CONFIG } = await import("../src/config.js"); + CONFIG.storagePath = tmpDir; + CONFIG.userProfileEmbeddingMinDescriptionLength = 5; + const { UserProfileManager } = + await import("../src/services/user-profile/user-profile-manager.js"); + return { mgr: new UserProfileManager(), CONFIG }; +} + +// Embedding not warmed up → mergeItems buffers non-explicit items (cold start). +const coldEmbed = { isWarmedUp: false } as any; +// Warmed up → buffered items drain into the merge. embed() is only used to seed a +// centroid on append; existing is empty here so no cosine comparison runs. +const warmEmbed = { + isWarmedUp: true, + embed: async () => new Float32Array(8).fill(0.25), +} as any; + +const empty = () => ({ preferences: [], patterns: [], workflows: [] }); + +describe("cold buffer per-user isolation (correctness #1)", () => { + beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), "opencode-mem-coldbuf-")); + }); + + afterEach(async () => { + await tursoConnectionManager.closeAll(); + await new Promise((r) => setTimeout(r, 50)); + try { + rmSync(tmpDir, { recursive: true, force: true }); + } catch {} + }); + + it("does not drain one user's buffered items into another user's merge", async () => { + const { mgr } = await makeManager(); + + // Cold start: each user's observation is buffered under its own profileId. + await mgr.mergeProfileData( + empty(), + { preferences: [{ category: "style", description: "A prefers tabs over spaces" }] }, + coldEmbed, + "profile_A" + ); + await mgr.mergeProfileData( + empty(), + { preferences: [{ category: "style", description: "B prefers spaces over tabs" }] }, + coldEmbed, + "profile_B" + ); + + // Warm merge for B must drain only B's bucket, never A's. + const mergedB = await mgr.mergeProfileData( + empty(), + { preferences: [] }, + warmEmbed, + "profile_B" + ); + const descsB = mergedB.preferences.map((p: any) => p.description); + expect(descsB).toContain("B prefers spaces over tabs"); + expect(descsB).not.toContain("A prefers tabs over spaces"); + + // A's bucket is untouched by B's drain and drains only for A. + const mergedA = await mgr.mergeProfileData( + empty(), + { preferences: [] }, + warmEmbed, + "profile_A" + ); + const descsA = mergedA.preferences.map((p: any) => p.description); + expect(descsA).toContain("A prefers tabs over spaces"); + expect(descsA).not.toContain("B prefers spaces over tabs"); + }); +});