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); } 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}`); 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 | 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; } 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"); + }); +});