MailUI4Agents/scripts/repair-v3-usermessage-ids.mjs

#!/usr/bin/env node
/**
 * 修 v3 会话里 `user/message` 缺 `id`/`role` 的事件。
 *
 * 为什么需要它（与 v0 那个 repair 的区别）：
 *   - `repair-legacy-spliced-ids.mjs` 修的是 **v0** 的 `agent/inbox/spliced.inserted[]`。
 *   - 修完之后，dsh 一旦 resume 该会话，就会把 v0 迁移成 **v3**（`session.v3.jsonl.zstd`）
 *     并按当时的写入路径继续追加。写入路径会给迁移出来的历史消息补 id
 *     （`legacy-message:<sid>:<seq>`），但**新注入的**那条 `user/message`
 *     仍然是 `{content, source}` —— 没有 id/role。
 *   - 生产读路径 `JsonlSessionPersistence.open()` 会走
 *     `validateStoredEvents → adoptSessionEvent → assertMessageEventShape`，
 *     要求每条 message 事件都带非空 `id`；于是整个会话读不出来。
 *     而 `readSession()` 因为 `SessionCorpus.load` 对 live 会话直接返回内存快照
 *     （不经磁盘校验），会**成功** —— 这就是两条路径结论相反的原因。
 *
 * 本脚本只做一件事：给 v3 里缺 `id`/`role` 的 `user/message` 补上，其余字节原样保留。
 *
 * ⚠️ 补进去的 id 是**伪造**的（原数据里没有）。它足以过生产档校验、让会话重新可读，
 *    但不是"历史的真实 id"。治本仍在写入侧规范化。
 *
 * 用法：
 *   node scripts/repair-v3-usermessage-ids.mjs            # dry-run（默认）
 *   node scripts/repair-v3-usermessage-ids.mjs --apply    # 写盘（须先停 dsh.service）
 */

import { execFileSync } from "node:child_process";
import { readdirSync, statSync, copyFileSync, renameSync, writeFileSync, mkdirSync, rmSync, chmodSync, existsSync } from "node:fs";
import { join, basename } from "node:path";
import { tmpdir } from "node:os";

const dsRoot = process.env.DSH_INSTALL ?? "/usr/lib/node_modules/@deepseek-ai/dsh";
const { sessionFormatCatalog } = await import(join(dsRoot, "node_modules/@deepseek-ai/dsh-session-format-catalog/lib/index.js"));

/** v3 是当前版本，两级校验都应通过；只要 current 档过就算修好。 */
function v3Ok(header, events) {
  for (const validation of ["transformed", "current"]) {
    try {
      const r = sessionFormatCatalog.createRestore(structuredClone(header), { recovery: "recoverable", validation });
      for (const e of structuredClone(events)) r.decodeRow(e);
      r.finish();
    } catch (err) {
      return { ok: false, msg: `[${validation}] ${err?.message}` };
    }
  }
  return { ok: true };
}

const argv = process.argv.slice(2);
const APPLY = argv.includes("--apply");
const ROOT = "/root/.dsh/sessions";

function serviceActive() {
  try {
    return execFileSync("systemctl", ["is-active", "dsh.service"], { encoding: "utf8" }).trim() === "active";
  } catch { return false; }
}

const readLines = (p) => execFileSync("zstd", ["-dc", p], { maxBuffer: 1 << 30 }).toString("utf8").split("\n").filter((l) => l.trim().length);

/** 生产同款帧布局：第 1 帧恰好一行 header，其后每 500 行一帧。 */
function writeArtifact(path, header, events) {
  const tmp = join(tmpdir(), `dsh-v3fix-${process.pid}-${Date.now()}`);
  mkdirSync(tmp, { recursive: true });
  const parts = [];
  const h = join(tmp, "h.jsonl");
  writeFileSync(h, JSON.stringify(header) + "\n");
  execFileSync("zstd", ["-q", "-f", h, "-o", join(tmp, "f0.zst")]);
  parts.push(join(tmp, "f0.zst"));
  const BATCH = 500;
  for (let i = 0, k = 0; i < events.length; i += BATCH, k++) {
    const b = join(tmp, `b${k}.jsonl`);
    writeFileSync(b, events.slice(i, i + BATCH).map((o) => JSON.stringify(o)).join("\n") + "\n");
    execFileSync("zstd", ["-q", "-f", b, "-o", join(tmp, `f${k + 1}.zst`)]);
    parts.push(join(tmp, `f${k + 1}.zst`));
  }
  const staged = `${path}.repair-staged`;
  execFileSync("bash", ["-c", `cat ${parts.map((p) => JSON.stringify(p)).join(" ")} > ${JSON.stringify(staged)}`]);
  rmSync(tmp, { recursive: true, force: true });
  return staged;
}

if (APPLY && serviceActive()) {
  console.error("拒绝执行：dsh.service 正在运行。请先 `systemctl stop dsh.service` 再 --apply。");
  process.exit(2);
}

const targets = [];
for (const store of readdirSync(ROOT)) {
  const storePath = join(ROOT, store);
  if (!statSync(storePath).isDirectory()) continue;
  for (const id of readdirSync(storePath)) {
    const p = join(storePath, id, "session.v3.jsonl.zstd");
    if (existsSync(p)) targets.push({ id, path: p });
  }
}

console.log(`模式: ${APPLY ? "APPLY（写盘）" : "dry-run（不写盘）"}  v3 候选: ${targets.length}\n`);
let fixed = 0, clean = 0;
const details = [];

for (const { id, path } of targets) {
  const lines = readLines(path);
  const header = JSON.parse(lines[0]);
  const events = lines.slice(1).map((l) => JSON.parse(l));
  let patched = 0;
  const out = events.map((e) => {
    if (e.type !== "user/message") return e;
    const d = e.data ?? {};
    if (typeof d.id === "string" && d.id !== "" && d.role === "user") return e;
    patched++;
    return { ...e, data: { ...d, id: `recovered-usermessage-${e.seq}`, role: "user" } };
  });
  if (patched === 0) { clean++; continue; }
  fixed++;
  details.push([id, patched, events.length]);
  console.log(`  需修复: ${id}（补 ${patched} 条 user/message，共 ${events.length} 事件）`);

  if (APPLY) {
    const backup = `${path}.bak-${new Date().toISOString().replace(/[:.]/g, "-")}`;
    const mode = statSync(path).mode & 0o777;
    copyFileSync(path, backup);
    const staged = writeArtifact(path, header, out);
    chmodSync(staged, mode);
    // 写盘前自检：落盘字节必须两级校验都过，否则放弃（原文件未动）
    try {
      const l2 = readLines(staged);
      const check = v3Ok(JSON.parse(l2[0]), l2.slice(1).map((x) => JSON.parse(x)));
      if (!check.ok) throw new Error(check.msg);
    } catch (err) {
      rmSync(staged, { force: true });
      console.error(`    !! 自检失败，已放弃（原文件未动）: ${String(err?.message).slice(0, 140)}`);
      fixed--;
      continue;
    }
    renameSync(staged, path);
    console.log(`    已修复；备份 ${basename(backup)}`);
  }
}

console.log(`\n=== 汇总 ===`);
console.log(`  已干净（无需处理）: ${clean}`);
console.log(`  需修复:            ${fixed}${APPLY ? "（已写入）" : "（dry-run：未写）"}`);