From 77f85ebc0f80bc3a5e0c6047cc62062efeaab43a Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 07:36:21 -0500 Subject: [PATCH] fix(server): signal chat milestones from retry and recovery transactions Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 10 +++ ...26-10-25-chat-run-milestone-sleep-audit.md | 38 +++++++++++ .../__tests__/chat-run-milestone-work.test.ts | 67 ++++++++++++++++++- .../modules/run-dispatch/adapters/postgres.ts | 7 +- .../modules/wake-queue/adapters/postgres.ts | 2 + 5 files changed, 122 insertions(+), 2 deletions(-) diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 4d1631825e4..669e89a8274 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1432,3 +1432,13 @@ The audit records the initial fixture FK failure, exact logs, remaining raw cancellation/promotion/receipt writers, and native reviewer acceptance gap. Periodic projection remains until producer coverage is complete; no new PR or full-goal completion is claimed. + + +Slice 28 retry/recovery checkpoint: due retry promotion, suppressed retry and +stale-run cancellation, deferred wake promotion and immediate recovery admission +now signal from their existing transactions. Five real missing-notice regressions +were reproduced before repair; all 166 selected worker/module/policy cases and +server production types pass. The slice audit preserves the initial invalid +staleness fixture and corrected causal evidence. Periodic projection remains +until all supported producers are covered. Remaining original inventory, open +PR gates/findings and final master reconciliation are unchanged. diff --git a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md index 49180b32c28..0687f5e1cae 100644 --- a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md @@ -397,3 +397,41 @@ conversation/configuration, question claim/retry and monitor/semantic receipt writers from the dependency table remain. Production still uses the periodic combined projector; the new worker is not registered yet. No original inventory or PR-stack acceptance requirement is removed by this checkpoint. + + +## Retry promotion and recovery admission checkpoint + +The run-dispatch PostgreSQL adapter now signals before due-retry promotion, +suppressed-retry cancellation, and stale-run cancellation. Its existing locked +state/gate checks still own authorization and no-op exits. Promotion passes the +already-resolved issue identity to its private writer; no application port or +provider contract changes. Missing issue identity uses company reconstruction. +Repeated attempts after settlement do not schedule another milestone sweep. + +The wake-queue transaction now signals before deferred promotion inserts its +run and before immediate recovery inserts its wake/run pair. The existing issue +lock, deferred claim, run linkage, and execution lock remain atomic. Tests use +these real adapter ports; external host dependencies return fixture values and +no provider execution is started. Notice creation occurs only after commit. + +Evidence: + +- `/tmp/sleep28-run-dispatch-before.log`: both real retry paths committed their + expected status but failed the missing-notice assertion. The first stale-run + fixture failed earlier because chat comments intentionally bypass terminal + task status; no production policy was changed to accommodate that fixture. +- `/tmp/sleep28-staleness-corrected-before.log`: a max-turn continuation correctly + cancelled but its terminal notice was absent before the signal was added. +- `/tmp/sleep28-wake-recovery-before.log`: both deferred and immediate admission + created their queued runs but failed the missing-notice assertions. +- `/tmp/sleep28-recovery-producers-tests.log`: all 166 selected cases pass, + comprising 29 milestone worker cases and existing run-dispatch/wake-queue + application and run-dispatch policy suites. Earlier 97-case verification + overlaps this set and is not an additional unique count. +- `/tmp/sleep28-recovery-producers-typecheck.log`: server production types pass; + test files are excluded from that command. `git diff --check` passes. + +Queued-comment changes, raw queue cancellation and wake-receipt writers, native +review admission acceptance, and remaining dependency producers are still open. +The periodic combined projector remains in production; this checkpoint does +not claim slice publication, full-suite/PR acceptance, or instance sleep. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 24fbc433482..e10e597c616 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-work.test.ts @@ -10,7 +10,8 @@ import { chatRunMilestoneWork } from "../services/chat-run-milestone-work.js"; import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { questionResponseDeliveryService } from "../services/question-response-delivery.js"; -import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; +import { createRunDispatch } from "../modules/run-dispatch/index.js"; +import { createAdmissionTransactionScope, createPostgresWakeQueueAdapter, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js"; @@ -141,6 +142,70 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu ].sort()); await quiet(); }); + it.each(["promote", "suppress"] as const)("wakes when real retry dispatch chooses to %s", async action => { + const row = await run(f, { status: "scheduled_retry", runtimeMode: "legacy", scheduledRetryAt: new Date(Date.now() - 1000), scheduledRetryReason: "transient_error" }); + if (action === "suppress") await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); + await start(); await flush(); expect(await publications()).toEqual([]); await quiet(); + const dispatch = createRunDispatch(db); + const result = await dispatch.promoteScheduledRetry({ companyId: f.companyId, runId: row.id }); + expect(result.outcome).toBe(action === "promote" ? "promoted" : "gate_suppressed"); + await flush(); + expect((await publications()).map(publication => publication.idempotencyKey)).toContain( + `run:${row.id}:${action === "promote" ? "queued" : "failed"}:${f.endpointId}`, + ); + await quiet(); + expect((await dispatch.promoteScheduledRetry({ companyId: f.companyId, runId: row.id })).outcome).toBe("not_promoted"); + await quiet(); + }); + it("wakes when the real staleness gate cancels queued work", async () => { + const row = await run(f, { status: "queued", runtimeMode: "legacy", scheduledRetryReason: "max_turns_continuation" }); + await start(); await flush(); await quiet(); + await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); + const dispatch = createRunDispatch(db); + await dispatch.cancelStaleQueuedRun({ companyId: f.companyId, runId: row.id, expectedStatus: "queued" }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, row.id)))[0].status).toBe("cancelled"); + await flush(); + expect((await publications()).map(publication => publication.idempotencyKey)).toContain(`run:${row.id}:failed:${f.endpointId}`); + await quiet(); + await dispatch.cancelStaleQueuedRun({ companyId: f.companyId, runId: row.id, expectedStatus: "queued" }); + await quiet(); + }); + it.each(["deferred", "immediate"] as const)("wakes when real %s recovery admits chat work", async kind => { + const finishing = await run(f, { status: "succeeded", runtimeMode: "legacy", finishedAt: new Date(), + contextSnapshot: { issueId: f.issueId, source: "issue_assigned" }, + }); + const wakeId = randomUUID(); + if (kind === "deferred") await db.insert(agentWakeupRequests).values({ id: wakeId, companyId: f.companyId, agentId: f.agentId, + source: "on_demand", status: "deferred_issue_execution", payload: { issueId: f.issueId }, + }); + const adapter = createPostgresWakeQueueAdapter(db, { + resolveResponsibleUserId: async () => f.userId, + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), + resolveSessionBeforeForWakeup: async () => null, + }); + await start(); await flush(); expect(await publications()).toEqual([]); await quiet(); + let admittedId: string | undefined; + await adapter.withIssueExecutionLock({ companyId: f.companyId, runId: finishing.id, now: new Date() }, async (locked, ports) => { + const input = { companyId: f.companyId, issue: locked.primaryIssue, finishingRun: locked.run, + contextSnapshot: runValues().contextSnapshot, responsibleUserId: f.userId, sessionBefore: null, now: new Date(), + }; + const agent = { id: f.agentId, companyId: f.companyId, name: "Maya", invokable: true }; + if (kind === "deferred") { + expect(await ports.transaction.claimDeferredWakeForPromotion({ companyId: f.companyId, wakeId, now: new Date() })).toBe(true); + admittedId = (await ports.transaction.finalizePromotedWake({ ...input, wakeId, deferredAgent: agent, + reason: "issue_execution_promoted", source: "on_demand", triggerDetail: null, payload: {}, + })).id; + } else { + admittedId = (await ports.transaction.queueImmediateRecoveryRun({ ...input, recoveryAgent: agent, reason: "issue_execution_recovery" })).id; + } + expect(await publications()).toEqual([]); + return { outcome: { kind: "released" }, postCommitEffects: [] }; + }); + expect(admittedId).toBeDefined(); + await flush(); + expect((await publications()).map(publication => publication.idempotencyKey)).toContain(`run:${admittedId}:queued:${f.endpointId}`); + await quiet(); + }); async function queue() { const dependencies = createHeartbeatQueueTestDependencies(db); const [agent] = await db.select().from(agents).where(eq(agents.id, f.agentId)); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index e45cd770bba..558b353534e 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -26,6 +26,7 @@ import { budgetService, budgetServiceInTransaction, recheckAdmittedRunBudgetRead import { lockRunBudgetRecoveryOwnerInTransaction, recheckRunBudgetReservationInTransaction } from "../../../services/budget-reservations.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; import { isHeartbeatWakeOnDemandEnabled } from "../../../services/heartbeat-policy.js"; +import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { connectionContinuationPendingResponse } from "../../../services/satisfied-connection-intents.js"; import { collectDispositionRepairSourceState } from "../../../services/recovery/disposition-repair.js"; @@ -660,10 +661,12 @@ export function createPostgresRunDispatchAdapter( async function promoteDueRetryInTx( tx: Db, input: PromoteOrCancelDueRetryInput, + issueId: string | null, ): Promise< | { applied: true; run: HeartbeatRun; postCommitEffects: PostCommitEffect[] } | { applied: false } > { + await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: issueId ?? undefined }); const [row] = await tx .update(heartbeatRuns) .set({ @@ -705,6 +708,7 @@ export function createPostgresRunDispatchAdapter( tx: Db, input: CancelSuppressedRetryInput, ): Promise { + await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); const [row] = await tx .update(heartbeatRuns) .set({ @@ -873,7 +877,7 @@ export function createPostgresRunDispatchAdapter( runId: input.runId, companyId: input.companyId, now, - }); + }, issueId); return promoted.applied ? { outcome: { @@ -911,6 +915,7 @@ export function createPostgresRunDispatchAdapter( providerDispatchNotStarted = false, ): Promise { await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); + await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [row] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 44717127131..409d2440dfd 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -472,6 +472,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }, async finalizePromotedWake(input) { + await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issue.id }); const newRun = await tx .insert(heartbeatRuns) .values({ @@ -688,6 +689,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru sessionBefore, now, }) { + await notifyChatRunMilestones(tx, { companyId, issueId: issue.id }); const wakeupRequest = await tx .insert(agentWakeupRequests) .values({