fix: correctness & data-integrity fixes across user profiles and turso sharding

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;
 }
 

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<ShardInfo | null> {
     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 `<scope>s/<basename>` (see createShard/registerExistingShard),
+    // so anchor on the "/" separator. Escape LIKE metacharacters in the filename — otherwise
+    // the "_" in shard names like `user_<hash>_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);
   }

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<string | null> {
   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}`);

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;

src/services/user-profile/profile-context.ts

@@ -50,7 +50,14 @@ export async function getUserProfileContext(userId: string): Promise<string | nu
     return null;
   }
 
-  const profileData: UserProfileData = JSON.parse(profile.profileData);
+  let profileData: UserProfileData;
+  try {
+    profileData = JSON.parse(profile.profileData);
+  } catch (e) {
+    // A single corrupt row must not break context injection for everyone.
+    log("profile context: failed to parse stored profileData", { error: String(e) });
+    return null;
+  }
   const parts: string[] = [];
 
   const injectPrefs = CONFIG.userProfileInjectPreferences ?? 5;

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<void> | 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<string, { preferences: any[]; patterns: any[]; workflows: any[] }>;
   private coldBufferPath: string;
   private dedupCheckedCache: Set<string> = 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<string, { preferences: any[]; patterns: any[]; workflows: any[] }>();
     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<any>(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<string, { preferences: any[]; patterns: any[]; workflows: any[] }> = {};
+      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<void> {
     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<UserProfile | null> {
@@ -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;
         }