Drop only the colliding error record on a duplicate flush

A stale-seq assembly flushing [seq0/dup, seq1/fresh] discarded the whole batch, losing fresh diagnostics. The flush now maps the colliding key back to its record, drops only that record, and retries the rest in the same flush.

Author: TheGreatAxios

vendor/intx-agent/PATCHES.md

@@ -23,8 +23,12 @@ entry and its marker.
 ## agent-ts-duplicate-error-flush
 
 `agent.ts` — `flushErrors` treats `Duplicate error record:` from
-`commitErrors` as already-durable and drops the colliding batch instead
-of failing `afterCheckpoint`.
+`commitErrors` as already-durable instead of failing `afterCheckpoint`.
+Only the record named by the colliding `<sessionId>/<seq>-<category>`
+key is dropped; the rest of the batch is retried in the same flush, so
+a stale-seq assembly flushing `[seq0/dup, seq1/fresh]` still persists
+the fresh record. An unparseable key falls back to dropping the batch.
+Identical-bytes exact retries still commit normally upstream.
 
 **Disposition:** Promotion candidate. **Removal path:** Upstream PR with
 the same duplicate-flush handling; then drop this entry and its marker.

vendor/intx-agent/src/agent.ts

@@ -526,6 +526,24 @@ export async function createAgent<EnvReq extends BaseEnv>(
     let flushInProgress: Promise<void> | undefined;
     let pendingFollowUp: Promise<void> | undefined;
 
+    // File key mirror of the isogit store's error filename scheme
+    // (`state/errors/<sessionId>/<seq>-<category>.json`, seq padded to
+    // 8, unsafe category chars replaced). Maps a
+    // `Duplicate error record: <key>` collision back to the batch member
+    // that caused it so only that record is dropped.
+    function errorFileKey(record: ErrorRecord): string {
+      const seq = String(record.seq).padStart(8, "0");
+      const category = record.category.replace(/[^a-zA-Z0-9_-]/g, "_");
+      return `${record.sessionId}/${seq}-${category}`;
+    }
+
+    function dropFromAccumulator(records: readonly ErrorRecord[]): void {
+      for (const record of records) {
+        const index = accumulatedErrors.indexOf(record);
+        if (index !== -1) accumulatedErrors.splice(index, 1);
+      }
+    }
+
     function flushErrors(): Promise<void> {
       if (flushInProgress !== undefined) {
         // If another caller already arranged a follow-up flush after
@@ -559,19 +577,39 @@ export async function createAgent<EnvReq extends BaseEnv>(
       // expectation is that commitErrors failures are transient.
       flushInProgress = (async () => {
         try {
-          await auditStore.commitErrors(batch);
-          accumulatedErrors.splice(0, count);
-        } catch (cause) {
-          // Locally patched — see vendor/intx-agent/PATCHES.md#agent-ts-duplicate-error-flush
-          if (
-            cause instanceof Error &&
-            cause.message.startsWith("Duplicate error record:")
-          ) {
-            logger.warn`duplicate error record already stored; dropping the colliding batch`;
-            accumulatedErrors.splice(0, count);
+          let remaining = batch;
+          for (;;) {
+            try {
+              await auditStore.commitErrors(remaining);
+            } catch (cause) {
+              // Locally patched — see vendor/intx-agent/PATCHES.md#agent-ts-duplicate-error-flush
+              if (
+                cause instanceof Error &&
+                cause.message.startsWith("Duplicate error record:")
+              ) {
+                const key = cause.message
+                  .slice("Duplicate error record:".length)
+                  .trim();
+                const index = remaining.findIndex(
+                  (record) => errorFileKey(record) === key,
+                );
+                if (index === -1) {
+                  logger.warn`duplicate error record already stored; dropping the colliding batch`;
+                  dropFromAccumulator(remaining);
+                  return;
+                }
+                const [colliding] = remaining.splice(index, 1);
+                if (colliding !== undefined)
+                  dropFromAccumulator([colliding]);
+                logger.warn`duplicate error record already stored; dropping the colliding record`;
+                if (remaining.length === 0) return;
+                continue;
+              }
+              throw cause;
+            }
+            dropFromAccumulator(remaining);
             return;
           }
-          throw cause;
         } finally {
           flushInProgress = undefined;
         }

vendor/intx-agent/src/flush-errors.test.ts

@@ -126,6 +126,44 @@ function makeDuplicateErrorAuditStore(): FailingAuditStore {
   };
 }
 
+// Audit store that collides on the first batch's leading record only,
+// simulating a stale-seq assembly flushing [seq0/dup, seq1/fresh]: the
+// first `commitErrors` throws `Duplicate error record` naming the
+// colliding record's file key, and the retry succeeds.
+function makePartialDuplicateAuditStore(): FailingAuditStore {
+  const committedErrors: ErrorRecord[][] = [];
+  let firstAttempt = true;
+  return {
+    async commitAudit(_records: AuditRecord[]): Promise<void> {
+      // No-op.
+    },
+    async commitErrors(records: ErrorRecord[]): Promise<void> {
+      if (firstAttempt) {
+        firstAttempt = false;
+        const colliding = records[0];
+        const seq = String(colliding?.seq ?? 0).padStart(8, "0");
+        const category = (colliding?.category ?? "").replace(
+          /[^a-zA-Z0-9_-]/g,
+          "_",
+        );
+        throw new Error(
+          `Duplicate error record: ${colliding?.sessionId ?? "session"}/${seq}-${category}`,
+        );
+      }
+      committedErrors.push([...records]);
+    },
+    async loadAudit(_sessionId: string): Promise<AuditRecord[]> {
+      return [];
+    },
+    async loadErrors(_sessionId: string): Promise<ErrorRecord[]> {
+      return [];
+    },
+    getCommittedErrors() {
+      return committedErrors;
+    },
+  };
+}
+
 // Director factory that closes over a caller-supplied `decide` to drive
 // the reactor through targeted event shapes. The factory shape requires
 // a configSchema (arktype) and returns a ReactorDirector; this helper
@@ -677,6 +715,49 @@ describe("agent error flushing", () => {
     expect(events.some((event) => event.type === "reactor.done")).toBe(true);
   });
 
+  test("a partial duplicate collision drops only the colliding record", async () => {
+    // A stale-seq assembly flushing [seq0/dup, seq1/fresh] must persist
+    // the fresh record: the first commit names only the colliding key,
+    // so the flush drops that record and retries the rest.
+    const audit = makePartialDuplicateAuditStore();
+    let inferenceErrors = 0;
+    const directors = makeDirectorRegistry(
+      async (
+        event: ReactorInboundEvent,
+        _state: ReactorState,
+        caps: ReactorCapabilities,
+      ) => {
+        if (event.type === "message.received") return caps.infer();
+        if (event.type === "inference.error") {
+          inferenceErrors += 1;
+          if (inferenceErrors === 1) return caps.infer();
+          return [caps.checkpoint("after-second"), caps.done()];
+        }
+        return caps.done();
+      },
+    );
+    const def = forbiddenAgentDef("cred-flush-partial-duplicate");
+    const env = await buildAgentEnv({ workdir: workDir, audit, directors });
+    const agent = await createAgent(def, { ...env, deps: FORBIDDEN_DEPS });
+    const events: Array<{ type: string; data?: unknown }> = [];
+    const stream = agent.stream();
+    try {
+      agent.deliver(inboundConversation());
+      for await (const event of stream) {
+        events.push(event);
+        if (event.type === "reactor.done") break;
+      }
+    } finally {
+      await agent.close();
+    }
+
+    expect(duplicateFlushFailures(events)).toEqual([]);
+    expect(events.some((event) => event.type === "reactor.done")).toBe(true);
+    const persisted = audit.getCommittedErrors().flat();
+    expect(persisted).toHaveLength(1);
+    expect(persisted[0]?.seq).toBe(1);
+  });
+
   test("createAgent still assembles when loadErrors throws", async () => {
     const audit = makeRecordingAuditStore();
     audit.loadErrors = async () => {