fix(persona-lib): 檔案層互斥鎖，read-modify-rewrite 不再吃掉並行的 append

short-term.jsonl / said.jsonl 的裁切是「整檔讀進來 → 過濾 → writeText 覆蓋」，中間沒有任何鎖。而同一個 session 的 sub agent 與主程序共用同一把人格鎖（設計如此），所以兩邊真的會同時寫——實測背景 prune 進行中 append 30 筆 salience 95 的承諾，會被吃掉 1～9 筆，正是 shortTermProtected() 明文要保護的那一類。 emotion.json 更嚴重：並行 8 次 `--apply joy=+5`，循序得 50.24，並行只得 38.91（增量遺失 45%）。加一把用 `fs.openSync(path, "wx")` sentinel 做的檔案鎖： * withFileLock(file, fn)：拿不到就退讓重試，超過 15 秒的殘留鎖視為死鎖並接手，真的等不到就直接做（寧可冒一次競態，也不要因為殘留鎖檔讓人格從此寫不進東西）。 sentinel 放 .runtime/locks/，不落在人格目錄裡，不會被同步上去。 * rewriteJsonl(file, transform)：整檔改寫的唯一入口，讀與寫都在鎖裡。 * updateJson / updateEmotion：JSON 檔的 read-modify-write 同樣進鎖。 * appendJsonl 也拿同一把鎖，否則 append 仍會落在別人的讀與寫之間被覆蓋掉。改用新入口的：pruneShortTermDetail、trimSaid、trimJsonl、turnContext 的情緒衰減、 persona.mjs 的 `emotion --apply`（讀→衰減→套用→寫回整段在鎖裡）。 selftest +5：並行 append vs prune 一筆不掉、並行 emotion 與循序同值、 withFileLock 互斥（12 程序各加一次＝12）、過期鎖檔可接手、鎖檔不落在人格目錄。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-31 09:19:39 +00:00
parent 9e7a9cf8e2
commit d9f865d7a4
3 changed files with 295 additions and 63 deletions
@@ -170,10 +170,120 @@ export function writeText(file, text) {
  fs.renameSync(tmp, file);
 }

-/** 單行 append（O_APPEND 對單行寫入是原子的），guest 也能安全使用。 */
+// --------------------------------------------------------------------------- //
+// 檔案層互斥鎖
+//
+// 人格鎖（`state/lock.json`）擋的是「兩個人格同時被載入」，不是「兩個程序同時寫同一個檔」：
+// 同一個 session 的 sub agent 與主程序**共用同一把人格鎖**（設計如此），所以兩邊真的會
+// 同時寫。而 said / short-term 的裁切是「整檔讀進來 → 過濾 → 覆蓋」，中間任何一筆 append
+// 都會被那次覆蓋吃掉——實測背景 prune 進行中寫入 30 筆 salience 95 的承諾，會掉幾筆。
+//
+// 所以：凡是碰同一個檔案的 append 與 rewrite，一律先拿這把鎖。
+// sentinel 檔用 `O_EXCL` 建（跨程序原子），放在 `.runtime/locks/` 而不是人格目錄裡，
+// 免得殘留的鎖檔被同步到 Gitea 或被 guard 掃到。
+// --------------------------------------------------------------------------- //
+
+export const FILE_LOCK_STALE_MS = 15_000; // 這麼久沒放掉就當持有者死了（避免整個系統卡住）
+export const FILE_LOCK_TIMEOUT_MS = 10_000;
+
+const SLEEP_BUF = new Int32Array(new SharedArrayBuffer(4));
+/** 同步小睡（這裡不能用 await：呼叫端全是同步 API）。 */
+function sleepSync(ms) {
+  if (ms > 0) Atomics.wait(SLEEP_BUF, 0, 0, ms);
+}
+
+export function fileLockPath(file) {
+  const key = crypto.createHash("md5").update(path.resolve(file)).digest("hex").slice(0, 16);
+  return path.join(runtimeDir(), "locks", `${path.basename(file)}.${key}.lock`);
+}
+
+/**
+ * 拿著 `file` 的互斥鎖跑 `fn()`，回傳 `fn` 的結果。
+ *
+ * 拿不到就短暫重試（含亂數退讓，避免兩邊同步互踩）；超過 `FILE_LOCK_STALE_MS`
+ * 沒被放掉的鎖視為死鎖並搶走。真的等不到就直接做——寧可冒一次競態，
+ * 也不要因為一個殘留的鎖檔讓人格從此寫不進東西。
+ */
+export function withFileLock(file, fn, { timeoutMs = FILE_LOCK_TIMEOUT_MS } = {}) {
+  const lock = fileLockPath(file);
+  fs.mkdirSync(path.dirname(lock), { recursive: true });
+  const deadline = Date.now() + timeoutMs;
+  let fd = null;
+  while (fd === null) {
+    try {
+      fd = fs.openSync(lock, "wx", 0o600);
+    } catch (err) {
+      if (err.code !== "EEXIST") throw err;
+      let stat = null;
+      try {
+        stat = fs.statSync(lock);
+      } catch {
+        continue; // 剛好被放掉了，再試一次
+      }
+      if (Date.now() - stat.mtimeMs > FILE_LOCK_STALE_MS) {
+        try {
+          fs.unlinkSync(lock);
+        } catch {
+          /* 別人先清掉了 */
+        }
+        continue;
+      }
+      if (Date.now() > deadline) break; // 等太久：不擋住呼叫端
+      sleepSync(1 + Math.floor(Math.random() * 5));
+    }
+  }
+  try {
+    if (fd !== null) fs.writeSync(fd, `${process.pid}\n`);
+    return fn();
+  } finally {
+    if (fd !== null) {
+      try {
+        fs.closeSync(fd);
+      } catch {
+        /* ignore */
+      }
+      try {
+        fs.unlinkSync(lock);
+      } catch {
+        /* ignore */
+      }
+    }
+  }
+}
+
+/** 單行 append。拿檔案鎖，才不會被同時進行的整檔覆蓋（prune／trim）吃掉。 */
 export function appendJsonl(file, obj) {
  fs.mkdirSync(path.dirname(file), { recursive: true });
-  fs.appendFileSync(file, JSON.stringify(obj) + "\n", { encoding: "utf8", mode: 0o600 });
+  const line = JSON.stringify(obj) + "\n";
+  withFileLock(file, () => {
+    fs.appendFileSync(file, line, { encoding: "utf8", mode: 0o600 });
+  });
+}
+
+/**
+ * 「整檔讀進來 → 過濾 → 覆蓋」的唯一入口：整段在鎖裡面做，
+ * 所以 `transform` 看到的一定是最新的內容，寫回去也不會蓋掉別人剛 append 的行。
+ *
+ * `transform(rows)` 回傳要留下的列；回傳的陣列跟原本一樣長就不寫（省一次 IO）。
+ */
+export function rewriteJsonl(file, transform) {
+  return withFileLock(file, () => {
+    const rows = readJsonl(file);
+    const kept = transform(rows) ?? rows;
+    if (kept.length !== rows.length) {
+      writeText(file, kept.map((r) => JSON.stringify(r)).join("\n") + (kept.length ? "\n" : ""));
+    }
+    return { rows, kept };
+  });
+}
+
+/** JSON 檔的 read-modify-write：`mutate(data)` 回傳要寫回去的物件。 */
+export function updateJson(file, mutate, fallback = null) {
+  return withFileLock(file, () => {
+    const next = mutate(readJson(file, fallback));
+    writeJson(file, next);
+    return next;
+  });
 }

 export function readJsonl(file, limit = null) {
@@ -266,6 +376,20 @@ export function loadEmotion(slug) {
  return state;
 }

+/**
+ * 情緒的 read-modify-write：`mutate(state)` 回傳要寫回去的狀態，整段在檔案鎖裡。
+ *
+ * `emotion.json` 是最容易掉更新的一個檔——它每輪都被讀出來、改一點、整份寫回去。
+ * 實測連續 8 次 `--apply joy=+5`，並行時大約有四成的增量直接消失。
+ */
+export function updateEmotion(slug, mutate) {
+  return withFileLock(emotionPath(slug), () => {
+    const next = mutate(loadEmotion(slug));
+    writeJson(emotionPath(slug), next);
+    return next;
+  });
+}
+
 /** 情緒朝 baseline 指數衰減；半衰期依情緒種類不同。 */
 export function decayEmotion(state, now = new Date()) {
  const last = parseIso(state.updated_at) ?? now;
@@ -1143,16 +1267,13 @@ export function archiveStaleThreads(slug, days = THREAD_STALE_DAYS) {

 /** `said.jsonl` 只服務「不要重講」判定，留最近 24 小時／300 行就夠。 */
 export function trimSaid(slug, { hours = SAID_KEEP_HOURS, lines = SAID_KEEP_LINES } = {}) {
-  const file = saidPath(slug);
-  const rows = readJsonl(file);
-  if (!rows.length) return 0;
  const cutoff = Date.now() - hours * 3_600_000;
-  let kept = rows.filter((r) => (parseIso(r.ts)?.getTime() ?? Date.now()) >= cutoff);
-  kept = kept.slice(-lines);
-  if (kept.length !== rows.length) {
-    writeText(file, kept.map((r) => JSON.stringify(r)).join("\n") + (kept.length ? "\n" : ""));
-  }
-  return kept.length;
+  return rewriteJsonl(saidPath(slug), (rows) => {
+    if (!rows.length) return rows;
+    return rows
+      .filter((r) => (parseIso(r.ts)?.getTime() ?? Date.now()) >= cutoff)
+      .slice(-lines);
+  }).kept.length;
 }

 /** 把「上個月以前」的 journal 壓成 .jsonl.gz（同步時省流量，也不再被讀）。 */
@@ -1352,36 +1473,36 @@ function shortTermProtected(row, now = Date.now()) {
 * 「還沒經過判斷就把今天清掉」是這裡最不能犯的錯。
 */
 export function pruneShortTermDetail(slug) {
-  const file = shortTermPath(slug);
-  const rows = readJsonl(file);
-  if (!rows.length) return { kept: 0, dropped: 0, by_age: 0, by_capacity: 0, protected: 0 };
  const now = Date.now();
  const cutoff = now - SHORT_TERM_DAYS * 86_400_000;
-  let kept = rows.filter((r) => (parseIso(r.ts)?.getTime() ?? now) >= cutoff);
-  const byAge = rows.length - kept.length;
-  // R6：超過軟上限，從「顯著度最低、最舊」開始清，但保護清單裡的不動
-  let byCapacity = 0;
-  if (kept.length > SHORT_TERM_SOFT_CAP) {
-    const droppable = kept
-      .map((row, i) => ({ row, i, protectedRow: shortTermProtected(row, now) }))
-      .filter((x) => !x.protectedRow)
-      .sort((a, b) => (Number(a.row.salience || 0) - Number(b.row.salience || 0)) || (a.i - b.i));
-    const need = kept.length - SHORT_TERM_SOFT_CAP;
-    const drop = new Set(droppable.slice(0, need).map((x) => x.i));
-    byCapacity = drop.size;
-    kept = kept.filter((_, i) => !drop.has(i));
-  }
-  const before = kept.length;
-  kept = kept.slice(-SHORT_TERM_KEEP);
-  const byHardCap = before - kept.length;
-  if (kept.length !== rows.length) {
-    writeText(file, kept.map((r) => JSON.stringify(r)).join("\n") + (kept.length ? "\n" : ""));
-  }
+  const stats = { by_age: 0, by_capacity: 0 };
+  // 讀 → 過濾 → 覆蓋整段都在檔案鎖裡：中途 append 進來的新紀錄不會被這次覆蓋吃掉
+  const { rows, kept } = rewriteJsonl(shortTermPath(slug), (all) => {
+    if (!all.length) return all;
+    let live = all.filter((r) => (parseIso(r.ts)?.getTime() ?? now) >= cutoff);
+    stats.by_age = all.length - live.length;
+    // R6：超過軟上限，從「顯著度最低、最舊」開始清，但保護清單裡的不動
+    if (live.length > SHORT_TERM_SOFT_CAP) {
+      const droppable = live
+        .map((row, i) => ({ row, i, protectedRow: shortTermProtected(row, now) }))
+        .filter((x) => !x.protectedRow)
+        .sort((a, b) => (Number(a.row.salience || 0) - Number(b.row.salience || 0)) || (a.i - b.i));
+      const need = live.length - SHORT_TERM_SOFT_CAP;
+      const drop = new Set(droppable.slice(0, need).map((x) => x.i));
+      stats.by_capacity = drop.size;
+      live = live.filter((_, i) => !drop.has(i));
+    }
+    const before = live.length;
+    live = live.slice(-SHORT_TERM_KEEP);
+    stats.by_capacity += before - live.length;
+    return live;
+  });
+  if (!rows.length) return { kept: 0, dropped: 0, by_age: 0, by_capacity: 0, protected: 0 };
  return {
    kept: kept.length,
    dropped: rows.length - kept.length,
-    by_age: byAge,
-    by_capacity: byCapacity + byHardCap,
+    by_age: stats.by_age,
+    by_capacity: stats.by_capacity,
    protected: kept.filter((r) => shortTermProtected(r, now)).length,
  };
 }
@@ -1953,11 +2074,7 @@ export function findRepeat(entries, text, {
 }

 function trimJsonl(file, keep) {
-  const rows = readJsonl(file);
-  if (rows.length <= keep * 1.5) return rows.length;
-  const kept = rows.slice(-keep);
-  writeText(file, kept.map((r) => JSON.stringify(r)).join("\n") + "\n");
-  return kept.length;
+  return rewriteJsonl(file, (rows) => (rows.length <= keep * 1.5 ? rows : rows.slice(-keep))).kept.length;
 }

 /** 記下「說出口的話」。同一句話 10 分鐘內只記一次（避免 room post 與 Stop hook 重複記）。 */
@@ -3116,8 +3233,7 @@ export function opsBrief(slug) {

 /** UserPromptSubmit 注入的人格上下文：身分 + 情緒 + 短期記憶 + 相關長期記憶 + 關係。 */
 export function turnContext(slug, sessionId, prompt = "") {
-  const state = decayEmotion(loadEmotion(slug));
-  writeJson(emotionPath(slug), state);
+  const state = updateEmotion(slug, (s) => decayEmotion(s));
  const session = loadSession(sessionId);
  const lines = [
    "<persona-context>",