fix: 保留取证顺序并防止异常丢失已验证工作

lib/app/ports.js

@@ -24,12 +24,12 @@
  *                                     epoch and turn counter; calls without one keep the legacy session namespace.
  * @property {(sessionId: string, note: object, scope?: {turn: number, sessionEpoch?: number}) => Promise<void>} append
  * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<object[]>} read
- * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ notes: object[], commit: () => Promise<void>, release: () => void }>} take
+ * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ notes: object[], ordered?: boolean, commit: () => Promise<void>, release: () => void }>} take
  * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<void>} clear
  *
  * @typedef {object} EditBaseline       Each file's content hash before the turn first edited it.
  * @property {(sessionId: string, file: string, turnId?: string|null, scope?: {turn: number, sessionEpoch?: number}) => Promise<void>} remember
- * @property {(sessionId: string, turnId?: string|null, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ before: Map<string, string>, commit: () => Promise<void> }>} take
+ * @property {(sessionId: string, turnId?: string|null, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ before: Map<string, string>, ordered?: boolean, commit: () => Promise<void>, release?: () => void }>} take
  *
  * @typedef {object} TableStore         One table keyed by session id (each session's range of work turns to distill)
  *                                      or a list (drafts, the last distillation outcomes, the calibration corpus).

lib/app/turn-end.js

@@ -22,7 +22,7 @@ const SETTLING_WAIT_MS = 2_000;
 const sameTurn = (session, expected) => session.turnId === expected.turnId && session.turn === expected.turn && session.sessionEpoch === expected.sessionEpoch;
 
 // A turn end that began later (a re-judge of the same turn) owns `settling` from then on; an earlier one leaves it.
-// A native turn whose evidence could not be read reopens, so the host's next turn end or recovery delivers it again.
+// A native turn whose evidence or bookkeeping could not finish reopens for the next turn end or recovery.
 const settleTurn = (ports, sessionId, opened, reopen = false) => ports.sessions.update(sessionId, (stored) => {
   const current = sessionOf(stored);
   if (!sameTurn(current, opened) || current.settling !== opened.settling) return undefined;
@@ -73,7 +73,7 @@ async function reportTurn({ identity, sessions, reports, clock, confirmTurnCompl
     if (current.sessionEpoch !== opened.sessionEpoch) return undefined;
     const session = markedReported(current, [...due, ...retries.map(({ asset }) => asset)], clock.now()) ?? current;
     if (!sameTurn(session, opened)) return withoutPendingControls(session, controls);
-    const labelable = opened.isCommand ? session.labelable : queued.map(({ report }) => reportedReuseOf(report));
+    const labelable = opened.isCommand || queued.length === 0 ? session.labelable : queued.map(({ report }) => reportedReuseOf(report));
     return { ...withoutPendingControls(session, controls), labelable };
   });
   return result;
@@ -182,7 +182,7 @@ async function readTurnEvidence(ports, sessionId, opened, recordOf) {
   try {
     return { observed, notes, edits: await ports.baseline.take(sessionId, opened.turnId, opened) };
   } catch (error) {
-    notes.release();
+    try { await notes.release(); } catch { /* Preserve the read failure if returning the notes is refused too. */ }
     throw error;
   }
 }
@@ -199,14 +199,19 @@ async function recordTurn(ports, sessionId, opened, { reason, answer, model, pro
   const ofThisTurn = (entry) => !entry.turnId || !opened.turnId || entry.turnId === opened.turnId;
   const record = observed ?? turnRecordOf(notes.notes.filter(ofThisTurn), { pairShellCalls: ports.confirmTurnCompletion });
   const finished = { ...record, answer: String(answer || record.answer || '').trim() };
-  const verified = !ports.confirmTurnCompletion || (reason === 'answer' && finished.answer
-    && finished.checks.length > 0 && finished.checks.every(check => verdictOf(check) !== 'unknown'));
-  const { queued, controls, strategies, failure } = await reportTurn(ports, sessionId, opened, { reason, checks: finished.checks, source, verified });
+  const verified = (observed || notes.ordered !== false) && (!ports.confirmTurnCompletion || (reason === 'answer' && finished.answer
+    && finished.checks.length > 0 && finished.checks.every(check => verdictOf(check) !== 'unknown')));
+  const { queued, controls, strategies, failure: reportFailure } = await reportTurn(ports, sessionId, opened, { reason, checks: finished.checks, source, verified });
   if ((await ports.sessions.read(sessionId)).sessionEpoch !== opened.sessionEpoch) return null;
   const reuses = queued.map(({ report }) => reportedReuseOf(report));
-  if (verified) await ports.retention.remember({ sessionId, projectDir, reuses, editedBefore: edits.before });
-  await notes.commit();
-  await edits.commit();
+  let failure = reportFailure;
+  // A prior attempt with cached reports may already have appended retention before its bookkeeping failed.
+  const retentionMayHaveRun = Object.values(opened.injected ?? {}).some(entry => entry?.turn === opened.turn && entry?.pendingReport);
+  if (verified && edits.ordered !== false && !retentionMayHaveRun) {
+    try { await ports.retention.remember({ sessionId, projectDir, reuses, editedBefore: edits.before }); }
+    // An append can fail after writing: surface it after preserving the verified work, never retry it blindly.
+    catch (error) { failure ??= error; }
+  }
   const isWork = verified && await rememberWorkTurn(ports, sessionId, opened, { record: finished, model, projectDir, source });
   if (!isWork) await ports.candidates.update((pending) => rangeInEpoch(pending, sessionId, opened.sessionEpoch)
     ? withSessionActivity(pending, sessionId, ports.clock.now()) : undefined);
@@ -222,18 +227,25 @@ export async function turnEnded(ports, { sessionId, turnId, reason, answer = '',
     await scheduleDueDistillation(ports);
     return null;
   }
-  let ended, evidenceRead = false;
+  let ended, evidence, processingComplete = false, cleanupFailure;
   try {
-    const evidence = await readTurnEvidence(ports, sessionId, opened, recordOf);
-    evidenceRead = true;
+    evidence = await readTurnEvidence(ports, sessionId, opened, recordOf);
     if (evidence) ended = await recordTurn(ports, sessionId, opened, { reason, answer, model, projectDir, source }, evidence);
+    processingComplete = true;
+    if (evidence) { await evidence.notes.commit(); await evidence.edits.commit(); }
+  } catch (error) {
+    for (const lease of [evidence?.notes, evidence?.edits]) {
+      try { await lease?.release?.(); } catch { /* Keep the original failure; stale recovery can reclaim refused releases. */ }
+    }
+    if (!processingComplete) throw error;
+    cleanupFailure = error;
   } finally {
-    await settleTurn(ports, sessionId, opened, ports.confirmTurnCompletion && !evidenceRead);
+    await settleTurn(ports, sessionId, opened, ports.confirmTurnCompletion && !processingComplete);
   }
-  if (!ended) return null;
+  if (!ended) { if (cleanupFailure) throw cleanupFailure; return null; }
   const { controls, failure, ...result } = ended;
   await scheduleDueDistillation(ports, result.reports.length > 0 || controls > 0 ? [TASKS.DRAIN] : []);
-  if (failure) throw failure;
+  if (failure || cleanupFailure) throw failure ?? cleanupFailure;
   return result;
 }
 

lib/runtime/store/file-lock.js

@@ -187,26 +187,25 @@ function linesOf(file) {
   });
 }
 
-function staleClaimsOf(target, now) {
+function claimsOf(target, now) {
   const prefix = `${path.basename(target)}.`;
   try {
     return fs.readdirSync(path.dirname(target))
       .filter((name) => name.startsWith(prefix) && name.endsWith('.taking'))
       .map((name) => path.join(path.dirname(target), name))
-      .filter((file) => now - fs.statSync(file).mtimeMs > CLAIM_STALE_MS);
-  } catch {
-    return [];
-  }
-}
-
-function claimed(source, target) {
-  const claim = `${target}.${process.pid}.${crypto.randomBytes(6).toString('hex')}.taking`;
-  try {
-    fs.renameSync(source, claim);
-    fs.utimesSync(claim, new Date(), new Date());
-    return claim;
-  } catch {
-    return null;
+      .flatMap((file) => {
+        try {
+          const marker = /\.order-(\d+|unknown)\.\d+\.[a-f0-9]+\.taking$/.exec(file)?.[1];
+          return [{ file, order: marker && marker !== 'unknown' ? BigInt(marker) : null,
+            unknown: marker === 'unknown', stale: now - fs.statSync(file).mtimeMs > CLAIM_STALE_MS }];
+        } catch (error) {
+          if (error?.code === 'ENOENT') return [];
+          throw error;
+        }
+      });
+  } catch (error) {
+    if (error?.code === 'ENOENT') return [];
+    throw error;
   }
 }
 
@@ -221,8 +220,9 @@ export function readStateLines(target) {
   }
 }
 
-export function takeStateLines(target, now = Date.now()) {
-  const claims = [...staleClaimsOf(target, now), target].map((source) => claimed(source, target)).filter(Boolean);
+export function takeStateLines(target, now = Date.now(), { requireComplete = false } = {}) {
+  const claims = [];
+  let ordered = true;
   // A released claim is dated just past stale, so the next take recovers it at once while pruning, which expires
   // evidence days old, keeps it.
   const release = () => {
@@ -235,7 +235,39 @@ export function takeStateLines(target, now = Date.now()) {
     for (const claim of claims) fs.rmSync(claim, { force: true });
   };
   try {
-    return { lines: claims.flatMap(linesOf), commit, release };
+    const rotated = admitted(`${target}.claims`, () => {
+      const earlier = claimsOf(target, now);
+      if (requireComplete && earlier.some(({ stale }) => !stale)) {
+        throw Object.assign(new Error('Turn evidence is still claimed; retry after its reader finishes.'), { code: 'ELOCKED' });
+      }
+      // One legacy chunk has a known position before the current append file; several have lost their order.
+      if (earlier.length === 1 && earlier[0].order === null && !earlier[0].unknown) earlier[0].order = 0n;
+      const next = earlier.reduce((max, { order }) => order !== null && order > max ? order : max, -1n) + 1n;
+      const sources = earlier.filter(({ stale }) => stale).sort((a, b) => {
+        if (a.order === b.order) return 0;
+        if (a.order === null) return -1;
+        if (b.order === null) return 1;
+        return a.order < b.order ? -1 : 1;
+      });
+      for (const { file, order } of [...sources, { file: target, order: next }]) {
+        const claim = `${target}.order-${order ?? 'unknown'}.${process.pid}.${crypto.randomBytes(6).toString('hex')}.taking`;
+        try { fs.renameSync(file, claim); }
+        catch (error) {
+          if (requireComplete && error?.code !== 'ENOENT') throw error;
+          continue;
+        }
+        claims.push(claim);
+        ordered &&= order !== null;
+        fs.utimesSync(claim, new Date(), new Date());
+      }
+      // A running older plugin may not use this admission guard.
+      if (requireComplete && claimsOf(target, now).some(({ file }) => !claims.includes(file))) {
+        throw Object.assign(new Error('Turn evidence is still claimed; retry after its reader finishes.'), { code: 'ELOCKED' });
+      }
+      return true;
+    }, release);
+    if (!rotated) throw Object.assign(new Error('Evidence claims are busy; retry after their reader finishes.'), { code: 'ELOCKED' });
+    return { lines: claims.flatMap(linesOf), ordered, commit, release };
   } catch (error) {
     release();
     throw error;

lib/runtime/store/file-stores.js

@@ -53,8 +53,8 @@ function linesStore(target) {
     append: async (value) => appendStateLine(target, value),
     read: async () => readStateLines(target),
     take: async () => {
-      const { lines, commit, release } = takeStateLines(target);
-      return { lines, commit: async () => commit(), release };
+      const { lines, ordered, commit, release } = takeStateLines(target, Date.now(), { requireComplete: true });
+      return { lines, ordered, commit: async () => commit(), release };
     },
     clear: async () => removeState(target),
   };
@@ -101,7 +101,7 @@ export function fileStoresOf({ dir, agent }) {
   const evidenceFile = (kind, sessionId, scope) => scope
     ? sessionFile(`${kind}-generation`, JSON.stringify([sessionId, scope.sessionEpoch, scope.turn]), 'jsonl')
     : sessionFile(kind, sessionId, 'jsonl');
-  const evidencePattern = (kind) => new RegExp(`^${agent}-${kind}(?:-generation)?-[0-9a-f]{16}\\.jsonl(?:\\.\\d+\\.[0-9a-f]{12}\\.taking)?$`);
+  const evidencePattern = (kind) => new RegExp(`^${agent}-${kind}(?:-generation)?-[0-9a-f]{16}\\.jsonl(?:(?:\\.order-(?:\\d+|unknown))?\\.\\d+\\.[0-9a-f]{12}\\.taking)?$`);
   const filesOf = (pattern) => {
     try {
       return fs.readdirSync(dir).filter((name) => pattern.test(name)).map((name) => path.join(dir, name));
@@ -154,20 +154,20 @@ export function fileStoresOf({ dir, agent }) {
     append: (sessionId, note, scope) => linesStore(evidenceFile('turn', sessionId, scope)).append(note),
     read: (sessionId, scope) => linesStore(evidenceFile('turn', sessionId, scope)).read(),
     take: async (sessionId, scope) => {
-      const { lines, commit, release } = await linesStore(evidenceFile('turn', sessionId, scope)).take();
-      return { notes: lines, commit, release };
+      const { lines, ordered, commit, release } = await linesStore(evidenceFile('turn', sessionId, scope)).take();
+      return { notes: lines, ordered, commit, release };
     },
     clear: (sessionId, scope) => linesStore(evidenceFile('turn', sessionId, scope)).clear(),
   };
 
   const baseline = {
     remember: (sessionId, target, turnId = null, scope) => linesStore(evidenceFile('edits', sessionId, scope)).append({ path: target, before: contentHashOf(target), turnId }),
     take: async (sessionId, turnId = null, scope) => {
-      const { lines, commit } = await linesStore(evidenceFile('edits', sessionId, scope)).take();
+      const { lines, ordered, commit, release } = await linesStore(evidenceFile('edits', sessionId, scope)).take();
       const before = new Map();
       const ofTurn = (line) => !line.turnId || !turnId || line.turnId === turnId;
-      for (const { path: target, before: hash } of lines.filter(ofTurn)) if (typeof target === 'string' && !before.has(target)) before.set(target, hash);
-      return { before, commit };
+      if (ordered) for (const { path: target, before: hash } of lines.filter(ofTurn)) if (typeof target === 'string' && !before.has(target)) before.set(target, hash);
+      return { before, ordered, commit, release };
     },
   };