From c2ff3eb92113fb16713b270e34932636956ff16b Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 21:49:38 -0500 Subject: [PATCH 1/4] fix: wake retry scheduling only when retry state changes Separate retry-schedule notifications from ordinary queued-run eligibility. Preserve exact deadlines and prove native producer commit/rollback behavior without global retry scans on ordinary agent or issue edits. Co-Authored-By: Paperclip --- doc/plans/2026-10-30-run-dispatch-sleep.md | 9 ++++ .../heartbeat-run-dispatch-lifecycle.test.ts | 19 ++++---- .../heartbeat-run-dispatch-work.test.ts | 48 ++++++++++++++++++- server/src/services/heartbeat/retries.ts | 6 +-- .../native-provider-capacity-retry.ts | 4 +- .../native-safe-replacement.test.ts | 10 ++++ .../native-runtime/native-safe-replacement.ts | 2 + server/src/services/recovery/service.ts | 6 +-- .../services/run-dispatch-notifications.ts | 7 ++- ...eduled-retry-cancellation-notifications.ts | 4 +- 10 files changed, 94 insertions(+), 21 deletions(-) diff --git a/doc/plans/2026-10-30-run-dispatch-sleep.md b/doc/plans/2026-10-30-run-dispatch-sleep.md index e5cd0a9e014..8ad35c128fc 100644 --- a/doc/plans/2026-10-30-run-dispatch-sleep.md +++ b/doc/plans/2026-10-30-run-dispatch-sleep.md @@ -12,6 +12,15 @@ invalidate retries notify the retry worker in their transaction. Promotion uses the existing gate and compare-and-set, and its committed write wakes ordinary queued dispatch. A queued run behind occupied capacity waits for the owning run's lifecycle transition. Configuration and authority changes also wake it. +Retry-schedule notifications are separate from ordinary queue eligibility. +`notifyRunDispatch` wakes only queued dispatch; ordinary issue, agent, budget +and company edits cannot repeatedly scan the global retry schedule. +`notifyScheduledRetryDispatch` wakes retry reconciliation as well as dispatch +for actual retry creation, rescheduling and cancellation. Its producers include +bounded and operator-requested retries, recovery retries, native capacity and +safe-replacement successors, and the existing retry-cancellation transition +boundary. Shared runtime changes do not impose a spacing floor on a saved deadline. + An idle queue has no recurring read or timer-driven dispatch attempt. Notifications batch for one second and carry the changed company to queued dispatch. Known retry deadlines still preempt batching. Cancellation-owned retries stay with their cancellation worker and do not retain a promotion deadline. The supported writer contract is service writes through the shared database diff --git a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts index f3398d828f8..a9e120387c8 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts @@ -3,7 +3,7 @@ import { PgDialect } from "drizzle-orm/pg-core"; import { afterEach, describe, expect, it, vi } from "vitest"; import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; -import { notifyRunDispatch } from "../services/run-dispatch-notifications.js"; +import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js"; import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js"; const cleanup: Array<() => Promise> = []; @@ -38,6 +38,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean }, () => !state.draining && !state.standby); cleanup.push(() => coordinator.stop()); return { db, state, promote, next, dispatch, coordinator, worker, + notifyRetry: () => db.transaction(tx => notifyScheduledRetryDispatch(tx)), notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)), settle: () => vi.advanceTimersByTimeAsync(1_000) }; } @@ -61,9 +62,9 @@ describe("run dispatch obligations", () => { }); it("rearms earlier work and cancels a future deadline on a committed producer change", async () => { const f = fixture({ due: 1_600_000 }); await f.worker.ready; - f.state.due = 1_005_000; await f.notify(); await f.settle(); + f.state.due = 1_005_000; await f.notifyRetry(); await f.settle(); expect(f.coordinator.nextWakeAt()).toBe(1_005_000); - f.state.due = null; await f.notify(); await f.settle(); + f.state.due = null; await f.notifyRetry(); await f.settle(); expect(f.coordinator.nextWakeAt()).toBeNull(); const before = f.next.mock.calls.length; await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before); @@ -76,7 +77,7 @@ describe("run dispatch obligations", () => { await f.notify(source); await vi.advanceTimersByTimeAsync(10); } - expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2); + expect(f.promote).toHaveBeenCalledOnce(); expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2); await f.settle(); expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" }); @@ -87,7 +88,7 @@ describe("run dispatch obligations", () => { }); it("remembers the first producer arriving during drain without owner SQL", async () => { const f = fixture(); await f.worker.ready; f.state.draining = true; - await f.notify(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000); + await f.notifyRetry(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce(); expect(idleWorkSnapshot().active).toBe(0); f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000); @@ -96,7 +97,7 @@ describe("run dispatch obligations", () => { }); it("releases a failed producer hold during drain and retains recovery for resume", async () => { const f = fixture(); await f.worker.ready; - f.state.fail = true; await f.notify(); await f.settle(); + f.state.fail = true; await f.notifyRetry(); await f.settle(); expect(f.promote).toHaveBeenCalledTimes(2); f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce(); @@ -109,8 +110,8 @@ describe("run dispatch obligations", () => { let release!: () => void; const held = new Promise(resolve => { release = resolve; }); f.promote.mockImplementationOnce(() => held); - await f.notify(); await f.settle(); - await f.notify(); await f.settle(); + await f.notifyRetry(); await f.settle(); + await f.notifyRetry(); await f.settle(); expect(f.promote).toHaveBeenCalledTimes(2); release(); await f.settle(); await vi.advanceTimersByTimeAsync(30_000); @@ -121,7 +122,7 @@ describe("run dispatch obligations", () => { await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).not.toHaveBeenCalled(); f.state.standby = false; await vi.advanceTimersByTimeAsync(30_000); expect(f.promote).toHaveBeenCalledOnce(); expect(f.state.due).toBeNull(); - f.state.due = Date.now() + 60_000; await f.notify(); await f.settle(); await f.worker.stop(); + f.state.due = Date.now() + 60_000; await f.notifyRetry(); await f.settle(); await f.worker.stop(); await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2); }); }); diff --git a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts index bcc8ed23972..1c0e5005a04 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts @@ -2,9 +2,11 @@ import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns } from "@paperclipai/db"; +import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns, issues } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { issueService } from "../services/issues.js"; +import { materializeNativeProviderCapacityRetry } from "../services/native-runtime/native-provider-capacity-retry.js"; import { agentService } from "../services/agents.js"; import { budgetService } from "../services/budgets.js"; import { heartbeatService } from "../services/heartbeat.js"; @@ -24,9 +26,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled let reads: ReturnType; let draining = false; const errors: unknown[] = []; + const retryQueries: string[] = []; beforeAll(async () => { if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-"); db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + db.$client.options.debug = (_connection, query) => { + if (query.startsWith("select") && query.includes('"heartbeat_runs"') && + query.includes("startupCancellation") && query.includes('"scheduled_retry_at"')) retryQueries.push(query); + }; await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); }, 90_000); beforeEach(async () => { @@ -88,6 +95,45 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled await fixture(); await worker!.stop(); await coordinator!.stop(); await start(); await quiet(); }); + it.each(["agent", "issue"] as const)("does not scan global scheduled retries for repeated ordinary %s edits", async kind => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Edit source", status: "backlog" }).returning(); + await start(); await quiet(); + const before = retryQueries.length; + expect(before).toBeGreaterThan(0); // The passive observer saw real startup retry SQL. + for (let index = 0; index < 12; index++) { + if (kind === "agent") await agentService(db).update(f.agentId, { name: `Edited agent ${index}` }); + else await issueService(db).update(issue.id, { title: `Edited issue ${index}` }); + await new Promise(resolve => setTimeout(resolve, 100)); + } + await quiet(); + expect(retryQueries).toHaveLength(before); + }); + it.each([false, true])("observes a native capacity retry only after its transaction settles (rollback=%s)", async rollback => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Capacity retry", status: "in_progress", + assigneeAgentId: f.agentId }).returning(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.runId)); + await start(); await quiet(); + let retryId = ""; + const writing = db.transaction(async tx => { + retryId = await materializeNativeProviderCapacityRetry({ tx: tx as unknown as typeof db, ...f, issueId: issue.id }); + expect(databaseWorkPending(db, DELIVERY_QUEUES.scheduledRetryDispatch)).toBe(true); + expect(await status(retryId)).toBeUndefined(); + if (rollback) throw new Error("native capacity rollback"); + }); + if (rollback) { + await expect(writing).rejects.toThrow("native capacity rollback"); + await quiet(); expect(await status(retryId)).toBeUndefined(); + } else { + await writing; + const retry = await status(retryId); + expect(retry.status).toBe("scheduled_retry"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(retry.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + await heartbeat.cancelRun(retryId, "fixture teardown", { suppressImmediateRecovery: true }); + await quiet(); + } + }); it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => { const f = await fixture(); await start(); await quiet(); const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 }); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 32333c0ebec..9eda1347ffa 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,4 +1,4 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js"; import { infrastructureRetryDelayMs } from "../system-recovery-state.js"; import { @@ -998,7 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen retryContextSnapshot.previousRunId = continuation.previousRunId; } - await notifyRunDispatch(tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); const wakeupRequest = await tx .insert(agentWakeupRequests) .values({ @@ -1756,7 +1756,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen }; const updated = await db.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: scheduled.run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId }); const row = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-provider-capacity-retry.ts b/server/src/services/native-runtime/native-provider-capacity-retry.ts index 83314ff9c51..99575395f52 100644 --- a/server/src/services/native-runtime/native-provider-capacity-retry.ts +++ b/server/src/services/native-runtime/native-provider-capacity-retry.ts @@ -1,4 +1,4 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db"; @@ -50,7 +50,7 @@ export async function materializeNativeProviderCapacityRetry(input: { scheduledRetryAt: dueAt.toISOString(), executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt), }; - await notifyRunDispatch(input.tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(input.tx, { companyId: run.companyId }); const [wake] = await input.tx.insert(agentWakeupRequests).values({ companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system", reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId, diff --git a/server/src/services/native-runtime/native-safe-replacement.test.ts b/server/src/services/native-runtime/native-safe-replacement.test.ts index 422b23dba6b..491bae17b2b 100644 --- a/server/src/services/native-runtime/native-safe-replacement.test.ts +++ b/server/src/services/native-runtime/native-safe-replacement.test.ts @@ -1,3 +1,4 @@ +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { getExecutionBlocker } from "../execution-blocker.js"; import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; @@ -577,7 +578,12 @@ const support = externalDatabaseUrl }); it("persists exactly one linked successor under competing sweepers and restarts", async () => { const source = await seed(2); + const owner = heartbeatService(db); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const worker = owner.registerRunDispatchWork(coordinator, () => true); + await worker.ready; const now = new Date(); + try { await Promise.all([ reconcileSafeNativeReplacements(db, now), reconcileSafeNativeReplacements(db, now), @@ -608,6 +614,10 @@ const support = externalDatabaseUrl .where(eq(nativeRunFinalizations.runId, source.runId)) )[0]?.failureDetail?.successorRunId, ).toBe(children[0]!.id); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(children[0]!.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + } finally { + await worker.stop(); await coordinator.stop(); await owner.drainActiveRunExecutions(); + } }); it("retries status publication after a crash without dispatching provider work", async () => { await deliverExecutionStatuses(db); diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index da485883cb0..12b556fa779 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; @@ -410,6 +411,7 @@ export async function reconcileSafeNativeReplacements( }) .returning(); await notifyExecutionRunWork(tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: run.companyId, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index fffc2699dbf..e3596791942 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,4 +1,4 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -2717,7 +2717,7 @@ export function recoveryService( const now = new Date(); const retryAt = readProviderQuotaRetryAt(input.latestRun, now); return database.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: input.issue.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ @@ -3318,7 +3318,7 @@ export function recoveryService( created = Boolean(enqueuedRun); } else { scheduledRun = await db.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: input.issue.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index e62b8c62105..276edacab23 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -4,6 +4,11 @@ export type RunDispatchScope = { companyId: string }; export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; - await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); } + +/** Only a retry row or deadline change needs to re-read the saved retry schedule. */ +export async function notifyScheduledRetryDispatch(transaction: object, scope?: RunDispatchScope) { + await notifyRunDispatch(transaction, scope); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch); +} diff --git a/server/src/services/scheduled-retry-cancellation-notifications.ts b/server/src/services/scheduled-retry-cancellation-notifications.ts index 24e57de66cb..945bb40e14b 100644 --- a/server/src/services/scheduled-retry-cancellation-notifications.ts +++ b/server/src/services/scheduled-retry-cancellation-notifications.ts @@ -1,9 +1,9 @@ -import { notifyRunDispatch } from "./run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "./run-dispatch-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type ScheduledRetryCancellationScope = { companyId: string; runId?: string }; export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) { - await notifyRunDispatch(transaction, scope); + await notifyScheduledRetryDispatch(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*"); } From 6ee17aae1c61886a77ae8570fc0d360689c4dc0d Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 21:55:27 -0500 Subject: [PATCH 2/4] test: settle saved retry cleanup before idle assertion Observe the specific succeeded retry before closing its task and draining all execution. Once execution has settled, advance bounded fake cleanup time while joining worker passes; retain the existing timeout, exact retry identity, output and zero-SQL assertions. Co-Authored-By: Paperclip --- .../deferred-execution-receipt-work.test.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index 78c56ff9a83..56b0b057af3 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -78,7 +78,15 @@ describe("deferred receipt durable dispatch", () => { return { companyId, agent, issue, source, actorId }; } async function quiet() { - await vi.waitFor(async () => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 }); + await vi.waitFor(async () => { + if (vi.isFakeTimers()) { + // Fake-clock cases have no live execution. Settle finite cleanup + // deadlines explicitly instead of advancing at waitFor's wall speed. + await vi.advanceTimersByTimeAsync(1000); + await Promise.allSettled([...passes]); await immediate(); + } + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 40_000 }); const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"); const selected = reads.mock.calls.length, executed = executes.mock.calls.length; if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); @@ -455,8 +463,8 @@ describe("deferred receipt durable dispatch", () => { const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)); expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId; }, { timeout: 40_000 }); + // Observe this Retry before closing the task and draining its follow-ups. await vi.waitFor(async () => { - await heartbeat.drainActiveRunExecutions(); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("receipt-proof"); From 804a64a842be3255cb4c8d7c1e2f42e61f84908d Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 22:10:55 -0500 Subject: [PATCH 3/4] test: settle delegated chat completion before plan assertions Handle completion notifications in the deterministic agent instead of replaying the original plan command. Wait for the real completion turn and retain exact plan revision, output, project, and assignee assertions. Co-Authored-By: Paperclip --- tests/e2e/agent-chat-projects.spec.ts | 3 +++ tests/e2e/fixtures/agent-chat.mjs | 4 ++++ 2 files changed, 7 insertions(+) diff --git a/tests/e2e/agent-chat-projects.spec.ts b/tests/e2e/agent-chat-projects.spec.ts index 994df38746c..89ec94ca54c 100644 --- a/tests/e2e/agent-chat-projects.spec.ts +++ b/tests/e2e/agent-chat-projects.spec.ts @@ -459,12 +459,15 @@ test("plan approval hands the preserved revision to an assigned project task", a { timeout: 60_000 }, ) .toBe(1); + // Task completion is delivered back to the chat as another real turn. + await idle(request, f.chatPath, 3); const task = ( await json(await request.get(`/api/companies/${f.company.id}/issues`)) )[0]; expect(task.parentId).toBeNull(); expect(task.projectId).toBeTruthy(); expect(task.assigneeAgentId).toBe(f.agent.id); + await expect(page.getByText(`Delegated work completed: ${task.identifier}`, { exact: true })).toBeVisible(); const plan = await json( await request.get(`/api/issues/${task.id}/documents/plan`), ); diff --git a/tests/e2e/fixtures/agent-chat.mjs b/tests/e2e/fixtures/agent-chat.mjs index 2a66bdd3111..d85ab355679 100644 --- a/tests/e2e/fixtures/agent-chat.mjs +++ b/tests/e2e/fixtures/agent-chat.mjs @@ -36,6 +36,10 @@ if (!task.conversationAgentId) { }); process.exit(0); } +if (ctx.wakeReason === "chat_task_completed") { + await comment(`Delegated work completed: ${ctx.chatCompletionUpdates.map((update) => update.identifier).join(", ")}`); + process.exit(0); +} const comments = await api(`/issues/${task.id}/comments?order=asc`); const current = comments.find((c) => c.id === ctx.wakeCommentId) ?? From 08e27cfe096a2f1bad63cd344f525c6b76008647 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 22:12:07 -0500 Subject: [PATCH 4/4] Remove unused aggregate watchdog diagnostics The deadline worker retains its incomplete-subtree error and classifier tests. Remove the obsolete reporter and its two isolated tests after its aggregate callers were removed. Co-Authored-By: Paperclip --- .../task-watchdogs-classifier.test.ts | 37 +------------------ .../src/services/task-watchdog-diagnostics.ts | 16 -------- 2 files changed, 1 insertion(+), 52 deletions(-) delete mode 100644 server/src/services/task-watchdog-diagnostics.ts diff --git a/server/src/__tests__/task-watchdogs-classifier.test.ts b/server/src/__tests__/task-watchdogs-classifier.test.ts index 16ef800c088..71865c75dfe 100644 --- a/server/src/__tests__/task-watchdogs-classifier.test.ts +++ b/server/src/__tests__/task-watchdogs-classifier.test.ts @@ -1,8 +1,6 @@ -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it } from "vitest"; import { classifyTaskWatchdogSubtree, type TaskWatchdogClassifierIssue } from "../services/task-watchdogs.ts"; -import { createIncompleteTaskWatchdogScanReporter } from "../services/task-watchdog-diagnostics.ts"; - const companyId = "company-1"; const sourceId = "source-1"; const childId = "child-1"; @@ -37,39 +35,6 @@ function classify(overrides: Partial { - it("reports an unchanged incomplete scan once across startup and a day of periodic ticks", () => { - const warn = vi.fn(); - const report = createIncompleteTaskWatchdogScanReporter(warn); - report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "startup"); - for (let tick = 0; tick < 2880; tick += 1) { - report({ incomplete: 2, incompleteIssueIds: [childId, sourceId] }, "periodic"); - } - expect(warn).toHaveBeenCalledTimes(1); - expect(warn).toHaveBeenCalledWith( - { incomplete: 2, newIncompleteIssueIds: [sourceId, childId] }, - "startup task-watchdog reconciliation found incomplete subtree scans", - ); - }); - - it("reports new incomplete issues and reports a resolved issue if it becomes incomplete again", () => { - const warn = vi.fn(); - const report = createIncompleteTaskWatchdogScanReporter(warn); - report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "startup"); - report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "periodic"); - expect(warn).toHaveBeenLastCalledWith( - { incomplete: 2, newIncompleteIssueIds: [childId] }, - "periodic task-watchdog reconciliation found incomplete subtree scans", - ); - report({ incomplete: 0, incompleteIssueIds: [] }, "periodic"); - expect(warn).toHaveBeenCalledTimes(2); - report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "periodic"); - expect(warn).toHaveBeenCalledTimes(3); - expect(warn).toHaveBeenLastCalledWith( - { incomplete: 1, newIncompleteIssueIds: [sourceId] }, - "periodic task-watchdog reconciliation found incomplete subtree scans", - ); - }); - it("does not produce a stopped fingerprint from an incomplete scan", () => { const result = classify({ subtreeIncomplete: true }); expect(result.state).toBe("incomplete"); diff --git a/server/src/services/task-watchdog-diagnostics.ts b/server/src/services/task-watchdog-diagnostics.ts deleted file mode 100644 index ea08eebe2b4..00000000000 --- a/server/src/services/task-watchdog-diagnostics.ts +++ /dev/null @@ -1,16 +0,0 @@ -export function createIncompleteTaskWatchdogScanReporter( - warn: (details: { incomplete: number; newIncompleteIssueIds: string[] }, message: string) => void, -) { - let previousIncompleteIssueIds = new Set(); - return (result: { incomplete: number; incompleteIssueIds: string[] }, phase: "startup" | "periodic") => { - const currentIssueIds = new Set(result.incompleteIssueIds); - const newIncompleteIssueIds = [...currentIssueIds].filter((id) => !previousIncompleteIssueIds.has(id)); - previousIncompleteIssueIds = currentIssueIds; - if (newIncompleteIssueIds.length === 0) return; - warn( - { incomplete: result.incomplete, newIncompleteIssueIds }, - `${phase} task-watchdog reconciliation found incomplete subtree scans`, - ); - }; -} -