fix(trigger): 幂等键 crash-atomicity 收口（收 codex #776 三轮 6 blocker，全部 fail-closed）

codex 三轮（review 4878310684）确认 v3 方向对、happy-path 已证；剩 6 组崩溃/IO 故障
原子性 blocker，全部按 fail-closed 修 + 补故障注入测试：

1. reserved 清理非 snapshot CAS（旧快照能删掉已推进到 attempting 的 fence）→ 新增
   compareAndRemoveByPath(fp, expect)：锁内重读 + 完整 identity/revision/state CAS，只删仍
   匹配快照的记录；不匹配保留。reconcile 改用它（替代无条件 removeByPathLocked）。
   测试：advancedState=attempting 后旧快照 remove → fence 仍在。
2. compareAndRemove 把 EIO/EROFS/EACCES 吞成 already-gone→返 true → 新增 strictUnlink：
   仅 ENOENT 当已删，其余 throw。调用方（barrier 前释放）检查返回。
3. durable failed 直接 continue 不 quarantine（崩在写 failed 后 close 前→下轮 restore 重注册）
   → reconcile 对 failed 也每次 quarantine + 重试 closeSession。
4. reconcile per-record / corrupt / daemon 外层错误被吞后继续 bind（注释写 fail readiness、
   行为 fail-open）→ reconcile 收集硬失败并在 sweep 后 throw；listAll({throwOnCorrupt})
   corrupt lease 直接抛（不 silent skip）；daemon.ts 改为 reconcile 抛错则 throw 中止该 bot
   启动（不再 log-and-continue bind）。
5. barrier 后 recordFailedStrict+close 双失败仍返回 state:failed（磁盘故障下轮询永久 running）
   → 只有 recordFailedStrict 成功（terminal 真 durable）才返回 state:failed；写也失败则返
   5xx trigger_failed（诚实硬错，lease 留 attempting 交下轮 reconcile）。测试：async 目标路径
   预置为目录使 strict 写失败 → 断言非 phantom failed、errorCode trigger_failed。
6. recordFailedStrict 写 strict 但读用 soft load()（corrupt/EIO/invalid 当空文件覆盖，可能抹掉
   completed/owner 证据）→ 新增 loadStrict：仅 ENOENT 当 absent，其余 throw；覆盖前校验
   owner 不匹配则 throw。测试：corrupt 文件不被覆盖、completed-wins、late-completed-wins、
   owner-proof、EIO throw。

验证：pnpm build 绿；affected+shared 10 套件 311/311 绿（store 17 / async-store 24 含 7 故障注入 /
trigger-session-idempotency 12 / e2e 5 含双故障 / trigger-api / api-only readiness 序 / …）。
docs 主路径契约不变（崩溃语义 caller-visible 不变）。普通 trigger/webhook 零行为变化。

Co-Authored-By: Claude <noreply@anthropic.com>

Author: deepcoldy

src/core/trigger-session.ts

@@ -233,34 +233,56 @@ export async function reconcileIdempotencyLeasesOnBoot(
 ): Promise<Set<string>> {
   const now = Date.now();
   const quarantined = new Set<string>();
-  for (const { file, record } of idempotencyStore.listAll()) {
+  // Fail-closed: any lease we cannot PROVE converged (terminal not durable, or a
+  // corrupt/unreadable lease we can't reason about) makes the whole reconcile
+  // throw — the daemon must then abort this bot's startup rather than bind and
+  // let a poller hang `running` or an orphan re-attach. We finish the sweep to
+  // converge everything we can, but remember the first hard failure and rethrow.
+  let hardFailure: Error | undefined;
+  const leases = idempotencyStore.listAll({ throwOnCorrupt: true }); // corrupt → throw (unprovable)
+  for (const { file, record } of leases) {
+    // Owner scoping (fail-closed) + skip current boot's own in-flight leases.
+    if (record.ownerLarkAppId !== ownerLarkAppId) continue;
+    if (record.ownerBootId === currentBootId) continue;
     try {
-      // Owner scoping (fail-closed) + skip current boot's own in-flight leases.
-      if (record.ownerLarkAppId !== ownerLarkAppId) continue;
-      if (record.ownerBootId === currentBootId) continue;
       const outcome = asyncTriggerStore.lookup(record.sessionId, record.triggerId)?.result.status;
-      if (outcome === 'completed' || outcome === 'failed') continue; // already converged
+      if (outcome === 'completed') continue; // converged good; retry reuses + polls
+      if (outcome === 'failed') {
+        // Already durable-failed, but a PREVIOUS boot may have crashed after
+        // writing failed and before closing → always re-quarantine and re-attempt
+        // close, so restore never re-attaches a session the caller already saw failed.
+        quarantined.add(record.sessionId);
+        if (getSession(record.sessionId)) await closeSession(record.sessionId);
+        continue;
+      }
       if (record.state === 'attempting') {
-        // Authoritative terminal FIRST (throws on failure → surfaced to caller,
-        // which fails boot readiness rather than silently leaving `running`).
+        // Write the authoritative terminal FIRST (throws on I/O failure), THEN
+        // quarantine + close. Quarantine happens regardless of close success.
         asyncTriggerStore.recordFailedStrict(record.sessionId, record.triggerId, now, ownerLarkAppId, 'dispatch_unknown');
         quarantined.add(record.sessionId);
-        if (getSession(record.sessionId)) {
-          try { await closeSession(record.sessionId); } catch (e) { logger.warn(`[idempotency] reconcile close ${record.sessionId} failed (terminal already durable): ${(e as Error).message}`); }
-        }
+        if (getSession(record.sessionId)) await closeSession(record.sessionId);
         continue;
       }
-      // reserved: provably never dispatched → drop the lease (by enumerated path,
-      // under its own lock) + close the empty session.
-      idempotencyStore.removeByPathLocked(file);
+      // reserved: provably never dispatched → CAS-remove by path (only if the
+      // on-disk record is still this exact reserved snapshot — never delete a
+      // fence that advanced to attempting), + close the empty session.
+      idempotencyStore.compareAndRemoveByPath(file, record);
       quarantined.add(record.sessionId);
-      if (getSession(record.sessionId)) {
-        try { await closeSession(record.sessionId); } catch (e) { logger.warn(`[idempotency] reconcile close ${record.sessionId} failed: ${(e as Error).message}`); }
-      }
+      if (getSession(record.sessionId)) await closeSession(record.sessionId);
     } catch (err) {
-      logger.warn(`[idempotency] boot reconcile skipped a lease: ${(err as Error).message}`);
+      // This lease could not be converged (strict-failed write threw, CAS-remove
+      // threw on EIO, or close threw). Do NOT skip-and-continue as "handled":
+      // record it and keep the session quarantined so restore can't revive it,
+      // then fail the whole reconcile after the sweep.
+      quarantined.add(record.sessionId);
+      const e = err as Error;
+      logger.error(`[idempotency] reconcile could not converge lease for ${record.sessionId}: ${e.message}`);
+      if (!hardFailure) hardFailure = e;
     }
   }
+  if (hardFailure) {
+    throw new Error(`idempotency boot reconcile failed to converge at least one lease: ${hardFailure.message}`);
+  }
   return quarantined;
 }
 
@@ -1027,11 +1049,33 @@ export async function triggerSessionTurn(
         ? triggerId
         : { turnId: triggerId, dispatchAttempt });
     } catch (err) {
+      // The ONLY thing that lets us honestly report a terminal `failed` is a
+      // DURABLE failed record (that is what trigger-result reads). If the strict
+      // write itself fails (disk full/EIO), we must NOT claim `state:failed` —
+      // the caller could never observe it and would see `running` forever. In
+      // that double-failure case return a 5xx so the caller treats it as an
+      // unknown hard error (and the next boot's reconcile will converge the
+      // still-`attempting` lease). Only on a successful durable write do we
+      // return the terminal failed. (finding: double storage failure must be a
+      // fail-closed 5xx, not a phantom `failed`.)
+      let terminalDurable = false;
       if (idempotencyKey) {
-        try { asyncTriggerStore.recordFailedStrict(session.sessionId, triggerId, Date.now(), larkAppId, 'dispatch_unknown'); }
-        catch (e) { logger.error(`[idempotency] failed to record dispatch_unknown after dispatch throw: ${(e as Error).message}`); }
+        try {
+          asyncTriggerStore.recordFailedStrict(session.sessionId, triggerId, Date.now(), larkAppId, 'dispatch_unknown');
+          terminalDurable = true;
+        } catch (e) {
+          logger.error(`[idempotency] dispatch threw AND recordFailedStrict failed — lease stays attempting for next-boot reconcile: ${(e as Error).message}`);
+        }
+      }
+      try { await closeSession(session.sessionId); } catch { /* best-effort; terminal already durable if terminalDurable */ }
+      if (idempotencyKey && !terminalDurable) {
+        return {
+          ok: false, errorCode: 'trigger_failed',
+          error: `dispatch failed and terminal outcome could not be persisted: ${(err as Error).message}`,
+          target: { kind: 'turn', sessionId: session.sessionId, chatId },
+          idempotencyKey,
+        };
       }
-      try { await closeSession(session.sessionId); } catch { /* best-effort */ }
       return {
         ok: false, state: 'failed', triggerId,
         errorCode: 'no_output', error: `dispatch failed with unknown outcome: ${(err as Error).message}`,

src/daemon.ts

@@ -18971,18 +18971,23 @@ export async function startDaemon(botIndex?: number): Promise<void> {
   // interleave with the sweep and have its fresh lease mistaken for stale) and is
   // scoped to THIS bot (the dataDir is shared across bots). Converges leases left
   // by a previous boot: `attempting` → durable failed(dispatch_unknown) + close;
-  // `reserved` → drop + close. Returns the sessionIds it terminalized/closed so
-  // restoreActiveSessions can quarantine them from re-attach (else a session the
+  // `reserved` → CAS-remove + close. Returns the sessionIds it terminalized/closed
+  // so restoreActiveSessions can quarantine them from re-attach (else a session the
   // poller now sees `failed` could be reattached and keep running — state/exec
   // divergence). sessionStore is init'd + worker pool is up by here.
-  let idempotencyQuarantinedSessionIds = new Set<string>();
+  //
+  // FAIL-CLOSED: if reconcile throws (a lease it could not prove converged — a
+  // strict-failed write that failed, a corrupt lease, an unlink/close that
+  // errored), we must NOT bind the IPC server and restore sessions as if
+  // everything converged — that is exactly the "poller hangs running / orphan
+  // re-attach" this feature exists to prevent. Abort this bot's startup so an
+  // operator/supervisor sees it, rather than fail-open into an inconsistent state.
+  let idempotencyQuarantinedSessionIds: Set<string>;
   try {
     idempotencyQuarantinedSessionIds = await reconcileIdempotencyLeasesOnBoot(cfg.larkAppId, getDaemonBootId());
   } catch (err) {
-    // A failed reconcile means an ambiguous turn might still poll `running` — do
-    // not proceed as if converged; surface loudly. (recordFailedStrict throwing
-    // is the main way this happens.)
-    logger.error(`[idempotency] boot reconcile failed — some leases may be unconverged: ${err instanceof Error ? err.message : err}`);
+    logger.error(`[idempotency] boot reconcile failed to converge — aborting bot startup (fail-closed): ${err instanceof Error ? err.message : err}`);
+    throw err instanceof Error ? err : new Error(String(err));
   }
   // Seed dashboard IPC botName with the custom displayName (falling back to the
   // bot's config id); the friendly name from /bot/v3/info is wired into the

src/services/async-trigger-store.ts

@@ -85,6 +85,26 @@ function load(sessionId: string): AsyncTriggerFile {
   }
 }
 
+/** STRICT loader for the authoritative failed-evidence RMW: ONLY a genuinely
+ *  absent file (ENOENT) is treated as empty. A present-but-unreadable file
+ *  (EIO/EACCES), corrupt JSON, or invalid shape THROWS — the soft `load()` would
+ *  fold these into `{results:{}}`, and recordFailedStrict would then durably
+ *  OVERWRITE a file that might hold a `completed` proof or another owner's data
+ *  (finding: strict write over a soft read defeats completed-wins/owner-proof). */
+function loadStrict(sessionId: string): AsyncTriggerFile {
+  const fp = getFilePath(sessionId);
+  try { readFileSync(fp, 'utf-8'); }
+  catch (err: any) {
+    if (err?.code === 'ENOENT') return { results: {} };
+    throw err; // EIO/EACCES/… — do NOT treat as empty
+  }
+  const data = JSON.parse(readFileSync(fp, 'utf-8')) as AsyncTriggerFile; // corrupt → throw
+  if (!data || typeof data !== 'object' || typeof data.results !== 'object') {
+    throw new Error(`corrupt async-trigger file (invalid shape): ${fp}`);
+  }
+  return { ownerLarkAppId: data.ownerLarkAppId, latestTriggerId: data.latestTriggerId, results: data.results ?? {} };
+}
+
 function save(sessionId: string, file: AsyncTriggerFile): void {
   ensureDir();
   const fp = getFilePath(sessionId);
@@ -178,7 +198,12 @@ export function recordFailedStrict(
   if (!ownerLarkAppId) throw new Error('recordFailedStrict requires ownerLarkAppId');
   ensureDir();
   withFileLockSync(getFilePath(sessionId), () => {
-    const file = load(sessionId);
+    const file = loadStrict(sessionId); // ONLY ENOENT is empty; corrupt/EIO throws
+    // Owner proof: never overwrite another bot's file (a hash/path mixup or a
+    // cross-bot mistake must fail-closed, not clobber their evidence).
+    if (file.ownerLarkAppId && file.ownerLarkAppId !== ownerLarkAppId) {
+      throw new Error(`recordFailedStrict owner mismatch: file owned by ${file.ownerLarkAppId}, caller ${ownerLarkAppId}`);
+    }
     const prev = file.results[triggerId];
     if (prev?.status === 'completed') return; // completed is stronger — keep it
     file.ownerLarkAppId = ownerLarkAppId;

src/services/idempotency-store.ts

@@ -230,9 +230,21 @@ export function transition(
   });
 }
 
+/** Delete a file, treating ONLY ENOENT as "already gone". EIO/EROFS/EACCES etc.
+ *  mean the file may still exist — the caller must NOT proceed as if released,
+ *  so we throw (finding: a swallowed unlink error left a reserved lease stuck to
+ *  a closed session). */
+function strictUnlink(fp: string): void {
+  try { unlinkSync(fp); }
+  catch (err: any) { if (err?.code !== 'ENOENT') throw err; }
+}
+
 /** Compare-and-remove: delete the lease ONLY if it still matches `expect`
  *  (identity + revision + state) under the lock. Used to release a `reserved`
- *  lease we created but abandoned before dispatch. Returns true if removed. */
+ *  lease we created but abandoned before dispatch. Returns true if removed,
+ *  false if the on-disk record changed (someone advanced it) or was already
+ *  gone. THROWS on an ambiguous unlink error (EIO/EROFS/…): the caller must be
+ *  able to trust that `true` means the lease is truly released. */
 export function compareAndRemove(ownerLarkAppId: string, key: string, expect: IdempotencyRecord): boolean {
   const fp = fileFor(ownerLarkAppId, key);
   return withKeyLock(fp, () => {
@@ -241,14 +253,17 @@ export function compareAndRemove(ownerLarkAppId: string, key: string, expect: Id
     if (current.revision !== expect.revision || current.state !== expect.state || !sameIdentity(current, expect)) {
       return false;
     }
-    try { unlinkSync(fp); } catch { /* already gone */ }
+    strictUnlink(fp); // throws on EIO/EROFS/… (never a silent success)
     return true;
   });
 }
 
-/** Enumerate every stored lease (boot reconcile). Best-effort per file: a
- *  corrupt file is logged + skipped so it can't abort the sweep. */
-export function listAll(): Array<{ file: string; record: IdempotencyRecord }> {
+/** Enumerate every stored lease (boot reconcile). By default a corrupt file is
+ *  logged + skipped. With `throwOnCorrupt`, a corrupt lease THROWS instead — the
+ *  reconcile can't prove such a lease converged, so it must fail-closed rather
+ *  than silently skip (a skipped corrupt lease could hide an unconverged
+ *  attempting fence). */
+export function listAll(opts: { throwOnCorrupt?: boolean } = {}): Array<{ file: string; record: IdempotencyRecord }> {
   const dir = getDir();
   if (!existsSync(dir)) return [];
   const out: Array<{ file: string; record: IdempotencyRecord }> = [];
@@ -259,16 +274,30 @@ export function listAll(): Array<{ file: string; record: IdempotencyRecord }> {
       const rec = readRecord(fp);
       if (rec) out.push({ file: fp, record: rec });
     } catch (err) {
+      if (opts.throwOnCorrupt) throw new Error(`unreadable idempotency lease ${fp}: ${(err as Error).message}`);
       logger.warn(`[idempotency] skipping unreadable lease ${fp}: ${err}`);
     }
   }
   return out;
 }
 
-/** Reconcile-only remove by path, under a lock keyed on that path. Used by boot
- *  reconcile to drop a pre-dispatch `reserved` lease. Best-effort. */
-export function removeByPathLocked(fp: string): void {
-  withKeyLock(fp, () => {
-    try { if (existsSync(fp)) unlinkSync(fp); } catch { /* ignore */ }
+/** Reconcile-only compare-and-remove BY PATH (reconcile enumerated the file via
+ *  listAll and holds a snapshot record; the plaintext key isn't recoverable from
+ *  the hashed filename). Re-reads under the lock and removes ONLY if the on-disk
+ *  record still matches the snapshot's full identity + revision + state — so a
+ *  stale reserved snapshot can NOT delete a fence that has since advanced to
+ *  `attempting` (finding: old sweep erasing a crossed commit-unknown barrier).
+ *  Returns true iff removed. THROWS on an ambiguous unlink error. */
+export function compareAndRemoveByPath(fp: string, expect: IdempotencyRecord): boolean {
+  return withKeyLock(fp, () => {
+    let current: IdempotencyRecord | undefined;
+    try { current = readRecord(fp); }
+    catch { return false; } // corrupt now → leave it for a human, never blind-delete
+    if (!current) return false;
+    if (current.revision !== expect.revision || current.state !== expect.state || !sameIdentity(current, expect)) {
+      return false; // advanced/changed under us — keep it
+    }
+    strictUnlink(fp);
+    return true;
   });
 }

test/async-trigger-store.test.ts

@@ -30,6 +30,7 @@ vi.mock('../src/utils/logger.js', () => ({
 import {
   recordPending,
   recordCompleted,
+  recordFailedStrict,
   lookup,
   deleteResults,
 } from '../src/services/async-trigger-store.js';
@@ -180,3 +181,46 @@ describe('robustness', () => {
     expect(lookup(sid)?.result.content).toBe('ok');
   });
 });
+
+describe('recordFailedStrict (authoritative dispatch_unknown terminal)', () => {
+  it('writes a durable failed(dispatch_unknown) that lookup surfaces', () => {
+    recordFailedStrict('sessF', 'trg_f', 7000, 'cli_test', 'dispatch_unknown');
+    const r = lookup('sessF', 'trg_f')?.result;
+    expect(r?.status).toBe('failed');
+    expect(r?.errorCode).toBe('no_output');
+    expect(r?.reason).toBe('dispatch_unknown');
+  });
+
+  it('COMPLETED WINS: does not overwrite an existing completed result', () => {
+    recordCompleted('sessC', 'trg_c', 'the answer', 5000, 'cli_test');
+    recordFailedStrict('sessC', 'trg_c', 6000, 'cli_test'); // must be a no-op
+    expect(lookup('sessC', 'trg_c')?.result.status).toBe('completed');
+    expect(lookup('sessC', 'trg_c')?.result.content).toBe('the answer');
+  });
+
+  it('LATE COMPLETED WINS: a completed arriving after failed overwrites it', () => {
+    recordFailedStrict('sessL', 'trg_l', 6000, 'cli_test');
+    expect(lookup('sessL', 'trg_l')?.result.status).toBe('failed');
+    recordCompleted('sessL', 'trg_l', 'done late', 7000, 'cli_test');
+    expect(lookup('sessL', 'trg_l')?.result.status).toBe('completed');
+  });
+
+  it('STRICT READ: throws on a corrupt existing file (never overwrites it as empty)', () => {
+    mkdirSync(join(tempDir, 'async-triggers'), { recursive: true });
+    writeFileSync(join(tempDir, 'async-triggers', 'sessCorrupt.json'), '{ not json', 'utf-8');
+    expect(() => recordFailedStrict('sessCorrupt', 'trg_x', 8000, 'cli_test')).toThrow();
+    // The corrupt file is left intact for a human — not silently replaced.
+    expect(existsSync(join(tempDir, 'async-triggers', 'sessCorrupt.json'))).toBe(true);
+  });
+
+  it('OWNER PROOF: refuses to overwrite a file owned by a different bot', () => {
+    recordCompleted('sessO', 'trg_o', 'x', 5000, 'cli_ownerA');
+    expect(() => recordFailedStrict('sessO', 'trg_o', 6000, 'cli_ownerB')).toThrow(/owner mismatch/);
+    // ownerA's data intact.
+    expect(lookup('sessO', 'trg_o')?.result.status).toBe('completed');
+  });
+
+  it('requires ownerLarkAppId', () => {
+    expect(() => recordFailedStrict('sessN', 'trg_n', 1, '')).toThrow(/ownerLarkAppId/);
+  });
+});

test/idempotency-store.test.ts

@@ -23,7 +23,7 @@ vi.mock('../src/utils/logger.js', () => ({
 }));
 
 import {
-  claim, transition, takeover, lookup, compareAndRemove, listAll, removeByPathLocked,
+  claim, transition, takeover, lookup, compareAndRemove, listAll, compareAndRemoveByPath,
   IdempotencyConflictError,
   type IdempotencyRecord,
 } from '../src/services/idempotency-store.js';
@@ -154,12 +154,17 @@ describe('reconcile enumeration', () => {
     expect(all.every(a => a.file.endsWith('.json'))).toBe(true);
   });
 
-  it('removeByPathLocked drops a lease by its enumerated path', () => {
+  it('compareAndRemoveByPath drops a lease only if the on-disk record still matches the snapshot', () => {
     claim(base());
-    const { file } = listAll()[0];
-    removeByPathLocked(file);
+    const { file, record } = listAll()[0];
+    // Stale snapshot (advanced to attempting under us) → must NOT delete the fence.
+    transition('cli_a', 'k1', record, { state: 'attempting', now: 5000 }); // rev2 attempting
+    expect(compareAndRemoveByPath(file, record)).toBe(false); // record is the rev1 reserved snapshot
+    expect(lookup('cli_a', 'k1')?.state).toBe('attempting'); // fence preserved (codex repro)
+    // Exact match → removed.
+    const current = lookup('cli_a', 'k1')!;
+    expect(compareAndRemoveByPath(file, current)).toBe(true);
     expect(lookup('cli_a', 'k1')).toBeUndefined();
-    expect(() => removeByPathLocked(file)).not.toThrow(); // idempotent
   });
 
   it('listAll skips (does not throw on) a corrupt file', () => {