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 b756dea4aca..0060c2c2dca 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2470,3 +2470,14 @@ opt-in at the existing committer transaction; no outer transaction or early publication is added. Worker discovery, producer/lifecycle/cost evidence and periodic removal remain outstanding. No new PR or merge is claimed; all original remaining duties and older stack gates remain open. + +Slice 50 scoped-worker checkpoint: discovery/check/register now use the shared +scheduler, with distinct dormant and retryable outcomes and exact saved deadlines. +A real retirement missed wake and a near-term deadline delay were reproduced and +fixed. Final local verification passes 116 status/worker cases plus 187 governed +response consumer cases; server types pass. Driver SQL capture proves quiet after +settlement and during drain for the tested path. Initial history cost measurements +and their synthetic-fixture limits are recorded in the owner audit. The worker is +not registered in startup and periodic reassessment remains. Producer coverage, +concurrent authority audit, selected-current/backlog cost and full acceptance are +still pending; no slice 50 PR or merge, and the full original scope stays open. 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 52e5c6b8ba5..a7d1b4575e9 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 @@ -149,3 +149,63 @@ that state is released. Logs: /tmp/sleep50-guard-before.log, This is not worker acceptance. Candidate selection, dormant-versus-retry outcome handling, producer coverage, lifecycle/cost tests and periodic removal remain. No slice 50 PR has been published and no additional polling has been removed. + +## Scoped worker and first producer + +A local, not-yet-registered worker now discovers the three existing activation +conditions with company/task/run/agent scopes and 100-row keyset pages. Candidate +SQL excludes superseded decisions, held native ownership, unknown finalization +leases and old pending effects. It is discovery, not mutation authority: check +reloads the source and the committer still applies the locked guard. Future retry +and live lease deadlines retain their maximum. A commit race requests a retry; +unchanged evidence or a dormant review prerequisite does not keep recurring work +scheduled. The shared scoped-deadline helper owns coalescing, fairness, admission +and retry bookkeeping. No new queue table is added. + +Nine initial discovery/owner tests pass for retired reviews, changed evidence, +authoritative task changes, future retries/leases, insufficient evidence, +ownership exclusions and pre-aborted discovery. A tenth regression exposed a +newly saved five-second deadline being pushed out to generic 30-second backoff. +The worker now returns that saved future deadline exactly; backoff applies only +when the source is still immediately eligible after a commit race. The complete +113-case status corpus and server types pass. Evidence: +/tmp/sleep50-worker-first.log, /tmp/sleep50-deadline-before.log, +/tmp/sleep50-worker-after.log and /tmp/sleep50-worker-after-types.log. + +An isolated lifecycle test reproduced a real missed wake: after the worker went +quiet, automatic review retirement left the task in review. Retirement now +signals the exact task/run inside its existing transaction, before changing the +card. The common native-finalization signal also includes reassessment, retaining +its supplied scope. All three lifecycle tests pass: real retirement after idle, +first retirement during drain followed by resume, and a saved future retry across +shutdown/restart without early dispatch. Driver-level SQL capture verifies no SQL +across two former polling minutes after completion or while drain is paused. +The existing governed-response suite passes all 187 cases after the shared +notification change. Logs: /tmp/sleep50-lifecycle-before.log, +/tmp/sleep50-lifecycle-after.log, /tmp/sleep50-lifecycle-types.log and +/tmp/sleep50-worker-responses.log. + +Initial read-only query measurement reused slice 49's 10,000 cloned history rows. +Those copies share a task and their decisions are superseded for reassessment; +this is not proof of 10,000 independently selected current assessments or an +outstanding reassessment backlog. Startup returned 62 candidates from other +retained history in one query (305 ms cold wall time; warm EXPLAIN execution about +47 ms). Company/task scopes returned none in 34/20 ms, exact run in 2.6 ms. +Plans and logs: /tmp/sleep50-query-{startup,company,issue,run}.plan and +/tmp/sleep50-query-plan.log. Selected-current history, backlog cost and cancellation +between pages still need measurement before enabling the worker. + +Remaining work is substantial: work-product and alternate writers, task changes, +classifier inputs while a retired review is dormant, current-contract/workspace/ +governance and ownership transitions, further transaction/lifecycle cases, +startup registration and periodic removal. Existing retirement prerequisite reads +precede the committer; audit their serialization with producer owners before +claiming concurrent authority coverage. In particular, completion-contract +creation uses its own advisory lock rather than the status committer's issue +lock. This checkpoint is not a claim that every business-fact race is covered. +The existing periodic reassessment path remains active and slice 50 has no PR. + +Final current-source local rerun passes 116 tests across the status corpus and +isolated worker lifecycle files (/tmp/sleep50-final-local.log). Together with the +187 governed-response cases this is 303 distinct passing cases, not whole-workspace +or remote PR acceptance. The last server typecheck also passes. diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index faa03542dc1..108c8964c1b 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -1,3 +1,4 @@ +import { nativeStatusReassessmentWork } from "../services/native-runtime/native-status-reassessment-work.js"; import { reassessNativeStatusFromSource } from "../services/native-runtime/native-status-reassessment.js"; import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; import { registerNativePendingEffectWork } from "../services/native-runtime/native-pending-effect-work.js"; @@ -2488,7 +2489,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { return transaction(callback, options); }) as typeof db.transaction); try { - expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toEqual({ kind: "deferred" }); + expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toEqual({ kind: "deferred", retry: true }); } finally { intercept.mockRestore(); } 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); @@ -2514,6 +2515,110 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }, 30_000, ); + describe("native status reassessment work", () => { + const selectedScope = (seeded: { issueId: string; runId: string }) => ({ companyId, issueId: seeded.issueId, runId: seeded.runId }); + + it.each(["retired review", "new evidence", "task status"] as const)("discovers and settles %s without revisiting settled history", async trigger => { + const seeded = await seedAutomaticReview(); + const work = nativeStatusReassessmentWork(db), scope = selectedScope(seeded); + expect(await work.load(scope)).toEqual([]); + if (trigger === "retired review") await dismissAutomaticCompletionReviews(db, seeded.issueId); + if (trigger === "new evidence") { + const [assessment] = await db.select().from(workAssessments).where(eq(workAssessments.id, seeded.assessmentId)); + await db.update(workAssessments).set({ createdAt: new Date(Date.now() - 2000), + assessmentJson: { ...assessment.assessmentJson, allCriteriaSatisfied: false } }).where(eq(workAssessments.id, seeded.assessmentId)); + await db.update(issueWorkProducts).set({ updatedAt: new Date() }).where(eq(issueWorkProducts.id, seeded.workProductId)); + } + if (trigger === "task status") await db.update(issues).set({ status: "blocked", statusVersion: 2 }).where(eq(issues.id, seeded.issueId)); + const pending = await work.load(scope); + expect(pending).toMatchObject([{ id: seeded.runId, dueAt: 0 }]); + expect(await work.check(pending[0])).toBeNull(); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe(trigger === "task status" ? "blocked" : "done"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2); + expect(await work.load(scope)).toEqual([]); + expect(await work.check(pending[0])).toBeNull(); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2); + }); + + it("respects a saved retry and live lease, then reloads the released owner", async () => { + const seeded = await seedAutomaticReview(); + await dismissAutomaticCompletionReviews(db, seeded.issueId); + const work = nativeStatusReassessmentWork(db), scope = selectedScope(seeded); + const future = new Date(Date.now() + 60_000), lease = new Date(Date.now() + 120_000); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: future, leaseOwner: "finalizer", leaseExpiresAt: lease }) + .where(eq(nativeRunFinalizations.runId, seeded.runId)); + const [pending] = await work.load(scope); + expect(pending.dueAt).toBe(lease.getTime()); + expect(await work.check(pending)).toEqual(pending); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("in_review"); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, seeded.runId)); + expect(await work.check(pending)).toBeNull(); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("done"); + }); + + it("leaves insufficient evidence dormant until a later change satisfies the contract", async () => { + const seeded = await seedAutomaticReview(); + const work = nativeStatusReassessmentWork(db), scope = selectedScope(seeded); + const [assessment] = await db.select().from(workAssessments).where(eq(workAssessments.id, seeded.assessmentId)); + await db.update(workAssessments).set({ createdAt: new Date(Date.now() - 2000), + assessmentJson: { ...assessment.assessmentJson, allCriteriaSatisfied: false } }).where(eq(workAssessments.id, seeded.assessmentId)); + await db.update(issueWorkProducts).set({ reviewState: "changes_requested", updatedAt: new Date() }) + .where(eq(issueWorkProducts.id, seeded.workProductId)); + const [pending] = await work.load(scope); + expect(pending).toBeDefined(); + expect(await work.check(pending)).toBeNull(); + 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); + await db.update(issueWorkProducts).set({ reviewState: "approved", updatedAt: new Date() }) + .where(eq(issueWorkProducts.id, seeded.workProductId)); + const [changed] = await work.load(scope); + expect(await work.check(changed)).toBeNull(); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("done"); + }); + + it.each(["unknown lease", "pending effect", "ownership"] as const)("waits for the %s owner to release its obligation", async held => { + const seeded = await seedAutomaticReview(); + await dismissAutomaticCompletionReviews(db, seeded.issueId); + const work = nativeStatusReassessmentWork(db), scope = selectedScope(seeded); + const [pending] = await work.load(scope); + if (held === "unknown lease") await db.update(nativeRunFinalizations).set({ leaseOwner: "unknown", leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, seeded.runId)); + if (held === "pending effect") await db.update(statusDecisionEffects).set({ deliveryState: "pending" }) + .where(eq(statusDecisionEffects.decisionId, seeded.decision.id)); + if (held === "ownership") await db.update(heartbeatRuns).set({ status: "running", errorCode: "native_execution_ownership_unverified", nativePhase: "terminal_failure" }) + .where(eq(heartbeatRuns.id, seeded.runId)); + expect(await work.load(scope)).toEqual([]); + expect(await work.check(pending)).toBeNull(); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("in_review"); + }); + + it("keeps a newly saved near-term deadline after a competing commit", async () => { + const seeded = await seedAutomaticReview(); + await dismissAutomaticCompletionReviews(db, seeded.issueId); + const work = nativeStatusReassessmentWork(db), scope = selectedScope(seeded); + const [pending] = await work.load(scope); + const future = new Date(Date.now() + 5000); + const transaction = db.transaction.bind(db); + const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + await db.update(nativeRunFinalizations).set({ nextAttemptAt: future }).where(eq(nativeRunFinalizations.runId, seeded.runId)); + return transaction(callback, options); + }) as typeof db.transaction); + try { expect(await work.check(pending)).toEqual({ ...pending, dueAt: future.getTime() }); } + finally { intercept.mockRestore(); } + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("in_review"); + }); + + it("does not query after discovery was cancelled", async () => { + const controller = new AbortController(); controller.abort(); + const select = vi.spyOn(db, "select"); + try { + await expect(nativeStatusReassessmentWork(db).load(undefined, controller.signal)).rejects.toThrow("discovery stopped"); + expect(select).not.toHaveBeenCalled(); + } finally { select.mockRestore(); } + }); + }); + it("retires a proven automatic review and applies the current successful completion exactly once", async () => { const seeded = await seedAutomaticReview(); // The persisted result is done, while its old assessment/decision required a review. diff --git a/server/src/__tests__/native-status-reassessment-work.test.ts b/server/src/__tests__/native-status-reassessment-work.test.ts new file mode 100644 index 00000000000..5212f871fc1 --- /dev/null +++ b/server/src/__tests__/native-status-reassessment-work.test.ts @@ -0,0 +1,143 @@ +import { randomUUID } from "node:crypto"; +import { setTimeout as delay } from "node:timers/promises"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { agents, companies, completionContracts, createDb, heartbeatRuns, issues, issueWorkProducts, + nativeRunFinalizations, nativeRunResults, statusDecisions, workAssessments, workspaceOperations } from "@paperclipai/db"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { nativeStatusReassessmentWork } from "../services/native-runtime/native-status-reassessment-work.js"; +import { dismissAutomaticCompletionReviews } from "../services/native-runtime/automatic-completion-reviews.js"; +import { commitNativeStatusDecision } from "../services/native-runtime/status-decision-committer.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "../services/native-runtime/status-arbiter.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +describe("native reassessment worker lifecycle", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let coordinator: ReturnType | undefined; + const companyId = randomUUID(), agentId = randomUUID(); + let admitted = true; + const errors: unknown[] = []; + beforeAll(async () => { + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STATUS_REASSESSMENT_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-reassessment-"); + db = createDb(external ?? temporary!.connectionString); + await db.insert(companies).values({ id: companyId, name: "Reassessment lifecycle", issuePrefix: `RS${randomUUID().slice(0, 8)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Reassessment agent", adapterType: "codex_local", status: "idle" }); + }, 30_000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + beforeEach(() => { + admitted = true; errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + expect(errors).toEqual([]); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + await nativeStatusReassessmentWork(db).register(coordinator, () => admitted).ready; + } + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 15_000 }); + } + async function quiet() { + const original = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_id, query) => { queries.push(query); }; + try { + await vi.advanceTimersByTimeAsync(120_000); + await delay(10); + expect(queries).toEqual([]); + } finally { db.$client.options.debug = original; } + } + async function status(issueId: string) { + return (await db.select({ status: issues.status }).from(issues).where(eq(issues.id, issueId)))[0].status; + } + async function seedReview() { + const issueId = randomUUID(), runId = randomUUID(), resultId = randomUUID(), contractId = randomUUID(); + const assessmentId = randomUUID(), workProductId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Persisted native result", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: issueId, completionContractId: contractId, contextSnapshot: { issueId }, finishedAt: new Date() }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "phase6-v1", risk: "low", + completionAuthority: "agent_claim_policy", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "test-v1", criteria: [{ id: "objective", requirement: "Verified artifact" }] }, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "reassessment-test" }); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, + serverFingerprint: resultId, schemaStatus: "accepted", canonicalSha256: resultId, + resultJson: { result: { reportedWorkDisposition: "done", summary: "Completed", attentionRequests: [], + completionClaim: { contractRevision: "test-v1", objectiveSatisfied: true, remainingWork: [], + criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: [`work_product:${workProductId}`] }] }, + verification: [{ commandOrCheck: "artifact", status: "passed", artifactRef: `work_product:${workProductId}` }], + }, terminal: { runTerminalState: "succeeded" } } }); + await db.insert(issueWorkProducts).values({ id: workProductId, companyId, issueId, type: "artifact", provider: "paperclip", + title: "Verified artifact", status: "ready_for_review", reviewState: "approved", + createdAt: new Date(Date.now() - 1000), updatedAt: new Date(Date.now() - 1000) }); + await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId, runId, contractId, resultId, + triggerKind: "native_result", triggerActorCompanyId: companyId, priorIssueStatus: "in_progress", priorStatusVersion: 0, + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, inputDigest: assessmentId, + assessmentJson: { reportedDisposition: "needs_review", attentionRequests: [], allCriteriaSatisfied: false } }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "assessing", resultId, assessmentId }); + const committed = await commitNativeStatusDecision({ db, companyId, issueId, runId, assessmentId, + priorStatus: "in_progress", priorStatusVersion: 0, priorDecisionId: null, + decision: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "in_review", toStatus: "in_review", + reasonCode: "completion_claim_incomplete", unblockDescriptor: null, + effects: [{ kind: "bind_reviewer", ownerUserId: null, + prompt: "Review the persisted native-run evidence and confirm whether this issue may be completed." }] } }); + await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: runId, issueId, phase: "workspace_finalize", + status: "succeeded", exitCode: 0, cwd: process.cwd(), finishedAt: new Date() }); + return { issueId, runId, decisionId: committed.decision.id }; + } + + it("retains the first review-retirement wake through idle drain", async () => { + const seeded = await seedReview(); + await start(); await flush(); await quiet(); + admitted = false; + await dismissAutomaticCompletionReviews(db, seeded.issueId); + await quiet(); + expect(await status(seeded.issueId)).toBe("in_review"); + admitted = true; + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(await status(seeded.issueId)).toBe("done"); + await quiet(); + }); + + it("reconstructs the saved retry after shutdown and never dispatches it early", async () => { + const seeded = await seedReview(); + await dismissAutomaticCompletionReviews(db, seeded.issueId); + const future = new Date(Date.now() + 60_000); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: future }).where(eq(nativeRunFinalizations.runId, seeded.runId)); + await start(); + expect(coordinator!.nextWakeAt()).toBe(future.getTime()); + await vi.advanceTimersByTimeAsync(20_000); + expect(await status(seeded.issueId)).toBe("in_review"); + await coordinator!.stop(); coordinator = undefined; + await start(); + expect(coordinator!.nextWakeAt()).toBe(future.getTime()); + await vi.advanceTimersByTimeAsync(Math.max(0, future.getTime() - Date.now())); await flush(); + expect(await status(seeded.issueId)).toBe("done"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2); + await quiet(); + }); + + it("wakes from real review retirement after idle and becomes SQL quiet with completed history", async () => { + const seeded = await seedReview(); + await start(); await flush(); await quiet(); + expect(await status(seeded.issueId)).toBe("in_review"); + await dismissAutomaticCompletionReviews(db, seeded.issueId); + await flush(); + expect(await status(seeded.issueId)).toBe("done"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2); + await quiet(); + }); +}); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index f62a357aa93..b419d4a200f 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + nativeStatusReassessment: "native-status-reassessment", nativeReviewResponse: "native-review-response", nativeResponsePresentation: "native-response-presentation", nativePendingEffects: "native-pending-effects", diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index 9f6a1dfe27f..f997eb845f2 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -5,6 +6,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type NativeFinalizationWorkScope = { companyId: string; runId?: string; issueId?: string }; export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { + await notifyNativeStatusReassessment(transaction, scope); await notifyNativeResponsePresentation(transaction, scope); await notifyNativeReviewResponse(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativePendingEffects, diff --git a/server/src/services/native-runtime/automatic-completion-reviews.ts b/server/src/services/native-runtime/automatic-completion-reviews.ts index 9f7eea8c326..42be62845cc 100644 --- a/server/src/services/native-runtime/automatic-completion-reviews.ts +++ b/server/src/services/native-runtime/automatic-completion-reviews.ts @@ -1,3 +1,4 @@ +import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import type { AutomaticCompletionReviewScope } from "../automatic-completion-review-notifications.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; @@ -138,6 +139,7 @@ async function dismissCandidates(db: Db, candidates: Awaited'projectedStatusVersion') = 'number' + then (${statusDecisions.decisionJson}->>'projectedStatusVersion')::numeric + else ${statusDecisions.decisionVersion} - case when ${statusDecisions.decisionJson}->>'statusAction' = 'preserve' then 1 else 0 end + end`; + const matchesDecision = sql`coalesce(${issues.lastStatusDecisionId} = ${nativeRunFinalizations.decisionId} + and ${statusDecisions.assessmentId} = ${workAssessments.id} and ${statusDecisions.toStatus} = ${issues.status} + and ${projectedVersion} = ${issues.statusVersion}, false)`; + const statusChanged = and(not(matchesDecision), or(ne(workAssessments.priorIssueStatus, issues.status), + ne(workAssessments.priorStatusVersion, issues.statusVersion))); + const evidenceChanged = exists(db.select({ id: issueWorkProducts.id }).from(issueWorkProducts).where(and( + eq(issueWorkProducts.companyId, nativeRunFinalizations.companyId), eq(issueWorkProducts.issueId, nativeRunFinalizations.issueId), + or(gt(issueWorkProducts.createdAt, workAssessments.createdAt), gt(issueWorkProducts.updatedAt, workAssessments.createdAt)), + ))); + const retiredReview = exists(db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, nativeRunFinalizations.companyId), eq(issueThreadInteractions.issueId, nativeRunFinalizations.issueId), + eq(issueThreadInteractions.status, "cancelled"), sql`${issueThreadInteractions.result}->>'reason' = 'automatic_completion_review_removed'`, + sql`(${issueThreadInteractions.payload}->'target'->>'revisionId' = ${statusDecisions.id}::text + or ${statusDecisions.decisionJson}->'effects' @> jsonb_build_array(jsonb_build_object('gate', + jsonb_build_object('kind', 'interaction', 'id', ${issueThreadInteractions.id}::text))))`, + ))); + return db.select({ runId: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + issueId: issues.id, issueStatus: issues.status, issueStatusVersion: issues.statusVersion, + issueDecisionId: issues.lastStatusDecisionId, coordinatorPhase: nativeRunFinalizations.phase, + assessmentId: workAssessments.id, decisionId: nativeRunFinalizations.decisionId, contextSnapshot: heartbeatRuns.contextSnapshot, + nextAttemptAt: nativeRunFinalizations.nextAttemptAt, leaseOwner: nativeRunFinalizations.leaseOwner, + leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) + .innerJoin(workAssessments, and(eq(workAssessments.id, nativeRunFinalizations.assessmentId), + eq(workAssessments.companyId, nativeRunFinalizations.companyId), eq(workAssessments.issueId, nativeRunFinalizations.issueId), + eq(workAssessments.runId, nativeRunFinalizations.runId))) + .leftJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId), + eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId))) + .where(and(eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition(), + ne(nativeRunFinalizations.phase, "terminal_failure"), isNotNull(nativeRunFinalizations.resultId), + or(isNull(nativeRunFinalizations.leaseOwner), isNotNull(nativeRunFinalizations.leaseExpiresAt)), + sql`not coalesce(${nativeRunFinalizations.phase} = 'committed' + and ${nativeRunFinalizations.decisionId} is not null and ${issues.lastStatusDecisionId} is not null + and ${issues.lastStatusDecisionId} <> ${nativeRunFinalizations.decisionId} + and not coalesce(${statusDecisions.decisionJson}->>'statusAction' = 'preserve' + and ${workAssessments.priorDecisionId} = ${issues.lastStatusDecisionId}, false), false)`, + or(statusChanged, evidenceChanged, and(matchesDecision, retiredReview)), + notExists(db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, nativeRunFinalizations.companyId), eq(statusDecisionEffects.issueId, nativeRunFinalizations.issueId), + eq(statusDecisionEffects.decisionId, nativeRunFinalizations.decisionId), inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + ))), + scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, + scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined, + scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined, + after ? gt(nativeRunFinalizations.runId, after) : undefined, + runId ? eq(nativeRunFinalizations.runId, runId) : undefined, + )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); + } + function entry(row: Awaited>[number]): Entry { + return { id: row.runId, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, + dueAt: Math.max(row.nextAttemptAt?.getTime() ?? 0, row.leaseOwner !== null ? row.leaseExpiresAt!.getTime() : 0) }; + } + async function load(scope?: NativeStatusReassessmentScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await candidates(scope, after); + entries.push(...rows.map(entry)); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].runId; + } + throw new Error("Native status reassessment discovery stopped"); + } + async function check(current: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Native status reassessment stopped"); + const scope = { companyId: current.companyId, issueId: current.issueId, runId: current.id }; + const [source] = await candidates(scope); + if (!source) return null; + const due = entry(source); + if (due.dueAt > Date.now()) return due; + if (signal?.aborted) throw new Error("Native status reassessment stopped"); + const outcome = await reassessNativeStatusFromSource(db, source, { onlyWhenDue: true }); + if (outcome.kind !== "deferred" || !outcome.retry) return null; + const [remaining] = await candidates(scope); + if (!remaining) return null; + const retry = entry(remaining); + return retry.dueAt > Date.now() ? retry : { ...retry, dueAt: Date.now() + 30_000 }; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeStatusReassessment, + decodeScope: (hint: string): NativeStatusReassessmentScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (item: Entry, scope?: NativeStatusReassessmentScope) => !scope || item.companyId === scope.companyId + && (!scope.issueId || item.issueId === scope.issueId) && (!scope.runId || item.id === scope.runId) + && (!scope.agentId || item.agentId === scope.agentId), + load: (scope, signal) => load(scope, signal, canRun), + check: (item, signal) => { + if (!canRun()) throw new Error("Native status reassessment paused"); + return check(item, signal); + }, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index 66697202336..eacb2da9588 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -18,7 +18,7 @@ export type NativeStatusReassessmentSource = { contextSnapshot: Record | null; }; -type Outcome = { kind: "unchanged" | "superseded" | "deferred" } | { +type Outcome = { kind: "unchanged" | "superseded" } | { kind: "deferred"; retry?: boolean } | { kind: "committed"; result: { runId: string; phase: "committed"; reconciliationAction: string | null; decisionId: string; reconciliationDecision: NativeStatusDecision }; @@ -215,7 +215,8 @@ export async function reassessNativeStatusFromSource( : undefined, }); } catch (error) { - if (error instanceof NativeStatusRaceError) return { kind: "deferred" }; + if (error instanceof NativeStatusRaceError) return options.onlyWhenDue + ? { kind: "deferred", retry: true } : { kind: "deferred" }; throw error; } return { kind: "committed", result: { diff --git a/server/src/services/native-status-reassessment-notifications.ts b/server/src/services/native-status-reassessment-notifications.ts new file mode 100644 index 00000000000..fb845523bb4 --- /dev/null +++ b/server/src/services/native-status-reassessment-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; + +export function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); +}