From 0337779e6d9eab4ad3eba51ae87ffd959824674b Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 05:09:20 -0500 Subject: [PATCH] fix(server): give scheduled reassessment its startup ownership Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 11 +++++ ...-native-status-reassessment-sleep-audit.md | 36 ++++++++++++++++ .../native-status-arbiter-corpus.test.ts | 20 ++++++++- .../server-startup-feedback-export.test.ts | 10 ++++- server/src/index.ts | 4 +- server/src/services/heartbeat.ts | 3 ++ server/src/services/heartbeat/recovery.ts | 3 +- .../native-finalization-reconciler.ts | 16 ++++--- .../native-status-reassessment.ts | 43 +++++++++++++++---- 9 files changed, 127 insertions(+), 19 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 7dee3ce8faf..39e270867ec 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2578,3 +2578,14 @@ A permanent 103-item backlog test proves real settlement and SQL quiet; all 31 lifecycle cases and server types pass. Remaining producer/authority audit, startup integration, periodic removal, remote gates and the full original stack scope remain open. Nothing merged. + + +Slice 50 startup checkpoint: enabled maintenance now registers the reassessment +worker and periodic recovery skips its classification/decision path. The separate +historical supersession guard is preserved before workspace recovery. Two missing +integration regressions fail before the change; all 246 corpus/startup/lifecycle +cases, 32 existing finalization/cleanup consumer cases, and server types pass after. +The owner audit records the intentional separation of reassessment from independent +workspace settlement and the retained failure evidence. Remaining producer/authority +audit, full acceptance, publication, remote gates and all other original duties +remain incomplete. Nothing merged; the goal remains active. diff --git a/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md b/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md index a3e731e8c0e..dd41e9b0d51 100644 --- a/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md +++ b/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md @@ -890,3 +890,39 @@ retained history, and server typecheck passes: All handles are terminal. Startup registration/periodic removal, remaining producer/authority audit and remote acceptance still remain; the full stack objective and its earlier unresolved gates are not complete. + + +## Startup integration and periodic ownership, 2026-10-26 + +The worker now registers through heartbeat under enabled maintenance, with the +same warm-standby, idle-drain and stopped gates as its sibling owners. The shared +coordinator owns shutdown and restart reconstruction. The periodic orphan reaper +passes `skipNativeStatusReassessment`; explicit/manual and one-time startup +recovery retain the existing reassessment entry point. + +The old reassessment call also supplied a superseded-decision early return before +workspace recovery. Removing the whole call without preserving that guard would +allow obsolete history into unrelated recovery. The reassessment owner now shares +that pure authority predicate and exposes a narrow guard for the periodic path. +It makes no queries for the usual matching-decision case, and reads only selected +decision/assessment authority when IDs differ; it does not classify evidence or +create assessments. Independent workspace settlement and committed result replay +remain owned by finalization recovery. Reassessment's dormant outcome no longer +implicitly suppresses those separate duties in the periodic path; their own +ownership and physical-resource guards remain applicable. + +Before the wiring change, the new regressions failed because periodic recovery +completed the task itself and startup never registered the worker. Historical +supersession was verified with the new option both enabled and disabled. Evidence: +`/tmp/sleep50-startup-before.log/.status` (two failures, four passing selected cases). +After the repair, all 246 corpus/startup/lifecycle cases pass and server typecheck +passes: `/tmp/sleep50-startup-after.log/.status` and +`/tmp/sleep50-startup-types.log/.status`. The existing native finalization and +retained cleanup consumer suite passes all 32 cases in a dedicated database: +`/tmp/sleep50-startup-consumers.log/.status`. All command handles are terminal. + +This integrates only reassessment ownership. The aggregate orphan/finalization +sweep still runs for the other listed duties, so this is not whole-process SQL +quiet or host suspension. Remaining producer/authority audit, final acceptance, +publication and remote PR gates remain open, along with the older stack findings +and every remaining recurring-server obligation. No merge occurred. diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index f2a79ad2284..df32659eec4 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -2854,6 +2854,22 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }, 30_000, ); + it("leaves reassessment to its scheduled owner while periodic finalization continues", async () => { + const seeded = await seedAutomaticReview(); + await dismissAutomaticCompletionReviews(db, seeded.issueId); + const options = { skipNativeStatusReassessment: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true }; + const result = await reconcileNativeFinalizations(db, [seeded.runId], options); + expect(result).toEqual([expect.objectContaining({ runId: seeded.runId, phase: "committed" })]); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("in_review"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(1); + const work = nativeStatusReassessmentWork(db); + const [pending] = await work.load({ companyId, issueId: seeded.issueId, runId: seeded.runId }); + expect(pending).toBeDefined(); + expect(await work.check(pending)).toBeNull(); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("done"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2); + }); + describe("native status reassessment work", () => { const selectedScope = (seeded: { issueId: string; runId: string }) => ({ companyId, issueId: seeded.issueId, runId: seeded.runId }); @@ -3633,7 +3649,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { ]); }); - it("ignores historical committed finalizations superseded by a newer authoritative decision", async () => { + it.each([false, true])("ignores superseded historical finalizations with scheduled reassessment %s", async scheduled => { const fixture = corpus.fixtures.find((candidate) => candidate.mode === "native"); if (!fixture) throw new Error("native corpus fixture missing"); const seeded = await seedFixture(fixture); @@ -3695,7 +3711,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { decisionId: historicalDecision!.id, }).where(eq(nativeRunFinalizations.runId, seeded.runId)); - await expect(reconcileNativeFinalizations(db, [seeded.runId])).resolves.toEqual([]); + await expect(reconcileNativeFinalizations(db, [seeded.runId], { skipNativeStatusReassessment: scheduled })).resolves.toEqual([]); await expect(db.select({ decisionId: nativeRunFinalizations.decisionId, }).from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seeded.runId))).resolves.toEqual([ diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 0059c660ced..78da1c65c81 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() })), + registerNativeStatusReassessmentWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeReviewResponseWork: 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() })), @@ -560,6 +561,13 @@ describe("startServer feedback export wiring", () => { } }); + it.each([false, true])("registers native reassessment only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerNativeStatusReassessmentWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerNativeStatusReassessmentWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers governed-response recovery only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -980,7 +988,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, skipNativeResponsePresentation: true, skipNativeReviewResponse: 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, skipNativeReviewResponse: true, skipNativeStatusReassessment: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 22bfe20fcb9..78e0b7ad1e3 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1778,6 +1778,8 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + if (config.heartbeatSchedulerEnabled) await heartbeat.registerNativeStatusReassessmentWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerNativeReviewResponseWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) { @@ -1927,7 +1929,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, skipNativeResponsePresentation: true, skipNativeReviewResponse: 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, skipNativeReviewResponse: true, skipNativeStatusReassessment: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 737c6bd8b0e..7857893de22 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { nativeStatusReassessmentWork } from "./native-runtime/native-status-reassessment-work.js"; import { registerNativeReviewResponseWork } from "./native-runtime/native-review-response-work.js"; import { registerNativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js"; import { registerNativePendingEffectWork } from "./native-runtime/native-pending-effect-work.js"; @@ -4069,6 +4070,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeStatusReassessmentWork: (coordinator: Parameters["register"]>[0], canRun: () => boolean) => + nativeStatusReassessmentWork(db).register(coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerNativeReviewResponseWork: (coordinator: Parameters[1], canRun: () => boolean) => registerNativeReviewResponseWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerNativeResponsePresentationWork: (coordinator: Parameters[1], canRun: () => boolean) => diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 79343eb99c2..22b8f61fd56 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; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: 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; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1939,6 +1939,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipNativePendingEffects: opts?.skipNativePendingEffects, skipNativeResponsePresentation: opts?.skipNativeResponsePresentation, skipNativeReviewResponse: opts?.skipNativeReviewResponse, + skipNativeStatusReassessment: opts?.skipNativeStatusReassessment, 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 07bbeaabbec..07f22b8f82d 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,4 +1,4 @@ -import { reassessNativeStatusFromSource } from "./native-status-reassessment.js"; +import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js"; import { resolveNativeReconciliationStatus } from "./native-reconciliation-status.js"; export { resolveNativeReconciliationStatus, type NativeReconciliationFacts } from "./native-reconciliation-status.js"; import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; @@ -518,6 +518,7 @@ type NativeReconciliationOptions = { skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; + skipNativeStatusReassessment?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -736,10 +737,15 @@ async function reconcileNativeFinalizationsPass( assessmentId: row.assessmentId, }); } - const reassessed = await reassessNativeStatusFromSource(db, { ...row, assessmentId: row.assessmentId }); - if (reassessed.kind !== "unchanged") { - if (reassessed.kind === "committed") results.push(reassessed.result); - continue; + const source = { ...row, assessmentId: row.assessmentId }; + if (options.skipNativeStatusReassessment) { + if (await nativeStatusReassessmentSourceIsSuperseded(db, source)) continue; + } else { + const reassessed = await reassessNativeStatusFromSource(db, source); + if (reassessed.kind !== "unchanged") { + if (reassessed.kind === "committed") results.push(reassessed.result); + continue; + } } } const barrier = await db.select({ status: workspaceOperations.status }) diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index d28072468fa..dad1a44d946 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -31,6 +31,39 @@ function record(value: unknown): Record { : {}; } +function isSupersededSource( + row: NativeStatusReassessmentSource, + decision: Record, + priorDecisionId: string | null | undefined, +): boolean { + return row.coordinatorPhase === "committed" + && row.decisionId !== null + && row.issueDecisionId !== null + && row.issueDecisionId !== row.decisionId + && !( + decision.statusAction === "preserve" + && priorDecisionId === row.issueDecisionId + ); +} + +// Workspace recovery still needs this authority guard when reassessment has +// its own scheduled owner. Avoid evidence classification on that path. +export async function nativeStatusReassessmentSourceIsSuperseded( + db: Db, + row: NativeStatusReassessmentSource, +): Promise { + if (row.coordinatorPhase !== "committed" || row.decisionId === null + || row.issueDecisionId === null || row.issueDecisionId === row.decisionId) return false; + const [decision] = await db.select({ decisionJson: statusDecisions.decisionJson }) + .from(statusDecisions).where(and(eq(statusDecisions.id, row.decisionId), + eq(statusDecisions.companyId, row.companyId), eq(statusDecisions.issueId, row.issueId))).limit(1); + if (!decision) throw new Error("native_committed_decision_missing"); + const [assessment] = await db.select({ priorDecisionId: workAssessments.priorDecisionId }) + .from(workAssessments).where(and(eq(workAssessments.id, row.assessmentId), + eq(workAssessments.companyId, row.companyId), eq(workAssessments.issueId, row.issueId))).limit(1); + return isSupersededSource(row, record(decision.decisionJson), assessment?.priorDecisionId); +} + export async function reassessNativeStatusFromSource( db: Db, row: NativeStatusReassessmentSource, @@ -72,15 +105,7 @@ export async function reassessNativeStatusFromSource( if (row.coordinatorPhase === "committed" && row.decisionId && !currentDecision) { throw new Error("native_committed_decision_missing"); } - const supersededCommittedDecision = row.coordinatorPhase === "committed" - && row.decisionId !== null - && row.issueDecisionId !== null - && row.issueDecisionId !== row.decisionId - && !( - currentDecisionJson.statusAction === "preserve" - && assessment?.priorDecisionId === row.issueDecisionId - ); - if (supersededCommittedDecision) return { kind: "superseded" }; + if (isSupersededSource(row, currentDecisionJson, assessment?.priorDecisionId)) return { kind: "superseded" }; const persistedProjectedVersion = Number(currentDecisionJson.projectedStatusVersion); const expectedProjectedVersion = currentDecision ? Number.isFinite(persistedProjectedVersion)