diff --git a/src/talk/client-voice-mutation-digest-owner.test.ts b/src/talk/client-voice-mutation-digest-owner.test.ts index 53c7db34d738..c4909da34f5f 100644 --- a/src/talk/client-voice-mutation-digest-owner.test.ts +++ b/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>; 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({ + 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>; + }> = []; + const owner = new ClientVoiceMutationDigestOwner({ + policy: { + maxRetainedIntents: 1, + maxRetainedIdentityBytes: 64, + maxConcurrentAttempts: 1, + maxAttemptFailures: 3, + attemptAbortAfterMs: 60_000, + }, + warn: vi.fn(), + attempt: async ({ context }) => { + const completion = deferred(); + 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)); + }); }); diff --git a/src/talk/client-voice-mutation-digest-owner.ts b/src/talk/client-voice-mutation-digest-owner.ts index 156914fd13a8..80ee9e01c3a4 100644 --- a/src/talk/client-voice-mutation-digest-owner.ts +++ b/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 = { voiceSessionId: string; context: TContext; identityBytes: number; + failedAttempts: number; }; type MutationDigestAttempt = { controller: AbortController; intent: MutationDigestIntent; + generation: number; }; type MutationDigestPolicy = { maxRetainedIntents: number; maxRetainedIdentityBytes: number; maxConcurrentAttempts: number; - attemptTimeoutMs: number; + maxAttemptFailures: number; + attemptAbortAfterMs: number; }; export class ClientVoiceMutationDigestOwner { @@ -114,6 +118,7 @@ export class ClientVoiceMutationDigestOwner { private readonly retryAfterActiveKeys = new Set(); private readonly activeAttempts = new Map>(); private retainedIdentityBytes = 0; + private generation = 0; constructor( private readonly options: { @@ -160,7 +165,7 @@ export class ClientVoiceMutationDigestOwner { 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 { 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 { private startAttempt(key: string, intent: MutationDigestIntent): 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; + 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); } diff --git a/src/talk/client-voice-session.test-support.ts b/src/talk/client-voice-session.test-support.ts index 4c03de132451..27a598287fb3 100644 --- a/src/talk/client-voice-session.test-support.ts +++ b/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;