diff --git a/packages/coding-agent/src/async/job-manager.ts b/packages/coding-agent/src/async/job-manager.ts index c67dd4e35..cc52670ca 100644 --- a/packages/coding-agent/src/async/job-manager.ts +++ b/packages/coding-agent/src/async/job-manager.ts @@ -47,6 +47,7 @@ export interface AsyncJobRegisterOptions { export class AsyncJobManager { readonly #jobs = new Map(); readonly #deliveries: AsyncJobDelivery[] = []; + readonly #suppressedDeliveries = new Set(); readonly #evictionTimers = new Map(); readonly #onJobComplete: AsyncJobManagerOptions["onJobComplete"]; readonly #maxRunningJobs: number; @@ -81,6 +82,7 @@ export class AsyncJobManager { } const id = this.#resolveJobId(options?.id); + this.#suppressedDeliveries.delete(id); const abortController = new AbortController(); const startTime = Date.now(); @@ -182,6 +184,23 @@ export class AsyncJobManager { return this.#deliveries.length > 0; } + acknowledgeDeliveries(jobIds: string[]): number { + const uniqueJobIds = Array.from(new Set(jobIds.map(id => id.trim()).filter(id => id.length > 0))); + if (uniqueJobIds.length === 0) return 0; + + for (const jobId of uniqueJobIds) { + this.#suppressedDeliveries.add(jobId); + } + + const before = this.#deliveries.length; + this.#deliveries.splice( + 0, + this.#deliveries.length, + ...this.#deliveries.filter(delivery => !this.#suppressedDeliveries.has(delivery.jobId)), + ); + return before - this.#deliveries.length; + } + cancelAll(): void { for (const job of this.getRunningJobs()) { job.status = "cancelled"; @@ -234,6 +253,7 @@ export class AsyncJobManager { this.#clearEvictionTimers(); this.#jobs.clear(); this.#deliveries.length = 0; + this.#suppressedDeliveries.clear(); return drained; } @@ -257,6 +277,7 @@ export class AsyncJobManager { #scheduleEviction(jobId: string): void { if (this.#retentionMs <= 0) { this.#jobs.delete(jobId); + this.#suppressedDeliveries.delete(jobId); return; } const existing = this.#evictionTimers.get(jobId); @@ -266,6 +287,7 @@ export class AsyncJobManager { const timer = setTimeout(() => { this.#evictionTimers.delete(jobId); this.#jobs.delete(jobId); + this.#suppressedDeliveries.delete(jobId); }, this.#retentionMs); timer.unref(); this.#evictionTimers.set(jobId, timer); @@ -278,7 +300,14 @@ export class AsyncJobManager { this.#evictionTimers.clear(); } + #isDeliverySuppressed(jobId: string): boolean { + return this.#suppressedDeliveries.has(jobId); + } + #enqueueDelivery(jobId: string, text: string): void { + if (this.#isDeliverySuppressed(jobId)) { + return; + } this.#deliveries.push({ jobId, text, @@ -308,10 +337,21 @@ export class AsyncJobManager { async #runDeliveryLoop(): Promise { while (this.#deliveries.length > 0) { const delivery = this.#deliveries[0]; + if (this.#isDeliverySuppressed(delivery.jobId)) { + this.#deliveries.shift(); + continue; + } const waitMs = delivery.nextAttemptAt - Date.now(); if (waitMs > 0) { await Bun.sleep(waitMs); } + if (this.#deliveries[0] !== delivery) { + continue; + } + if (this.#isDeliverySuppressed(delivery.jobId)) { + this.#deliveries.shift(); + continue; + } try { await this.#onJobComplete(delivery.jobId, delivery.text, this.#jobs.get(delivery.jobId)); @@ -321,7 +361,9 @@ export class AsyncJobManager { delivery.lastError = error instanceof Error ? error.message : String(error); delivery.nextAttemptAt = Date.now() + this.#getRetryDelay(delivery.attempt); this.#deliveries.shift(); - this.#deliveries.push(delivery); + if (!this.#isDeliverySuppressed(delivery.jobId)) { + this.#deliveries.push(delivery); + } logger.warn("Async job completion delivery failed", { jobId: delivery.jobId, attempt: delivery.attempt, diff --git a/packages/coding-agent/src/tools/await-tool.ts b/packages/coding-agent/src/tools/await-tool.ts index 4d5898f5b..e1983aa48 100644 --- a/packages/coding-agent/src/tools/await-tool.ts +++ b/packages/coding-agent/src/tools/await-tool.ts @@ -79,7 +79,7 @@ export class AwaitTool implements AgentTool j.status === "running"); if (runningJobs.length === 0) { - return this.#buildResult(jobsToWatch); + return this.#buildResult(manager, jobsToWatch); } // Block until at least one running job finishes or the call is aborted @@ -100,13 +100,14 @@ export class AwaitTool implements AgentTool, jobs: { id: string; type: "bash" | "task"; @@ -128,6 +129,8 @@ export class AwaitTool implements AgentTool j.status !== "running").map(j => j.id)); + const completed = jobResults.filter(j => j.status !== "running"); const running = jobResults.filter(j => j.status === "running"); diff --git a/packages/coding-agent/test/async-job-manager.test.ts b/packages/coding-agent/test/async-job-manager.test.ts index f519f4643..1861d92da 100644 --- a/packages/coding-agent/test/async-job-manager.test.ts +++ b/packages/coding-agent/test/async-job-manager.test.ts @@ -181,6 +181,37 @@ describe("AsyncJobManager", () => { expect(manager.getJob(runningJobId)).toBeUndefined(); }); + test("acknowledgeDeliveries suppresses pending retries for completed jobs", async () => { + let attempts = 0; + const manager = new AsyncJobManager({ + onJobComplete: async () => { + attempts += 1; + throw new Error("delivery failed"); + }, + }); + + const jobId = manager.register("task", "awaited-job", async () => "done"); + await manager.waitForAll(); + + const firstAttemptDeadline = Date.now() + 2_000; + while (attempts === 0) { + if (Date.now() >= firstAttemptDeadline) throw new Error("Timed out waiting for first delivery attempt"); + await Bun.sleep(5); + } + + expect(manager.hasPendingDeliveries()).toBe(true); + const removed = manager.acknowledgeDeliveries([jobId]); + expect(removed).toBeGreaterThanOrEqual(1); + + const drained = await manager.drainDeliveries({ timeoutMs: 200 }); + expect(drained).toBe(true); + expect(manager.hasPendingDeliveries()).toBe(false); + + const attemptsAfterAck = attempts; + await Bun.sleep(700); + expect(attempts).toBe(attemptsAfterAck); + }); + test("dispose clears jobs and pending deliveries", async () => { const manager = new AsyncJobManager({ onJobComplete: async () => {