fix(server): give scheduled reassessment its startup ownership

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-26 05:09:20 -05:00
1 parent b8b8075268
commit 0337779e6d
9 files changed
+127 -19

No files matched your search

@@ -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.
@@ -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.
@@ -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([
@@ -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<void>((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 {
+3 -1
View File
@@ -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();
+3
View File
@@ -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<ReturnType<typeof nativeStatusReassessmentWork>["register"]>[0], canRun: () => boolean) =>
nativeStatusReassessmentWork(db).register(coordinator, () => canRun() && readTaskDrain(new Date()) === null),
registerNativeReviewResponseWork: (coordinator: Parameters<typeof registerNativeReviewResponseWork>[1], canRun: () => boolean) =>
registerNativeReviewResponseWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null),
registerNativeResponsePresentationWork: (coordinator: Parameters<typeof registerNativeResponsePresentationWork>[1], canRun: () => boolean) =>
+2 -1
View File
@@ -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,
@@ -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 })
@@ -31,6 +31,39 @@ function record(value: unknown): Record<string, unknown> {
: {};
}
function isSupersededSource(
row: NativeStatusReassessmentSource,
decision: Record<string, unknown>,
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<boolean> {
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)