Keep a second credential error from failing the session

A rebuilt agent reused seq 0 for the next same-category inference
error, so commitErrors threw Duplicate error record and
afterCheckpoint failed the run. Resume the durable error sequence
on assembly and drop a colliding flush instead of failing the
session.

Author: TheGreatAxios

src/session/assemble-runtime.test.ts

@@ -143,6 +143,7 @@ function stubAuditStore(): AuditStore {
     commitAudit: async () => undefined,
     commitErrors: async () => undefined,
     loadAudit: async () => [],
+    loadErrors: async () => [],
   };
 }
 

src/session/optimized-context-store.test.ts

@@ -800,6 +800,7 @@ describe("createSessionStores", () => {
     expect(typeof audit.commitAudit).toBe("function");
     expect(typeof audit.commitErrors).toBe("function");
     expect(typeof audit.loadAudit).toBe("function");
+    expect(typeof audit.loadErrors).toBe("function");
   });
 });
 

src/session/optimized-context-store.ts

@@ -848,6 +848,7 @@ export async function createSessionStores(
     commitErrors: (records, signal) =>
       withResolvedDirLock(dir, () => base.commitErrors(records, signal)),
     loadAudit: (sessionId, signal) => base.loadAudit(sessionId, signal),
+    loadErrors: (sessionId, signal) => base.loadErrors(sessionId, signal),
   };
 
   return { storage: store, audit: store };

vendor/intx-agent/src/agent.ts

@@ -512,7 +512,12 @@ export async function createAgent<EnvReq extends BaseEnv>(
     // the next flush, so a fourth caller arriving after the follow-up
     // begins still observes a clean state and starts its own flush.
     const accumulatedErrors: ErrorRecord[] = [];
+    // Resume from durable records so a rebuilt agent does not reuse seq 0
+    // and collide with files the previous assembly already committed.
     let errorSeq = 0;
+    for (const record of await auditStore.loadErrors(sessionId)) {
+      if (record.seq >= errorSeq) errorSeq = record.seq + 1;
+    }
     let flushInProgress: Promise<void> | undefined;
     let pendingFollowUp: Promise<void> | undefined;
 
@@ -551,6 +556,16 @@ export async function createAgent<EnvReq extends BaseEnv>(
         try {
           await auditStore.commitErrors(batch);
           accumulatedErrors.splice(0, count);
+        } catch (cause) {
+          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);
+            return;
+          }
+          throw cause;
         } finally {
           flushInProgress = undefined;
         }

vendor/intx-agent/src/audit-integration.test.ts

@@ -71,6 +71,9 @@ function makeRecordingAuditStore(): RecordingAuditStore {
     async loadAudit(_sessionId: string): Promise<AuditRecord[]> {
       return committedAudit.flat();
     },
+    async loadErrors(_sessionId: string): Promise<ErrorRecord[]> {
+      return committedErrors.flat();
+    },
     getCommittedAudit() {
       return committedAudit;
     },

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

@@ -14,6 +14,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { type } from "arktype";
 
+import { createDefaultDependencies } from "@intx/inference/providers";
 import { createInboundMessage } from "@intx/mime";
 import { createIsogitStore } from "@intx/storage-isogit/node";
 import type { AuditRecord, ErrorRecord } from "@intx/types/audit";
@@ -59,6 +60,9 @@ function makeRecordingAuditStore(): RecordingAuditStore {
     async loadAudit(_sessionId: string): Promise<AuditRecord[]> {
       return [];
     },
+    async loadErrors(_sessionId: string): Promise<ErrorRecord[]> {
+      return committedErrors.flat();
+    },
     getCommittedErrors() {
       return committedErrors;
     },
@@ -91,12 +95,37 @@ function makeFailFirstAuditStore(): FailingAuditStore {
     async loadAudit(_sessionId: string): Promise<AuditRecord[]> {
       return [];
     },
+    async loadErrors(_sessionId: string): Promise<ErrorRecord[]> {
+      return committedErrors.flat();
+    },
     getCommittedErrors() {
       return committedErrors;
     },
   };
 }
 
+function makeDuplicateErrorAuditStore(): FailingAuditStore {
+  return {
+    async commitAudit(_records: AuditRecord[]): Promise<void> {
+      // No-op.
+    },
+    async commitErrors(records: ErrorRecord[]): Promise<void> {
+      throw new Error(
+        `Duplicate error record: ${records[0]?.sessionId ?? "session"}/00000000-credential_failure`,
+      );
+    },
+    async loadAudit(_sessionId: string): Promise<AuditRecord[]> {
+      return [];
+    },
+    async loadErrors(_sessionId: string): Promise<ErrorRecord[]> {
+      return [];
+    },
+    getCommittedErrors() {
+      return [];
+    },
+  };
+}
+
 // 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
@@ -153,6 +182,86 @@ async function waitForReactorDone(
   }
 }
 
+const FORBIDDEN_DEPS = {
+  ...createDefaultDependencies(),
+  fetch: async () =>
+    new Response("Unauthorized", { status: 401, statusText: "Unauthorized" }),
+};
+
+function credentialFailureDirectors(): BaseEnv["directors"] {
+  return makeDirectorRegistry(
+    async (
+      event: ReactorInboundEvent,
+      _state: ReactorState,
+      caps: ReactorCapabilities,
+    ) => {
+      if (event.type === "message.received") return caps.infer();
+      if (event.type === "inference.error") {
+        return [caps.checkpoint("after-error"), caps.done()];
+      }
+      return caps.done();
+    },
+  );
+}
+
+function forbiddenAgentDef(id: string) {
+  return defineAgent({
+    id,
+    systemPrompt: "test",
+    tools: [],
+    capabilities: [],
+    inference: {
+      sources: [
+        {
+          provider: UNREACHABLE_SOURCE.provider,
+          model: UNREACHABLE_SOURCE.model,
+        },
+      ],
+    },
+  });
+}
+
+function duplicateFlushFailures(
+  events: ReadonlyArray<{ type: string; data?: unknown }>,
+): ReadonlyArray<{ type: string; data?: unknown }> {
+  return events.filter((event) => {
+    if (event.type !== "reactor.error") return false;
+    return JSON.stringify(event.data ?? {}).includes("Duplicate error record");
+  });
+}
+
+async function runForbiddenCycle(opts: {
+  workdir: string;
+  sessionId: string;
+  agentId: string;
+}): Promise<{ events: Array<{ type: string; data?: unknown }> }> {
+  const store = await createIsogitStore(opts.workdir);
+  const env: BaseEnv = {
+    sources: [UNREACHABLE_SOURCE],
+    defaultSource: UNREACHABLE_SOURCE.id,
+    storage: store,
+    workdir: opts.workdir,
+    audit: store,
+    authorize: permissiveAuthorize(),
+    directors: credentialFailureDirectors(),
+    sessionId: opts.sessionId,
+    deps: FORBIDDEN_DEPS,
+  };
+  const agent = await createAgent(forbiddenAgentDef(opts.agentId), env);
+  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();
+  }
+  return { events };
+}
+
 describe("agent error flushing", () => {
   let workDir: string;
 
@@ -470,4 +579,101 @@ describe("agent error flushing", () => {
     expect(batches.length).toBe(1);
     expect(batches[0]?.[0]?.source).toBe("reactor");
   });
+
+  test("two credential_failure errors in one session persist without failing the run", async () => {
+    const sessionId = "session-credential-once";
+    let inferenceErrors = 0;
+    const store = await createIsogitStore(workDir);
+    const env: BaseEnv = {
+      sources: [UNREACHABLE_SOURCE],
+      defaultSource: UNREACHABLE_SOURCE.id,
+      storage: store,
+      workdir: workDir,
+      audit: store,
+      authorize: permissiveAuthorize(),
+      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.checkpoint("after-first"), caps.infer()];
+            }
+            return [caps.checkpoint("after-second"), caps.done()];
+          }
+          return caps.done();
+        },
+      ),
+      sessionId,
+      deps: FORBIDDEN_DEPS,
+    };
+    const agent = await createAgent(forbiddenAgentDef("cred-flush-once"), env);
+    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([]);
+    const records = (await store.loadErrors(sessionId)).filter(
+      (record) => record.category === "credential_failure",
+    );
+    expect(records).toHaveLength(2);
+    expect(new Set(records.map((record) => record.seq)).size).toBe(2);
+  });
+
+  test("two credential_failure errors persist across re-assembly without failing the session", async () => {
+    const sessionId = "session-credential";
+    const first = await runForbiddenCycle({
+      workdir: workDir,
+      sessionId,
+      agentId: "cred-flush-1",
+    });
+    const second = await runForbiddenCycle({
+      workdir: workDir,
+      sessionId,
+      agentId: "cred-flush-2",
+    });
+
+    expect(duplicateFlushFailures(first.events)).toEqual([]);
+    expect(duplicateFlushFailures(second.events)).toEqual([]);
+    const store = await createIsogitStore(workDir);
+    const records = (await store.loadErrors(sessionId)).filter(
+      (record) => record.category === "credential_failure",
+    );
+    expect(records).toHaveLength(2);
+    expect(new Set(records.map((record) => record.seq)).size).toBe(2);
+  });
+
+  test("a duplicate error record from commitErrors does not fail the session", async () => {
+    const audit = makeDuplicateErrorAuditStore();
+    const directors = credentialFailureDirectors();
+    const def = forbiddenAgentDef("cred-flush-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);
+  });
 });

vendor/intx-agent/src/testing/audit-noop.test.ts

@@ -18,6 +18,11 @@ describe("noopAuditStore", () => {
     expect(await store.loadAudit("sess")).toEqual([]);
   });
 
+  test("loadErrors returns an empty array", async () => {
+    const store = noopAuditStore();
+    expect(await store.loadErrors("sess")).toEqual([]);
+  });
+
   test("each call returns a fresh object", () => {
     expect(noopAuditStore()).not.toBe(noopAuditStore());
   });

vendor/intx-agent/src/testing/audit-noop.ts

@@ -25,5 +25,8 @@ export function noopAuditStore(): AuditStore {
     async loadAudit(_sessionId: string): Promise<AuditRecord[]> {
       return [];
     },
+    async loadErrors(_sessionId: string): Promise<ErrorRecord[]> {
+      return [];
+    },
   };
 }

vendor/intx-inference/src/assembly.test.ts

@@ -198,6 +198,9 @@ function makeRecordingAuditStore(): AuditStore & {
     async commitErrors() {
       /* noop */
     },
+    async loadErrors() {
+      return [];
+    },
     getCommitted() {
       return committed;
     },

vendor/intx-storage-isogit/src/store.test.ts

@@ -692,6 +692,38 @@ describe("error store", () => {
       ),
     ).toHaveLength(1);
   });
+
+  test("loadErrors round-trips records ordered by seq", async () => {
+    const dir = await tempDir();
+    const store = await createAuditStore(dir);
+    const later = makeErrorRecord({ seq: 2, category: "retryable" });
+    const earlier = makeErrorRecord({ seq: 1, category: "credential_failure" });
+
+    await store.commitErrors([later]);
+    await store.commitErrors([earlier]);
+
+    expect(await store.loadErrors("session-1")).toEqual([earlier, later]);
+  });
+
+  test("loadErrors returns empty array for nonexistent session", async () => {
+    const dir = await tempDir();
+    const store = await createAuditStore(dir);
+
+    expect(await store.loadErrors("no-such-session")).toEqual([]);
+  });
+
+  test("rejects sessionId with path traversal on loadErrors", async () => {
+    const dir = await tempDir();
+    const store = await createAuditStore(dir);
+
+    let thrown: Error | undefined;
+    try {
+      await store.loadErrors("../escape");
+    } catch (cause) {
+      thrown = cause instanceof Error ? cause : new Error(String(cause));
+    }
+    expect(thrown?.message).toContain("unsafe characters");
+  });
 });
 
 describe("audit and error durability retries", () => {