From 79bf36cbf87def157790df8243e92a7cdcf3654d Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 00:08:45 -0500 Subject: [PATCH] fix: schedule accepted response recovery through delivery work Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 11 ++++++++ ...ative-response-presentation-sleep-audit.md | 26 ++++++++++++++++++- .../chat-channels.integration.test.ts | 20 ++++++++++++++ .../server-startup-feedback-export.test.ts | 14 +++++++++- server/src/index.ts | 11 +++++++- server/src/services/heartbeat.ts | 3 +++ server/src/services/heartbeat/recovery.ts | 3 ++- .../native-finalization-reconciler.ts | 5 +++- .../native-response-presentation-work.ts | 5 ++++ .../native-runtime/native-run-finalizer.ts | 3 ++- 10 files changed, 95 insertions(+), 6 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 0db605ede5c..67938973e3a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2339,3 +2339,14 @@ runs; server types pass. A real source-row lock verifies retry across drain. Earlier harness/test failures and the 10,000-unpresented-row discovery measurements are preserved in the response audit. No startup wiring, periodic removal, PR publication, remote acceptance, or whole-instance sleep is claimed yet. + + +## Slice 48 startup integration, 2026-10-26 + +The response worker now registers under enabled maintenance and releases its +subscription during shutdown. Periodic recovery skips both its direct repair and +already-committed finalizer replay; fresh finalization and explicit recovery stay +active. The startup suite passes 66 cases, and two database regressions confirm +periodic calls defer to the new worker. Workspace types pass. Full local tests +remain in progress with embedded-database initialization failures; build and +remote checks remain pending. The stack goal is still incomplete. diff --git a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md index 4a2f067ae37..84ba37c2306 100644 --- a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md +++ b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md @@ -1,6 +1,6 @@ # Native accepted-response presentation recovery -Status: worker and owner guards implemented locally; producer audit and startup/periodic integration incomplete. +Status: worker, producers, and startup/periodic integration implemented; full local and remote acceptance pending. Branch: `codex/sleep-48-native-response-presentation`. Parent: [#17690](https://github.com/paperclipai/paperclip/pull/17690), `56c3601ea605b333141e86e0a810ed790fb6ce48`. @@ -311,3 +311,27 @@ Startup/shutdown integration, removal of this duty's periodic call, final produc coverage review, full PR verification, and publication remain outstanding. The old backstop is still present. Master was refreshed to `6bdcb70244`; the full stack still needs reconciliation and its earlier failing/review gates remain open. + + +## Startup integration, 2026-10-26 + +The heartbeat facade now registers this worker with the existing task-drain gate. +Enabled maintenance awaits reconstruction during startup. Shutdown and listener +close release its live-authority subscription; coordinator shutdown retains +ownership of cancellation and in-flight work. + +The periodic orphan-recovery caller now passes `skipNativeResponsePresentation`. +The reconciler skips its direct repair call and passes the same option into +canonical finalizer replay. The finalizer skips only its already-committed response +repair in that mode. Fresh finalization still delivers immediately. Explicit +recovery and startup recovery keep their default behavior. Governed-review +presentation and other native finalization duties remain separate. + +All 66 startup tests pass, including disabled maintenance and close subscription +registration. Two real PostgreSQL regressions verify that both periodic entry +points leave the response unmaterialized, and the worker then creates exactly one +publication from the accepted result. Log: `/tmp/sleep48-wiring-tests.log`. +Workspace typecheck passes (`/tmp/sleep48-full-typecheck.log`). The full local suite +is still running and has reported embedded PostgreSQL initialization failures; +its final result and build status remain pending. No full-suite or remote check +acceptance is claimed at this checkpoint. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 46c545bd45b..d13c43499b5 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -72613,6 +72613,26 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }, ); + it.each(["reconciler", "finalizer"])("leaves accepted-response presentation to its worker during periodic %s replay", async caller => { + const context = await committedChatResponseRecoveryFixture("telegram"); + try { + if (caller === "reconciler") { + await reconcileNativeFinalizations(db, [context.runId], { skipNativeResponsePresentation: true }); + } else { + await finalizeNativeRun({ db, runId: context.runId, workspaceFinalizeStatus: "succeeded", + projectRunStatus: true, skipNativeResponsePresentation: true }); + } + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(0); + const work = nativeResponsePresentationWork(db); + const entries = await work.load({ companyId: context.fixture.companyId, runId: context.runId }); + expect(entries).toHaveLength(1); + await expect(work.check(entries[0])).resolves.toBeNull(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + await expect(context.repair()).resolves.toBe(false); + } finally { await context.service.shutdown(); } + }); + it("recovers accepted-response work on startup and leaves selected history quiet", async () => { const context = await committedChatResponseRecoveryFixture("telegram"); const errors: unknown[] = []; diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 0407f6c24a8..27ce690c836 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -108,6 +108,7 @@ const { reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerNativeResponsePresentationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), registerNativePendingEffectWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerUnsafeWorkspaceExportWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerAutomaticCompletionReviewWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -547,6 +548,17 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers response recovery and its shutdown subscription only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerNativeResponsePresentationWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) { + expect(heartbeatServiceMock.registerNativeResponsePresentationWork.mock.calls[0][1]()).toBe(true); + const worker = heartbeatServiceMock.registerNativeResponsePresentationWork.mock.results[0].value; + expect(fakeServer.on).toHaveBeenCalledWith("close", worker.stop); + } + }); + it.each([false, true])("registers native pending effects only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -960,7 +972,7 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index e2c2bff55d4..52c8ffdd0fb 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1367,6 +1367,7 @@ async function startServerWithDatabaseTeardown( hasPending: () => caseHandoffDeliveries.hasPending(), }); let unregisterTerminalWorkspaceProbe: (() => void) | undefined; + let unsubscribeNativeResponsePresentation: (() => void) | undefined; let unsubscribeNativeWorkspaceSyncRetention: (() => void) | undefined; const fastResponses = app.locals.fastResponses as ReturnType; const fastResponseWorker = deliveryWork.register(DELIVERY_QUEUES.fastResponse, { @@ -1777,6 +1778,13 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + if (config.heartbeatSchedulerEnabled) { + const worker = heartbeat.registerNativeResponsePresentationWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); + unsubscribeNativeResponsePresentation = worker.stop; + server.on("close", worker.stop); + await worker.ready; + } if (config.heartbeatSchedulerEnabled) await heartbeat.registerNativePendingEffectWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerUnsafeWorkspaceExportWork(deliveryWork, @@ -1917,7 +1925,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. trackHeartbeatSchedulerWork(heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); @@ -2110,6 +2118,7 @@ async function startServerWithDatabaseTeardown( await skillVersionRetention.dispose(); await skillSourceSync.stop(); unsubscribeChatCompletions(); + unsubscribeNativeResponsePresentation?.(); await deliveryWork.stop(); unregisterTerminalWorkspaceProbe?.(); unsubscribeNativeWorkspaceSyncRetention?.(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 04bb371f9cb..931a326436b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerNativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js"; import { registerNativePendingEffectWork } from "./native-runtime/native-pending-effect-work.js"; import { registerUnsafeWorkspaceExportWork } from "./native-runtime/native-workspace-export-recovery.js"; import { registerAutomaticCompletionReviewWork } from "./native-runtime/automatic-completion-reviews.js"; @@ -4067,6 +4068,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeResponsePresentationWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerNativeResponsePresentationWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerNativePendingEffectWork: (coordinator: Parameters[1], canRun: () => boolean) => registerNativePendingEffectWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerUnsafeWorkspaceExportWork: (coordinator: Parameters[1], canRun: () => boolean) => diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 8aa28f5dba6..b557e3ce10d 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1915,7 +1915,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1937,6 +1937,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews, skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports, skipNativePendingEffects: opts?.skipNativePendingEffects, + skipNativeResponsePresentation: opts?.skipNativeResponsePresentation, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 1955f8ab0df..b7905414006 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -638,6 +638,7 @@ type NativeReconciliationOptions = { skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; + skipNativeResponsePresentation?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -855,7 +856,7 @@ async function reconcileNativeFinalizationsPass( )).limit(1).then((entries) => entries[0] ?? null) : null; const currentDecisionJson = record(currentDecision?.decisionJson); - if (row.coordinatorPhase === "committed") { + if (row.coordinatorPhase === "committed" && !options.skipNativeResponsePresentation) { // A later decision can supersede this run's task status, but cannot // erase its accepted, still-authorized response. Repair presentation // before the status-only early return below, without rerunning work. @@ -1112,6 +1113,7 @@ async function reconcileNativeFinalizationsPass( runId: row.runId, workspaceFinalizeStatus, projectRunStatus: true, + skipNativeResponsePresentation: options.skipNativeResponsePresentation, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( record(row.runnerProfileJson).nativeWorkspaceSync, @@ -1170,6 +1172,7 @@ async function reconcileNativeFinalizationsPass( runId: row.runId, workspaceFinalizeStatus: barrier.status as "succeeded" | "failed", projectRunStatus: true, + skipNativeResponsePresentation: options.skipNativeResponsePresentation, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( record(row.runnerProfileJson).nativeWorkspaceSync, diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts index 243fd788ddf..55bf1cbb20a 100644 --- a/server/src/services/native-runtime/native-response-presentation-work.ts +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -7,6 +7,11 @@ import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wake import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; +export function registerNativeResponsePresentationWork(db: Db, + coordinator: ReturnType, canRun: () => boolean) { + return nativeResponsePresentationWork(db).register(coordinator, canRun); +} + type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; export function nativeResponsePresentationWork(db: Db) { diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 6429e083646..253d2499ecb 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1159,6 +1159,7 @@ export async function finalizeNativeRun(input: { workspaceFinalizeStatus: "succeeded" | "failed"; /** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */ projectRunStatus?: boolean; + skipNativeResponsePresentation?: boolean; /** Workspace-only replay must not consume the provider recovery budget. */ preserveProviderAttempt?: boolean; failpoint?: NativeStatusCommitFailpoint; @@ -1208,7 +1209,7 @@ export async function finalizeNativeRun(input: { await projectCommittedRun({ db: input.db, run, coordinator }); if (input.projectRunStatus && !presentationAlreadyMaterialized) await materializeCommittedReviewResponse(input.db, input.runId); - if (input.projectRunStatus) + if (input.projectRunStatus && !input.skipNativeResponsePresentation) await repairCommittedNativeChatResponse(input.db, { companyId: run.companyId, issueId: coordinator.issueId,