diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 325148106d2..d93ae631b6c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -756,6 +756,8 @@ Normal issue-thread feedback can still wake the current assignee according to th Restart recovery revisits accepted deferred messages through the latest issue run, including historical mention queues for another agent. Ordinary messages for a former owner retain their assignment handoff. Native sources require their matching committed finalization, settled environment leases, and released finalizer and copyback ownership before recovery clears old issue claims. A successful retained warm release can satisfy the lease check. Released local bookkeeping without a provider allocation and a matching SSH bookkeeping release need no provider cleanup receipt. Plugin releases record their successful acknowledgement. Pending or failed cleanup and durable owners still hold work. A successful fenced cleanup completion clears its in-flight claim in the same database write, so a restart before retry bookkeeping cannot strand saved input. The normal release checks preserve cleanup holds, pauses, and operator Stops. Recovery checks for a competing turn under the issue lock before it promotes saved input. +Repeated eligibility notifications must not move an unchecked, ready deferred receipt behind receipts already reconsidered. Paginated reconstruction preserves pending ready receipts' dispatch priority while refreshing their current state. Receipts whose wait deadline has passed join the ready queue on reconstruction. Future wait grace periods, claimed-work deadlines, infrastructure retry backoffs, and current admission checks remain effective. + A paused recipient does not fail an accepted deferred request. Release and periodic recovery preserve its pending receipt and may process an eligible request behind it. Resuming the agent allows normal admission to reconsider the saved input. Missing, foreign, terminated, and unapproved recipients retain their rejection rules. Explicit task holds and operator Stops still prevent unauthorized execution. This does not replay historical failed requests. Accepted agent feedback must survive a child changing to `done` before its active run exits. Deferred wake promotion may reopen that completed child only for its current assignee, with explicit agent resume intent and live tracked comments from another author. Claim promotion before reopening. Cancelled tasks, deleted comments, self-authored comments, empty continuations, and agent continuations without explicit intent keep their existing suppression rules. Normal pause, ownership, authorization, and budget gates still apply. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 25d571411f2..89f7c9741d8 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -64441,6 +64441,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { await f.service.shutdown(); } }); + it("retries saved GitHub input after an endpoint admission lock collision", async () => { + const f = await successfulGitHubQueuedInputFixture(); + const heartbeat = heartbeatService(db); + const failures: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => failures.push(error) }); + let unlock!: () => void, locked!: () => void; + const acquired = new Promise(resolve => { locked = resolve; }); + const released = new Promise(resolve => { unlock = resolve; }); + const competingAdmission = db.transaction(async tx => { + await tx.select({ id: chatEndpoints.id }).from(chatEndpoints) + .where(eq(chatEndpoints.id, f.endpoint.id)).for("no key update"); + locked(); + await released; + }); + try { + await acquired; + await heartbeat.registerDeferredReceiptWork(coordinator, () => true).ready; + const priorFailures = failures.length; + const retry = await requestTaskRetry(f.input); + await expect.poll(() => failures.length, { timeout: 5_000 }).toBeGreaterThan(priorFailures); + expect((await readAgentWakeupRequestsById(f.pending!.id))[0].runId).toBeNull(); + unlock(); await competingAdmission; + await expect.poll(async () => (await readAgentWakeupRequestsById(f.pending!.id))[0].runId, + { timeout: 40_000 }).toBeTruthy(); + await expect.poll(async () => (await readAgentWakeupRequestsById(retry.requestId!))[0].status, + { timeout: 40_000 }).toBe("coalesced"); + const [saved] = await readAgentWakeupRequestsById(f.pending!.id); + expect((await readHeartbeatRunsByCompanyId(f.fixture.companyId)) + .filter(run => run.wakeupRequestId === f.pending!.id)).toHaveLength(1); + expect((await readAgentWakeupRequestsById(retry.requestId!))[0].runId).toBe(saved.runId); + expect((await readHeartbeatRunsById(f.runId))[0]).toMatchObject({ status: "succeeded" }); + } finally { + unlock(); await competingAdmission; + await coordinator.stop(); + await f.service.shutdown(); + } + }, 90_000); + it.each(["reconnected", "newer_input", "reconnected_newer_input", "snapshot_newer_input"] as const)( "admits current GitHub task Retry without replaying obsolete input: %s", async kind => { const f = await currentGitHubTaskRetryFixture(kind.includes("newer_input")); diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index e902c734429..7c028e3c2e8 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -72,14 +72,18 @@ describe("deferred receipt durable dispatch", () => { // during actual process execution; accelerated steps only settle idle work. vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); heartbeat = heartbeatService(db); - coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true, + coordinator = createCoordinator(); + }); + function createCoordinator() { + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true, onError: error => { errors.push(error); errorTimes.push(Date.now()); } }); - const register = coordinator.register.bind(coordinator); - coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const register = worker.register.bind(worker); + worker.register = (queue, task) => register(queue, { ...task, run: signal => { const pass = Promise.resolve(task.run(signal)); passes.add(pass); void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass; } }); - }); + return worker; + } afterEach(async () => { await coordinator.stop(); await heartbeat.drainActiveRunExecutions(); for (const stop of extraStops.splice(0)) await stop(); @@ -153,6 +157,78 @@ describe("deferred receipt durable dispatch", () => { source: "on_demand", status: "completed", reason: "task_retry", payload: { issueId: f.issue.id } }); await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; await quiet(); }); + it.each([false, true])("reconsiders every saved receipt during repeated company eligibility changes (expired wait: %s)", async expiredWait => { + const f = await fixture(); + const ids = Array.from({ length: 178 }, (_, index) => `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`); + await db.insert(agentWakeupRequests).values(ids.map(id => ({ id, companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution", + updatedAt: new Date(Date.now() - 60_000), + payload: { issueId: f.issue.id, _paperclipWakeContext: { wakeCommentIds: [randomUUID()] }, + ...(expiredWait ? { executionWait: { reason: "recovery" } } : {}) }, + }))); + const checked = new Set(); + await registerDeferredReceiptWork(db, coordinator, { + resume: async scope => { checked.add(scope.requestId!); }, + settleTaskChat: async () => {}, createNativeBatch: () => async () => {}, schedulingSuppressed: async () => false, + }, () => !drain).ready; + for (let pass = 0; pass < 12; pass++) { + await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId })); + await vi.advanceTimersByTimeAsync(31_000); + await Promise.allSettled([...passes]); + await immediate(); + } + expect(checked.size).toBe(ids.length); + await quiet(); + }); + it("dispatches task Retry past a repeatedly invalidated backlog into real process output", async () => { + const f = await fixture(); + // Hold executor capacity while queue time advances; reconstruct the saved + // run with fresh workers at PostgreSQL time before real process execution. + const [occupant] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "running", startedAt: new Date() }).returning(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, + wakeOnDemand: true, maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agent.id)); + const [closed] = await db.insert(issues).values({ companyId: f.companyId, title: "Closed task history", + status: "done", assigneeAgentId: f.agent.id }).returning(); + await db.insert(agentWakeupRequests).values(Array.from({ length: 178 }, (_, index) => ({ + id: `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`, + companyId: f.companyId, agentId: f.agent.id, source: "on_demand", status: "deferred_issue_execution", + payload: { issueId: closed.id, _paperclipWakeContext: { wakeCommentIds: [randomUUID()] } }, + }))); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + const retry = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId, + observedRunId: f.source.id, idempotencyKey: randomUUID() }); + const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retry.requestId!)))[0]; + for (let pass = 0; pass < 12 && !(await read()).runId; pass++) { + await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId })); + await vi.advanceTimersByTimeAsync(31_000); + await Promise.allSettled([...passes]); + await immediate(); + } + const receipt = await read(); + expect(receipt.runId).toBeTruthy(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)))[0].status).toBe("queued"); + await coordinator.stop(); + await heartbeat.drainActiveRunExecutions(); + vi.setSystemTime(vi.getRealSystemTime()); + heartbeat = heartbeatService(db); + coordinator = createCoordinator(); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, occupant.id)); + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await waitForReal(async () => { + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)); + expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded"); + expect(run.stdoutExcerpt).toContain("receipt-proof"); + }, { timeout: 15_000 }); + expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${retry.requestId}`)).toHaveLength(1); + await finish(f.issue.id); + await quiet(); + }); it("dispatches an ordinary committed task Retry into saved process output", async () => { const f = await fixture(); await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; diff --git a/server/src/services/heartbeat/deferred-receipt-work.ts b/server/src/services/heartbeat/deferred-receipt-work.ts index 9cca7528228..b799f10b0a7 100644 --- a/server/src/services/heartbeat/deferred-receipt-work.ts +++ b/server/src/services/heartbeat/deferred-receipt-work.ts @@ -6,7 +6,7 @@ import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; -type Entry = { id: string; companyId: string; dueAt: number }; +type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean }; type NativeIntentOptions = DeferredExecutionScope & { onDeferredDeadline(at: number): void; onError(error: unknown): void; @@ -39,6 +39,11 @@ export function registerDeferredReceiptWork( }, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint), matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.requestId || entry.id === scope.requestId), + // A busy company can invalidate a paginated scan before the pending + // batch drains. Keep unread ready receipts ahead of freshly reloaded + // receipts already checked; actual wait/claim deadlines still replace. + mergeEntry: (pending, loaded) => pending.immediate && loaded.immediate + ? { ...loaded, dueAt: Math.min(pending.dueAt, loaded.dueAt) } : loaded, load: async scope => { const key = scope ? JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }) : "*"; if (!enabled) { cursors.delete(key); return []; } @@ -76,12 +81,15 @@ export function registerDeferredReceiptWork( )).orderBy(asc(agentWakeupRequests.id)).limit(100); const complete = rows.length < 100; if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); - return { rows: rows.map(row => ({ id: row.id, companyId: row.companyId, - dueAt: kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000 + const now = Date.now(); + return { rows: rows.map(row => { + const deadline = kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000 : kind === "native" && row.error ? row.updatedAt.getTime() + 30_000 : kind === "deferred" && row.delayed && !row.interruptedLegacy ? row.updatedAt.getTime() + 30_000 - : Date.now(), - })), complete }; + : null; + const immediate = deadline === null || deadline <= now; + return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate }; + }), complete }; }, check: async entry => { if (!enabled) return null; @@ -101,7 +109,7 @@ export function registerDeferredReceiptWork( : kind === "deferred" && current.status === "deferred_issue_execution" && !current.interruptedLegacy && (current.payload?.queuedCommentInterrupt || current.payload?.executionWait) ? current.updatedAt.getTime() + 30_000 : null; - if (due !== null && due > Date.now()) return { ...entry, dueAt: due }; + if (due !== null && due > Date.now()) return { ...entry, dueAt: due, immediate: false }; } const errors: unknown[] = []; let next: number | null = null; @@ -117,7 +125,7 @@ export function registerDeferredReceiptWork( catch (error) { failed(error); } } if (errors.length) throw new AggregateError(errors, `Deferred receipt ${entry.id} could not settle`); - return next === null ? null : { ...entry, dueAt: next }; + return next === null ? null : { ...entry, dueAt: next, immediate: false }; }, }, canRun); } diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index cdd69a6801e..6ce95086ad0 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -18,6 +18,8 @@ export function registerScopedDeadlineWork( decodeScope(hint: string): Scope | undefined; matches(entry: Entry, scope?: Scope): boolean; load(scope?: Scope, signal?: AbortSignal): Promise; + // Owners may retain pending queue priority while refreshing eligibility. + mergeEntry?(pending: Entry, loaded: Entry): Entry; check(entry: Entry, signal?: AbortSignal): Promise; }, canRun: () => boolean, @@ -90,7 +92,8 @@ export function registerScopedDeadlineWork( const loaded = await options.load(scope, signal); const result = Array.isArray(loaded) ? { rows: loaded, complete: true } : loaded; for (const entry of result.rows ?? []) { - entries.set(entry.id, entry); + const pending = entries.get(entry.id); + entries.set(entry.id, pending && options.mergeEntry ? options.mergeEntry(pending, entry) : entry); for (const [other, progress] of scans) if (options.matches(entry, options.decodeScope(other))) progress.seen.add(entry.id); } if (!result.complete) { dirty.add(hint); continue; }