From d09253f7f510ca142875b3d818a7191501a7b0a3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 01:43:13 -0500 Subject: [PATCH] fix(server): schedule governed review response recovery Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 15 ++ ...26-governed-review-response-sleep-audit.md | 67 +++++++ packages/db/src/client.ts | 1 + .../src/migrations/0401_numerous_vulcan.sql | 1 + ...{0396_snapshot.json => 0401_snapshot.json} | 172 +++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/issue_comments.ts | 4 + .../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 +- server/src/services/issues.ts | 4 +- .../external-chat-wait.integration.test.ts | 185 ++++++++++++++++++ .../native-finalization-reconciler.ts | 6 +- .../native-review-response-work.ts | 5 + .../native-runtime/native-run-finalizer.ts | 5 +- 16 files changed, 483 insertions(+), 9 deletions(-) create mode 100644 packages/db/src/migrations/0401_numerous_vulcan.sql rename packages/db/src/migrations/meta/{0396_snapshot.json => 0401_snapshot.json} (99%) 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 70891ec8608..57c854ee05a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2410,3 +2410,18 @@ external pool-fixture and callback-narrowing failures and repairs are preserved in the owner audit. Execution-state-only restoration, real coordinator release, query cost, registration/periodic removal and final acceptance remain. No PR yet, no merge, and the full original stack scope remains open. + + +## Slice 49 startup and cost checkpoint + +Governed-review response recovery now owns both periodic repair entrances and +registers through heartbeat maintenance. All 255 owner/startup cases, workspace +typecheck and build pass. Pending-policy release, real coordinator retry, SQL-quiet +drain, contention/resume and shutdown/restart are covered. A run-first concurrent +receipt index removes the measured quadratic history lookup (about 3,000 ms to +45 ms on 10,000 selected historical responses). The 10,000 outstanding-item +fixture remains proportional: 101 queries, 2.25 seconds, about 2.05 MB; it stops +between pages on drain. Fresh migration and interrupted/concurrent replay pass. +The audit preserves the initial query, migration and test-clock failures. Full +local tests are running; remote acceptance, older stack findings, master +reconciliation and every remaining original duty are still open. Nothing merged. diff --git a/doc/plans/2026-10-26-governed-review-response-sleep-audit.md b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md index f3e11b286f4..5c790032235 100644 --- a/doc/plans/2026-10-26-governed-review-response-sleep-audit.md +++ b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md @@ -294,3 +294,70 @@ Still inspect execution-state-only transitions (pending policy state can block a response without changing task status), prove real coordinator lease/retry release, and finish candidate query-cost/lifecycle/wiring acceptance. The worker remains unregistered and both periodic entrances remain intact. + + +## Policy release and discovery-cost checkpoint + +A pending execution policy can block presentation without changing task status. +The regression failed before the issue mutation boundary signaled pending-to-clear +transitions. All 182 owner cases now pass, including a real coordinator claim and +retry release; server types pass. Evidence: /tmp/sleep49-policy-before.log, +/tmp/sleep49-policy-after.log and /tmp/sleep49-policy-types.log. + +Discovery was measured against 10,000 cloned, already-presented governed response +rows plus retained test history. The clones preserve relational selection shape; +they are not valid authorization proofs and are never used to test publication. +Initial startup/company reconstruction took about 3,000 ms: receipt lookup scanned +roughly 4,795 unrelated comments per candidate. A company-first receipt index and +expression statistics did not reliably fix startup's plan. A run-first partial +receipt index alone reduced measured startup to 45 ms, company/task discovery to +33–35 ms and exact-run discovery to 1.5 ms. No expression statistics are retained. +The generated migration builds that index concurrently and preserves deleted +comments as receipts. These are local query measurements, not host sleep evidence. +Logs: /tmp/sleep49-query-plan.log, /tmp/sleep49-query-index-plan.log, +/tmp/sleep49-query-stats-plan.log, /tmp/sleep49-query-run-index-plan.log, +/tmp/sleep49-query-run-only-plan.log. The first root-level tsx invocation failed +because tsx is a server-package dependency; the corrected command uses its owner. + +Full outstanding-backlog cost, lifecycle coverage, startup registration and both +periodic entrance removals remain required before PR acceptance. + + +## Startup integration and lifecycle acceptance + +The worker now registers with enabled heartbeat maintenance and uses its existing +instance/task-drain gates. The shared coordinator owns notification teardown and +shutdown. Both periodic repair entrances defer to this worker: direct reconciler +repair and already-committed finalizer replay. Fresh finalization and explicit +recovery keep their behavior. Three regression cases failed before this wiring +(two periodic entrances and enabled startup registration); all now pass. + +The combined owner/startup selection passes all 255 cases. Actual SQL capture +shows no passive SQL during idle drain or after shutdown. Real approval cancellation +reactivates dormant work through drain; endpoint-lock contention retains a retry +that succeeds after resume; a real coordinator retry survives shutdown and is +reconstructed without early publication. One initial drain test mixed a real timer +with a subsequently installed fake clock. Installing the clock before coordinator +creation repaired the fixture; all three new lifecycle cases then passed. +Evidence: /tmp/sleep49-drain.log, /tmp/sleep49-drain-v2.log, +/tmp/sleep49-wiring-before.log, /tmp/sleep49-wiring-after.log. + +Workspace typecheck and build pass. Migration generation and safety checks pass. +The first migration application exposed the missing top-level migration allowlist +entry; after registration, fresh application, invalid concurrent-index repair, +concurrent migrator replay and no-op replay all pass. Evidence: +/tmp/sleep49-migration-apply.log, /tmp/sleep49-migration-apply-v2.log, +/tmp/sleep49-migration-replay.log, /tmp/sleep49-workspace-types.log, +/tmp/sleep49-build.log. The five newest schema snapshots are retained. + +The synthetic 10,000-item unpresented cohort requires 101 candidate queries, +about 2.25 seconds and 2.05 MB of serialized entries. Discovery checks admission +and cancellation between pages: the drain experiment stops after two queries; +an already aborted signal executes none. This remains proportional startup or +legitimate broad-authority reconstruction, not a bounded total-memory claim. +Normal writes retain task/run/agent scope. Synthetic proof copies establish +selection cost only; genuine owner fixtures establish publication authority. +Evidence: /tmp/sleep49-backlog.log. + +Full local tests are running. Remote CI/review, earlier stack findings and master +reconciliation remain open. No merge or whole-instance sleep is claimed. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index a74fdb5d893..aeba4ffdcaf 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -526,6 +526,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0398_shocking_molecule_man.sql", "0399_jazzy_ben_grimm.sql", "0400_futuristic_earthquake.sql", + "0401_numerous_vulcan.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0401_numerous_vulcan.sql b/packages/db/src/migrations/0401_numerous_vulcan.sql new file mode 100644 index 00000000000..1a3a6438728 --- /dev/null +++ b/packages/db/src/migrations/0401_numerous_vulcan.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "issue_comments_review_response_receipt_idx" ON "issue_comments" USING btree ("created_by_run_id","company_id","issue_id","author_agent_id") WHERE "issue_comments"."author_type" = 'agent' AND "issue_comments"."metadata"->>'authorizationReason' = 'allow_chat_run_presentation'; diff --git a/packages/db/src/migrations/meta/0396_snapshot.json b/packages/db/src/migrations/meta/0401_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0396_snapshot.json rename to packages/db/src/migrations/meta/0401_snapshot.json index 0bd80702427..a2da91fc051 100644 --- a/packages/db/src/migrations/meta/0396_snapshot.json +++ b/packages/db/src/migrations/meta/0401_snapshot.json @@ -1,6 +1,6 @@ { - "id": "27bec2e7-b637-4317-aa3c-a457bcc709fe", - "prevId": "e9c56475-b2c8-4cb4-b7dd-89bfa645a93a", + "id": "1288008c-b412-41f9-a5db-3ce62fd3fd60", + "prevId": "d51b06de-ba83-4221-8ea1-863fd947f021", "version": "7", "dialect": "postgresql", "tables": { @@ -33671,6 +33671,56 @@ } }, "indexes": { + "heartbeat_runs_native_unpresented_response_idx": { + "name": "heartbeat_runs_native_unpresented_response_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "native_issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" in ('succeeded', 'failed', 'timed_out') AND \"heartbeat_runs\".\"finished_at\" is not null AND jsonb_typeof(\"heartbeat_runs\".\"result_json\" #> '{presentationDecision,commentId}') is distinct from 'string'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "heartbeat_runs_cancelled_workspace_preservation_idx": { + "name": "heartbeat_runs_cancelled_workspace_preservation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" = 'cancelled'\n AND \"heartbeat_runs\".\"runner_profile_json\"->'nativeWorkspaceSync'->>'state' = 'prepared'\n AND \"heartbeat_runs\".\"result_json\"->'nativeCancellation'->>'dispatchState' = 'acknowledged'\n AND (\"heartbeat_runs\".\"result_json\"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'\n AND \"heartbeat_runs\".\"result_json\"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_retained_recovery_adoption_idx": { "name": "heartbeat_runs_retained_recovery_adoption_idx", "columns": [ @@ -36353,6 +36403,40 @@ "method": "btree", "with": {} }, + "issue_comments_review_response_receipt_idx": { + "name": "issue_comments_review_response_receipt_idx", + "columns": [ + { + "expression": "created_by_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "author_agent_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"issue_comments\".\"author_type\" = 'agent' AND \"issue_comments\".\"metadata\"->>'authorizationReason' = 'allow_chat_run_presentation'", + "concurrently": false, + "method": "btree", + "with": {} + }, "issue_comments_company_idx": { "name": "issue_comments_company_idx", "columns": [ @@ -42852,6 +42936,62 @@ } }, "indexes": { + "native_run_finalizations_pending_effect_owner_idx": { + "name": "native_run_finalizations_pending_effect_owner_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"decision_id\" is not null AND \"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"phase\" <> 'terminal_failure'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_unsafe_export_recovery_idx": { + "name": "native_run_finalizations_unsafe_export_recovery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"phase\" = 'terminal_failure' AND \"native_run_finalizations\".\"failure_code\" = 'native_workspace_sync_out_unsafe_archive'", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ @@ -52038,6 +52178,34 @@ } }, "indexes": { + "status_decision_effects_pending_delivery_idx": { + "name": "status_decision_effects_pending_delivery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"status_decision_effects\".\"delivery_state\" in ('pending', 'failed')", + "concurrently": false, + "method": "btree", + "with": {} + }, "status_decision_effects_decision_ordinal_uq": { "name": "status_decision_effects_decision_ordinal_uq", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 3380415c237..1cfb8b818d5 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2794,6 +2794,13 @@ "when": 1792989084087, "tag": "0400_futuristic_earthquake", "breakpoints": true + }, + { + "idx": 401, + "version": "7", + "when": 1792996335782, + "tag": "0401_numerous_vulcan", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/issue_comments.ts b/packages/db/src/schema/issue_comments.ts index 8d17c216628..ba5d74007e4 100644 --- a/packages/db/src/schema/issue_comments.ts +++ b/packages/db/src/schema/issue_comments.ts @@ -5,6 +5,7 @@ import type { IssueCommentPresentation, SourceTrustMetadata, } from "@paperclipai/shared"; +import { sql } from "drizzle-orm"; import { pgTable, uuid, text, timestamp, index, jsonb, unique, integer } from "drizzle-orm/pg-core"; import { companies } from "./companies.js"; import { issues } from "./issues.js"; @@ -50,6 +51,9 @@ export const issueComments = pgTable( clientRequestUq: unique("issue_comments_client_request_uq").on(table.issueId, table.authorUserId, table.clientRequestId), companyIdUq: unique("issue_comments_company_id_uq").on(table.companyId, table.id), issueIdx: index("issue_comments_issue_idx").on(table.issueId), + reviewResponseReceiptIdx: index("issue_comments_review_response_receipt_idx").on( + table.createdByRunId, table.companyId, table.issueId, table.authorAgentId, + ).where(sql`${table.authorType} = 'agent' AND ${table.metadata}->>'authorizationReason' = 'allow_chat_run_presentation'`), companyIdx: index("issue_comments_company_idx").on(table.companyId), companyIssueCreatedAtIdx: index("issue_comments_company_issue_created_at_idx").on( table.companyId, diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 27ce690c836..0059c660ced 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() })), + 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() })), registerUnsafeWorkspaceExportWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -559,6 +560,13 @@ describe("startServer feedback export wiring", () => { } }); + 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(); + expect(heartbeatServiceMock.registerNativeReviewResponseWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerNativeReviewResponseWork.mock.calls[0][1]()).toBe(true); + }); + 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(); @@ -972,7 +980,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 }); + 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.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 52c8ffdd0fb..22bfe20fcb9 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.registerNativeReviewResponseWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) { const worker = heartbeat.registerNativeResponsePresentationWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); @@ -1925,7 +1927,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 }) + .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 }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 931a326436b..737c6bd8b0e 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +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"; import { registerUnsafeWorkspaceExportWork } from "./native-runtime/native-workspace-export-recovery.js"; @@ -4068,6 +4069,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeReviewResponseWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerNativeReviewResponseWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerNativeResponsePresentationWork: (coordinator: Parameters[1], canRun: () => boolean) => registerNativeResponsePresentationWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerNativePendingEffectWork: (coordinator: Parameters[1], canRun: () => boolean) => diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index b557e3ce10d..79343eb99c2 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 }) { + 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 }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1938,6 +1938,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports, skipNativePendingEffects: opts?.skipNativePendingEffects, skipNativeResponsePresentation: opts?.skipNativeResponsePresentation, + skipNativeReviewResponse: opts?.skipNativeReviewResponse, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 0633ed5cfc4..4695d1f2209 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -11312,7 +11312,9 @@ export function issueService(db: Db) { await notifyDecisionTargetWork(tx, `issue:${id}`); } if ((patch.status !== undefined && patch.status !== receiptExisting.status) || - (patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId)) { + (patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId) || + (patch.executionState !== undefined && parseObject(receiptExisting.executionState).status === "pending" && + parseObject(patch.executionState).status !== "pending")) { await notifyChatResponseAvailability(tx, { companyId: receiptExisting.companyId, issueId: id }); } if (patch.status !== undefined && patch.status !== receiptExisting.status) { diff --git a/server/src/services/native-runtime/external-chat-wait.integration.test.ts b/server/src/services/native-runtime/external-chat-wait.integration.test.ts index 86415417bb2..0687218b48d 100644 --- a/server/src/services/native-runtime/external-chat-wait.integration.test.ts +++ b/server/src/services/native-runtime/external-chat-wait.integration.test.ts @@ -53,6 +53,7 @@ import { configureAgentLifecycle, createAgentLifecycle, processAgentLifecycle } import { createOAuthStateStore } from "../tool-access/oauth/state.js"; import { approvalRecords } from "../approvals.js"; import { issueApprovalService } from "../issue-approvals.js"; +import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; import { nativeReviewResponseWork } from "./native-review-response-work.js"; import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { idleWorkSnapshot } from "../task-admission.js"; @@ -418,6 +419,10 @@ describe("native external-chat response wait", () => { userId, commentId, principalId, + storeBinding: { companyId, issueId, runId, agentId, normalizedSessionId: sessionId, + runnerSourceInstanceId: runnerInstanceId, completionContractId: contractId, + completionContractSha256: contractSha256, completionContractRevision: "external-chat-wait-v1", + completionContractCriterionIds: ["response"] }, }; } @@ -2802,6 +2807,23 @@ describe("native external-chat response wait", () => { return { ...fixture, gate, input }; } + it.each(["reconciler", "finalizer"] as const)("defers periodic governed-response %s repair to its registered worker", async path => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` }) + .where(eq(heartbeatRuns.id, fixture.runId)); + if (path === "reconciler") await reconcileNativeFinalizations(db, [fixture.runId], { skipNativeReviewResponse: true }); + else await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", + projectRunStatus: true, skipNativeReviewResponse: true }); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run.resultJson?.externalChatReviewPresentation).toBeUndefined(); + const work = nativeReviewResponseWork(db); + const [entry] = await work.load({ companyId: fixture.companyId, runId: fixture.runId }); + expect(entry?.id).toBe(fixture.runId); + expect(await work.check(entry)).toBeNull(); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1); + }); + it("recovers scheduled governed responses once under concurrent callers and preserves deleted receipts", async () => { const fixture = await seedRecoverableReviewResponse(); const work = nativeReviewResponseWork(db); @@ -3064,6 +3086,169 @@ describe("native external-chat response wait", () => { } finally { await coordinator.stop(); } }); + it("reactivates governed responses when pending execution policy is cleared without changing task status", async () => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(issues).set({ executionState: { status: "pending", currentStageId: randomUUID(), + currentStageIndex: 0, currentStageType: "review", currentParticipant: { type: "user", userId: fixture.userId, agentId: null }, + returnAssignee: { type: "agent", agentId: fixture.agentId, userId: null }, reviewRequest: null, + completedStageIds: [], lastDecisionId: null, lastDecisionOutcome: null, monitor: null, changesRequestedCount: 0 } }) + .where(eq(issues.id, fixture.issueId)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + await issueService(db).updateForCompany(fixture.issueId, fixture.companyId, { executionState: null }); + expect((await db.select().from(issues).where(eq(issues.id, fixture.issueId)))[0].status).toBe("in_review"); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("wakes governed responses from the real coordinator retry release", async () => { + const fixture = await seedRecoverableReviewResponse(); + const store = new NativeRunCoordinatorStore(db, fixture.storeBinding); + await store.claimFinalization({ leaseOwner: "review-response-fixture", leaseTtlMs: 30_000 }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + await store.markFinalizationRetry({ leaseOwner: "review-response-fixture", failureCode: "review_response_retry", retryAfterMs: 1000 }); + expect(await comments()).toHaveLength(0); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("retains the first governed-response wake through idle drain without passive SQL", async () => { + const fixture = await seedRecoverableReviewResponse(); + const [approval] = await db.insert(approvals).values({ companyId: fixture.companyId, + type: "hire_agent", status: "pending", payload: {} }).returning(); + await issueApprovalService(db).link(fixture.issueId, approval.id); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + let canRun = true; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => canRun); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + canRun = false; + await approvalRecords(db).cancel(approval.id); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(await comments()).toHaveLength(0); + expect(coordinator.nextWakeAt()).not.toBeNull(); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + canRun = true; worker.wake(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await comments()).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("retains failed governed-response recovery across drain and retries after resume", async () => { + const fixture = await seedRecoverableReviewResponse(); + let release!: () => void, observed!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { observed = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, fixture.endpointId)).for("update"); + observed(); await hold; + }); + await locked; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + let canRun = true; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => canRun); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ errors: errors.length, active: idleWorkSnapshot().active }), + { timeout: 10_000 }).toEqual({ errors: 1, active: 0 }); + expect(await comments()).toHaveLength(0); + expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); + canRun = false; + worker.wake(); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + release(); await holder; + canRun = true; worker.wake(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await comments()).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toHaveLength(1); + } finally { release(); await holder; await coordinator.stop(); vi.useRealTimers(); } + }); + + it("reconstructs the governed-response deadline after shutdown without early presentation", async () => { + const fixture = await seedRecoverableReviewResponse(); + const store = new NativeRunCoordinatorStore(db, fixture.storeBinding); + await store.claimFinalization({ leaseOwner: "review-future", leaseTtlMs: 30_000 }); + await store.markFinalizationRetry({ leaseOwner: "review-future", failureCode: "review_fixture", retryAfterMs: 60_000 }); + const [state] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const at = state.nextAttemptAt!.getTime(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: at }); + await coordinator.stop(); + expect(coordinator.nextWakeAt()).toBeNull(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(30_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + const resumed = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const restarted = nativeReviewResponseWork(db).register(resumed, () => true); + await restarted.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: resumed.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: at }); + expect(await comments()).toHaveLength(0); + await vi.advanceTimersByTimeAsync(Math.max(0, at - Date.now())); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await comments()).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(resumed.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { await resumed.stop(); vi.useRealTimers(); } + }); + it("reconstructs scheduled governed responses at startup and leaves their history quiet", async () => { const fixture = await seedRecoverableReviewResponse(); const errors: unknown[] = []; diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index b7905414006..50faf1722d9 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -639,6 +639,7 @@ type NativeReconciliationOptions = { skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; + skipNativeReviewResponse?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -869,7 +870,8 @@ async function reconcileNativeFinalizationsPass( row.decisionId && row.resultId && row.assessmentId && - currentDecisionJson.externalChatReviewPresentation + currentDecisionJson.externalChatReviewPresentation && + !options.skipNativeReviewResponse ) { // A later chat turn may retain the same pending review. Recover the // earlier response independently before status reconciliation skips @@ -1114,6 +1116,7 @@ async function reconcileNativeFinalizationsPass( workspaceFinalizeStatus, projectRunStatus: true, skipNativeResponsePresentation: options.skipNativeResponsePresentation, + skipNativeReviewResponse: options.skipNativeReviewResponse, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( record(row.runnerProfileJson).nativeWorkspaceSync, @@ -1173,6 +1176,7 @@ async function reconcileNativeFinalizationsPass( workspaceFinalizeStatus: barrier.status as "succeeded" | "failed", projectRunStatus: true, skipNativeResponsePresentation: options.skipNativeResponsePresentation, + skipNativeReviewResponse: options.skipNativeReviewResponse, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( record(row.runnerProfileJson).nativeWorkspaceSync, diff --git a/server/src/services/native-runtime/native-review-response-work.ts b/server/src/services/native-runtime/native-review-response-work.ts index 32df4ac1041..b11fb9b0646 100644 --- a/server/src/services/native-runtime/native-review-response-work.ts +++ b/server/src/services/native-runtime/native-review-response-work.ts @@ -6,6 +6,11 @@ import type { NativeReviewResponseScope } from "../native-review-response-notifi import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js"; +export function registerNativeReviewResponseWork(db: Db, + coordinator: ReturnType, canRun: () => boolean) { + return nativeReviewResponseWork(db).register(coordinator, canRun); +} + type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; export function nativeReviewResponseWork(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 dc30b8112d5..973c70db9cf 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1173,6 +1173,7 @@ export async function finalizeNativeRun(input: { /** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */ projectRunStatus?: boolean; skipNativeResponsePresentation?: boolean; + skipNativeReviewResponse?: boolean; /** Workspace-only replay must not consume the provider recovery budget. */ preserveProviderAttempt?: boolean; failpoint?: NativeStatusCommitFailpoint; @@ -1193,14 +1194,14 @@ export async function finalizeNativeRun(input: { const coordinator = claim.coordinator; if (!claim.leaseOwner && coordinator.phase === "committed") { if (isNativeRunnerOwnershipHeld(run)) return coordinator; - const presentationAlreadyMaterialized = coordinator.decisionId + const presentationAlreadyMaterialized = input.skipNativeReviewResponse || (coordinator.decisionId ? await hasMaterializedNativeReviewResponse(input.db, { companyId: run.companyId, issueId: coordinator.issueId, runId: run.id, decisionId: coordinator.decisionId, }) - : false; + : false); if ( !presentationAlreadyMaterialized && coordinator.decisionId &&