fix(talk): terminalize failed mutation digests

src/talk/client-voice-mutation-digest-owner.test.ts

@@ -31,7 +31,8 @@ describe("client voice mutation digest owner", () => {
         maxRetainedIntents: 4,
         maxRetainedIdentityBytes: 64,
         maxConcurrentAttempts: 2,
-        attemptTimeoutMs: 60_000,
+        maxAttemptFailures: 3,
+        attemptAbortAfterMs: 60_000,
       },
       warn,
       attempt: async ({ voiceSessionId }) => {
@@ -91,7 +92,8 @@ describe("client voice mutation digest owner", () => {
         maxRetainedIntents: 2,
         maxRetainedIdentityBytes: 64,
         maxConcurrentAttempts: 1,
-        attemptTimeoutMs: 60_000,
+        maxAttemptFailures: 3,
+        attemptAbortAfterMs: 60_000,
       },
       warn: vi.fn(),
       attempt: async () => {
@@ -109,7 +111,7 @@ describe("client voice mutation digest owner", () => {
     await vi.waitFor(() => expect(owner.snapshot().retained).toBe(0));
   });
 
-  it("keeps an ignored-abort attempt active until its real promise settles", async () => {
+  it("keeps an ignored abort request active until its real promise settles", async () => {
     const attempts: Array<{
       completion: ReturnType<typeof deferred<boolean>>;
       signal: AbortSignal;
@@ -119,7 +121,8 @@ describe("client voice mutation digest owner", () => {
         maxRetainedIntents: 2,
         maxRetainedIdentityBytes: 64,
         maxConcurrentAttempts: 1,
-        attemptTimeoutMs: 10,
+        maxAttemptFailures: 3,
+        attemptAbortAfterMs: 10,
       },
       warn: vi.fn(),
       attempt: async ({ signal }) => {
@@ -161,7 +164,8 @@ describe("client voice mutation digest owner", () => {
         maxRetainedIntents: 2,
         maxRetainedIdentityBytes: 8,
         maxConcurrentAttempts: 1,
-        attemptTimeoutMs: 60_000,
+        maxAttemptFailures: 3,
+        attemptAbortAfterMs: 60_000,
       },
       warn,
       attempt,
@@ -190,7 +194,8 @@ describe("client voice mutation digest owner", () => {
         maxRetainedIntents: 4,
         maxRetainedIdentityBytes: 10,
         maxConcurrentAttempts: 1,
-        attemptTimeoutMs: 60_000,
+        maxAttemptFailures: 3,
+        attemptAbortAfterMs: 60_000,
       },
       warn,
       attempt: async ({ voiceSessionId }) => {
@@ -218,4 +223,72 @@ describe("client voice mutation digest owner", () => {
     await vi.waitFor(() => expect(owner.snapshot().retained).toBe(0));
     expect(attempts.map((attempt) => attempt.id)).toEqual(["v1", "v2"]);
   });
+
+  it("drops a permanently failing intent after a bounded budget and releases capacity", async () => {
+    const warn = vi.fn();
+    const attempts: string[] = [];
+    const owner = new ClientVoiceMutationDigestOwner<number>({
+      policy: {
+        maxRetainedIntents: 1,
+        maxRetainedIdentityBytes: 64,
+        maxConcurrentAttempts: 1,
+        maxAttemptFailures: 2,
+        attemptAbortAfterMs: 60_000,
+      },
+      warn,
+      attempt: async ({ agentId }) => {
+        attempts.push(agentId);
+        if (agentId === "first") {
+          throw new Error("permanent");
+        }
+        return true;
+      },
+    });
+
+    owner.record({ agentId: "first", voiceSessionId: "v1", context: 1 });
+    await vi.waitFor(() => expect(owner.snapshot().active).toBe(0));
+    owner.retry({ agentId: "first", voiceSessionId: "v1" });
+    await vi.waitFor(() => expect(owner.snapshot().retained).toBe(0));
+
+    owner.record({ agentId: "second", voiceSessionId: "v2", context: 2 });
+    await vi.waitFor(() => expect(owner.snapshot().retained).toBe(0));
+    expect(attempts).toEqual(["first", "first", "second"]);
+    expect(warn).toHaveBeenLastCalledWith(
+      "voice mutation digest dropped after 2 failed attempts: permanent",
+    );
+  });
+
+  it("ignores settlement from an attempt owned by a cleared generation", async () => {
+    const attempts: Array<{
+      context: number;
+      completion: ReturnType<typeof deferred<boolean>>;
+    }> = [];
+    const owner = new ClientVoiceMutationDigestOwner<number>({
+      policy: {
+        maxRetainedIntents: 1,
+        maxRetainedIdentityBytes: 64,
+        maxConcurrentAttempts: 1,
+        maxAttemptFailures: 3,
+        attemptAbortAfterMs: 60_000,
+      },
+      warn: vi.fn(),
+      attempt: async ({ context }) => {
+        const completion = deferred<boolean>();
+        attempts.push({ context, completion });
+        return await completion.promise;
+      },
+    });
+
+    owner.record({ agentId: "a", voiceSessionId: "v1", context: 1 });
+    owner.clear();
+    owner.record({ agentId: "a", voiceSessionId: "v1", context: 2 });
+    owner.record({ agentId: "a", voiceSessionId: "v1", context: 3 });
+
+    attempts[0]?.completion.reject(new Error("old generation"));
+    attempts[1]?.completion.resolve(false);
+    await vi.waitFor(() => expect(attempts).toHaveLength(3));
+    expect(attempts[2]?.context).toBe(3);
+    attempts[2]?.completion.resolve(true);
+    await vi.waitFor(() => expect(owner.snapshot().retained).toBe(0));
+  });
 });

src/talk/client-voice-mutation-digest-owner.ts

@@ -14,7 +14,8 @@ export const CLIENT_VOICE_MUTATION_DIGEST_POLICY = {
   maxRetainedIntents: 64,
   maxRetainedIdentityBytes: 64 * 1024,
   maxConcurrentAttempts: 2,
-  attemptTimeoutMs: 30_000,
+  maxAttemptFailures: 3,
+  attemptAbortAfterMs: 30_000,
 } as const;
 
 function formatMutationDigest(effects: ClientVoiceToolEffect[]): string | undefined {
@@ -94,18 +95,21 @@ type MutationDigestIntent<TContext> = {
   voiceSessionId: string;
   context: TContext;
   identityBytes: number;
+  failedAttempts: number;
 };
 
 type MutationDigestAttempt<TContext> = {
   controller: AbortController;
   intent: MutationDigestIntent<TContext>;
+  generation: number;
 };
 
 type MutationDigestPolicy = {
   maxRetainedIntents: number;
   maxRetainedIdentityBytes: number;
   maxConcurrentAttempts: number;
-  attemptTimeoutMs: number;
+  maxAttemptFailures: number;
+  attemptAbortAfterMs: number;
 };
 
 export class ClientVoiceMutationDigestOwner<TContext> {
@@ -114,6 +118,7 @@ export class ClientVoiceMutationDigestOwner<TContext> {
   private readonly retryAfterActiveKeys = new Set<string>();
   private readonly activeAttempts = new Map<string, MutationDigestAttempt<TContext>>();
   private retainedIdentityBytes = 0;
+  private generation = 0;
 
   constructor(
     private readonly options: {
@@ -160,7 +165,7 @@ export class ClientVoiceMutationDigestOwner<TContext> {
       this.options.warn("voice mutation digest retry owner is full");
       return;
     }
-    const intent = { ...params, identityBytes };
+    const intent = { ...params, identityBytes, failedAttempts: 0 };
     this.intents.set(key, intent);
     this.retainedIdentityBytes += identityBytes;
     this.pendingKeys.add(key);
@@ -213,6 +218,7 @@ export class ClientVoiceMutationDigestOwner<TContext> {
     for (const attempt of this.activeAttempts.values()) {
       attempt.controller.abort(new Error("voice mutation digest delivery owner reset"));
     }
+    this.generation += 1;
     this.intents.clear();
     this.pendingKeys.clear();
     this.retryAfterActiveKeys.clear();
@@ -255,27 +261,47 @@ export class ClientVoiceMutationDigestOwner<TContext> {
 
   private startAttempt(key: string, intent: MutationDigestIntent<TContext>): void {
     const controller = new AbortController();
-    const attempt = { controller, intent };
+    const attempt = { controller, intent, generation: this.generation };
     this.activeAttempts.set(key, attempt);
     const timeout = setTimeout(
-      () => controller.abort(new Error("voice mutation digest delivery timed out")),
-      this.policy.attemptTimeoutMs,
+      () => controller.abort(new Error("voice mutation digest delivery abort requested")),
+      this.policy.attemptAbortAfterMs,
     );
     timeout.unref?.();
-    // Do not race the timeout. An adapter that ignores abort keeps this exact
-    // attempt and concurrency slot until its underlying promise really settles.
-    void this.options
-      .attempt({ ...intent, signal: controller.signal })
+    // Abort is cooperative, not a wall-clock completion guarantee. An adapter
+    // that ignores it keeps this exact slot so repeated retries cannot fan out.
+    let completion: Promise<boolean>;
+    try {
+      completion = this.options.attempt({ ...intent, signal: controller.signal });
+    } catch (error) {
+      completion = Promise.reject(error);
+    }
+    void completion
       .then((complete) => {
         if (complete) {
           this.deleteIntent(key, intent);
         }
       })
       .catch((error: unknown) => {
-        this.options.warn(error instanceof Error ? error.message : String(error));
+        if (attempt.generation !== this.generation) {
+          return;
+        }
+        intent.failedAttempts += 1;
+        const message = error instanceof Error ? error.message : String(error);
+        if (intent.failedAttempts >= this.policy.maxAttemptFailures) {
+          this.deleteIntent(key, intent);
+          this.options.warn(
+            `voice mutation digest dropped after ${intent.failedAttempts} failed attempts: ${message}`,
+          );
+          return;
+        }
+        this.options.warn(message);
       })
       .finally(() => {
         clearTimeout(timeout);
+        if (attempt.generation !== this.generation) {
+          return;
+        }
         if (this.activeAttempts.get(key) === attempt) {
           this.activeAttempts.delete(key);
         }

src/talk/client-voice-session.test-support.ts

@@ -7,7 +7,8 @@ type ClientVoiceSessionTestApi = {
     maxRetainedIntents: number;
     maxRetainedIdentityBytes: number;
     maxConcurrentAttempts: number;
-    attemptTimeoutMs: number;
+    maxAttemptFailures: number;
+    attemptAbortAfterMs: number;
   };
   digestDeliverySnapshot(): {
     active: number;