fix(server): schedule governed review response recovery

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-26 01:43:13 -05:00
1 parent ea46d0846a
commit d09253f7f5
16 files changed
+483 -9

No files matched your search

@@ -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.
@@ -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.
+1
View File
@@ -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",
]);
@@ -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';
@@ -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": [
+7
View File
@@ -2794,6 +2794,13 @@
"when": 1792989084087,
"tag": "0400_futuristic_earthquake",
"breakpoints": true
},
{
"idx": 401,
"version": "7",
"when": 1792996335782,
"tag": "0401_numerous_vulcan",
"breakpoints": true
}
]
}
+4
View File
@@ -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,
@@ -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<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 });
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 {
+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.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();
+3
View File
@@ -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<typeof registerNativeReviewResponseWork>[1], canRun: () => boolean) =>
registerNativeReviewResponseWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null),
registerNativeResponsePresentationWork: (coordinator: Parameters<typeof registerNativeResponsePresentationWork>[1], canRun: () => boolean) =>
registerNativeResponsePresentationWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null),
registerNativePendingEffectWork: (coordinator: Parameters<typeof registerNativePendingEffectWork>[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 }) {
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,
+3 -1
View File
@@ -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) {
@@ -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<void>(resolve => { release = resolve; });
const locked = new Promise<void>(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[] = [];
@@ -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,
@@ -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<typeof createDeliveryWorkCoordinator>, 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) {
@@ -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 &&