From 85c94ef85dbb379aa942afb41b8f77f0e44500b4 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 23:37:37 -0500 Subject: [PATCH] feat: add native response recovery worker and deadline fences Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 19 ++ ...ative-response-presentation-sleep-audit.md | 99 ++++++++- packages/db/src/client.ts | 1 + .../migrations/0400_futuristic_earthquake.sql | 3 + ...{0395_snapshot.json => 0400_snapshot.json} | 160 ++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/heartbeat_runs.ts | 3 + .../chat-channels.integration.test.ts | 188 ++++++++++++++++++ .../committed-chat-response-authority.test.ts | 41 ++++ .../chat-run-milestone-notifications.ts | 1 + .../services/delivery-work-notifications.ts | 1 + server/src/services/durable-chat-wakeup.ts | 16 ++ .../native-response-presentation-work.ts | 91 +++++++++ .../native-runtime/native-run-finalizer.ts | 13 +- server/src/services/scoped-deadline-work.ts | 5 +- 15 files changed, 643 insertions(+), 5 deletions(-) create mode 100644 packages/db/src/migrations/0400_futuristic_earthquake.sql rename packages/db/src/migrations/meta/{0395_snapshot.json => 0400_snapshot.json} (99%) create mode 100644 server/src/__tests__/committed-chat-response-authority.test.ts create mode 100644 server/src/services/native-runtime/native-response-presentation-work.ts 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 f4268e249a7..2f6c23a2c4c 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2308,3 +2308,22 @@ Slice 48 remains local and incomplete: owner error separation and the 33-case baseline pass; candidate discovery, producer coverage, lifecycle wiring, and periodic removal remain. Older unresolved stack gates and every other original recurring-server duty still require completion. The active goal is not complete. + + +## Slice 48 local worker progress + +Accepted-response recovery now has a local scoped worker, locked retry/effect +fences, milestone-topic notifications, and live authority registration invalidation. +The final selected real-PostgreSQL/lifecycle run passes 45 cases, and server types +pass. An elapsed-deadline race was reproduced before repair. A generated concurrent +partial index plus expression statistics avoids selected-history scans in the +10,000-row fixture; fresh migration and replay pass. See the +[response audit](2026-10-25-native-response-presentation-sleep-audit.md) for exact +scope and preserved failures. + +The worker is not wired into startup and periodic response recovery remains. +Producer coverage, remaining variants/failure tests, large dormant-backlog cost, +full local/remote acceptance, and PR publication are still required. Master was +refreshed to `7703220d38`; the complete stack has not yet been reconciled with it. +No merge, complete migration, whole-instance sleep, or full-goal completion is +claimed. diff --git a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md index 018667c0df5..5106a4ba796 100644 --- a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md +++ b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md @@ -1,6 +1,6 @@ # Native accepted-response presentation recovery -Status: source and baseline audit; no worker migration yet. +Status: worker and owner guards implemented locally; producer audit and startup/periodic integration incomplete. Branch: `codex/sleep-48-native-response-presentation`. Parent: [#17690](https://github.com/paperclipai/paperclip/pull/17690), `56c3601ea605b333141e86e0a810ed790fb6ce48`. @@ -133,3 +133,100 @@ source changes can authorize descendants. Audit its coverage for this duty befor adding caller wrappers. It already reaches status-decision settlement and run projection, but raw coordinator-store/session release and accepted-turn journal writers need individual inspection. Reuse is not proof that all producers exist. + + +## Local worker and locked scheduling fences + +The local worker now discovers exact run/coordinator/result/decision bindings, +filters selected answers and structurally irrelevant results, and dispatches +through `recoverCommittedNativeChatResponse`. It does not own publication policy. +`onlyWhenDue` rechecks the saved retry and outstanding status effects while the +existing coordinator lock is held. Existing explicit callers retain their old +behavior. A future retry is retained even if it becomes due between the locked +check and the follow-up read; its next attempt is immediate rather than discarded. + +Operational errors remain retryable through the shared coordinator. A structurally +eligible but unauthorized row becomes dormant until an eligibility mutation or +live authority registration invalidates its scope. Selected comments, including +deleted selections, remain excluded. No provider rerun or new work table is added. + +The existing milestone notification boundary now includes the response topic. +Hints explicitly select company and issue IDs: structural TypeScript callers may +pass larger objects, so serializing the whole argument is not safe. The first +integration run exposed that mistake as circular-object serialization during +status materialization; it was corrected before the passing runs below. + +Two narrow lifecycle capabilities support the new consumer: + +- `registerScopedDeadlineWork.invalidate` marks reconstruction dirty before waking. + A plain wake is insufficient after a prior dormant result removed its entry. + Existing consumers keep the same registration and scheduling behavior. +- The committed-response authority owner exposes registration notifications. + Registering/replacing the actual chat service reactivates dormant responses. + The worker's `stop` removes this subscription; the delivery coordinator still + owns worker cancellation and awaiting in-flight work. Repeated unsubscription + cannot delete a later listener set. The authority remains scoped to the exact Db. + +No startup registration or periodic skip is wired yet. The old recovery path is +still present. Full producer coverage, internal/continuation variants, recovery +failure/drain tests, and repeated large-scope mutation cost remain acceptance work. +Current discovery pages by run ID but reconstructs all matching pages in one +load; its resource bound under a large unresolved/dormant backlog needs proof +before activation. Do not describe the worker as a completed migration yet. + +## Query-plan and migration evidence + +The actual discovery query was captured from the driver. On a disposable database +with 10,000 additional selected runs, results, assessments, decisions, and +coordinators, it initially scanned completed history (22.656 ms). A partial +unpresented-run index alone still chose the wrong join order (16.532 ms). +Expression statistics for the selected-comment predicate allowed PostgreSQL to +start from the one unpresented run (0.091 ms). + +Generated migration `0400_futuristic_earthquake.sql` creates the partial index +concurrently, adds idempotent expression statistics, and analyzes the table. +The existing top-level migration runner owns concurrent-index replay/repair. +Snapshot retention contains exactly 0396–0400. Fresh migration, journal replay, +direct SQL replay, ready/valid index flags, and statistics presence passed. + +Rechecking the actual migration's objects (after removing experimental objects) +used the driver's real global, company, and issue-scoped queries. Global discovery +found one candidate in 0.078 ms; history-only company/issue scopes returned none +in 0.024/0.019 ms without sequential scans. These are query-plan fixtures, not +latency or fleet-sleep claims. Large unselected ineligible histories and repeated +configuration signals still need separate qualification. + +Evidence: `/tmp/sleep48-query.json`, `/tmp/sleep48-query-plan.sql`, +`/tmp/sleep48-query-plan.log`, `/tmp/sleep48-query-statistics.log`, +`/tmp/sleep48-production-plan.log`, `/tmp/sleep48-migration-verify.log`. + +## Current verification and preserved failures + +The expanded final selection passed 45 cases in two files: 33 existing accepted- +response cases, ten new worker/owner cases, and two authority lifecycle cases. +The 1,551 unselected chat cases are not claimed passed. Server typecheck passed. +The worker cases establish startup recovery, selected-history quiet over former +poll periods, real chat-service authority restoration, outer commit/rollback, +first notification during drain, SQL-free drain waiting, resume, saved future +retry, pending/failed status-effect ordering, and an elapsed-deadline race. + +The elapsed-deadline test failed before its fix: the worker returned null even +though the locked owner had deferred the response. It now retains the due entry +and the next attempt publishes exactly once. Log: +`/tmp/sleep48-elapsed-deadline-before.log`. + +Earlier worker runs exposed the hint serialization defect (nine fixture failures), +then two test-observation mistakes. A timer created before fake-time installation +was not advanced by the fake clock; the test now installs time before registration. +A null next deadline can occur while a sweep is in flight; commit assertions now +wait for the persisted publication, zero tracked activity, and no next deadline +jointly. The original logs remain `/tmp/sleep48-worker-tests.log`, +`/tmp/sleep48-worker-tests-v2.log`, `/tmp/sleep48-drain-isolated.log`, and +`/tmp/sleep48-worker-tests-v3.log`. The corrected initial selection passed eleven +cases in `/tmp/sleep48-worker-tests-v4.log`; the final expanded result is +`/tmp/sleep48-acceptance-v2.log` with server types in +`/tmp/sleep48-acceptance-typecheck-v2.log`. + +No slice 48 PR, full workspace acceptance, remote CI/review, or provider sleep +qualification is claimed by these local results. The earlier stack gates and +remaining recurring-server duties are unchanged. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index ffd237fca33..a74fdb5d893 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -525,6 +525,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0397_lumpy_talos.sql", "0398_shocking_molecule_man.sql", "0399_jazzy_ben_grimm.sql", + "0400_futuristic_earthquake.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0400_futuristic_earthquake.sql b/packages/db/src/migrations/0400_futuristic_earthquake.sql new file mode 100644 index 00000000000..ca0c9e198e3 --- /dev/null +++ b/packages/db/src/migrations/0400_futuristic_earthquake.sql @@ -0,0 +1,3 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_native_unpresented_response_idx" ON "heartbeat_runs" USING btree ("company_id","native_issue_id","id") 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';--> statement-breakpoint +CREATE STATISTICS IF NOT EXISTS "heartbeat_runs_native_response_selection_stats" ON (jsonb_typeof("result_json" #> '{presentationDecision,commentId}')) FROM "heartbeat_runs";--> statement-breakpoint +ANALYZE "heartbeat_runs"; diff --git a/packages/db/src/migrations/meta/0395_snapshot.json b/packages/db/src/migrations/meta/0400_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0395_snapshot.json rename to packages/db/src/migrations/meta/0400_snapshot.json index e5734bb2d4a..e06b1cad761 100644 --- a/packages/db/src/migrations/meta/0395_snapshot.json +++ b/packages/db/src/migrations/meta/0400_snapshot.json @@ -1,6 +1,6 @@ { - "id": "e9c56475-b2c8-4cb4-b7dd-89bfa645a93a", - "prevId": "90ac76c4-b6d8-4c38-b4b8-d51699ca2ec1", + "id": "d51b06de-ba83-4221-8ea1-863fd947f021", + "prevId": "5b5659ef-e4b6-4584-9b4f-ed075047ef2f", "version": "7", "dialect": "postgresql", "tables": { @@ -2637,6 +2637,28 @@ } }, "indexes": { + "agent_instruction_copies_collection_idx": { + "name": "agent_instruction_copies_collection_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "NOT (coalesce(\"agent_instruction_working_copies\".\"receipt\", '{}'::jsonb) ? 'retainedByRunId') AND (\n (\"agent_instruction_working_copies\".\"attempts\" < 3 AND (\"agent_instruction_working_copies\".\"state\" IN ('prepared', 'pending_collection', 'unchanged_turn', 'warm_saved')\n OR (\"agent_instruction_working_copies\".\"state\" = 'preparing' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1')))\n OR (\"agent_instruction_working_copies\".\"state\" = 'unavailable' AND \"agent_instruction_working_copies\".\"process_stopped_at\" IS NULL\n AND \"agent_instruction_working_copies\".\"location\" LIKE 'remote:%' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1'))", + "concurrently": false, + "method": "btree", + "with": {} + }, "agent_instruction_copies_cleanup_idx": { "name": "agent_instruction_copies_cleanup_idx", "columns": [ @@ -33649,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": [ @@ -42830,6 +42902,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": [ @@ -52016,6 +52144,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 f1e41ee5c3e..3380415c237 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2787,6 +2787,13 @@ "when": 1792986633369, "tag": "0399_jazzy_ben_grimm", "breakpoints": true + }, + { + "idx": 400, + "version": "7", + "when": 1792989084087, + "tag": "0400_futuristic_earthquake", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 99f06a955dc..9e215f825f0 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -111,6 +111,9 @@ export const heartbeatRuns = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + nativeUnpresentedResponseIdx: index("heartbeat_runs_native_unpresented_response_idx") + .on(table.companyId, table.nativeIssueId, table.id) + .where(sql`${table.runtimeMode} = 'native' AND ${table.status} in ('succeeded', 'failed', 'timed_out') AND ${table.finishedAt} is not null AND jsonb_typeof(${table.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`), cancelledWorkspacePreservationIdx: index("heartbeat_runs_cancelled_workspace_preservation_idx").on(table.id, table.companyId) .where(sql`${table.runtimeMode} = 'native' AND ${table.status} = 'cancelled' AND ${table.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared' diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 1f710a4ce09..32bc05faddb 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,5 @@ +import { nativeResponsePresentationWork } from "../services/native-runtime/native-response-presentation-work.js"; +import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js"; import { registerGitHubReviewProjectionWork } from "../services/chat-channels/github/review-projection-work.js"; import { retainGitHubReviewReceipt } from "../services/chat-channels/github/task-preparation.js"; import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js"; @@ -132,6 +134,7 @@ import { issues, nativeRunResults, nativeRunFinalizations, + statusDecisionEffects, environmentLeases, environments, principalPermissionGrants, @@ -72595,6 +72598,191 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }, ); + it("recovers accepted-response work on startup and leaves selected history quiet", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const work = nativeResponsePresentationWork(db); + const worker = work.register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId)); + expect(comments).toHaveLength(1); + expect(comments[0].body).toBe(context.result.summary); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(await work.load({ companyId: context.fixture.companyId })).toEqual([]); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("reactivates dormant accepted-response work when the live chat authority registers", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + await context.service.shutdown(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + let replacement: ReturnType | undefined; + try { + await worker.ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + replacement = createService(new FakeChatSdkRuntime(), fakeTelegramFetch() as typeof fetch); + await expect.poll(async () => (await readChatPublicationsByEndpointId(context.endpoint.id)).length, { timeout: 5000 }).toBe(1); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); await replacement?.service.shutdown(); } + }); + + it.each(["commit", "rollback"])("keeps accepted-response work behind the outer %s boundary", async outcome => { + const context = await committedChatResponseRecoveryFixture("telegram"); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, context.runId)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + const transaction = db.transaction(async tx => { + await tx.transaction(async nested => { + await notifyChatRunMilestones(nested, { companyId: context.fixture.companyId, issueId: context.issue.id }); + await nested.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, context.runId)); + }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + if (outcome === "rollback") throw new Error("response fixture rollback"); + }); + if (outcome === "rollback") await expect(transaction).rejects.toThrow("response fixture rollback"); + else await transaction; + await expect.poll(async () => ({ + publications: (await readChatPublicationsByEndpointId(context.endpoint.id)).length, + active: idleWorkSnapshot().active, + next: coordinator.nextWakeAt(), + }), { timeout: 5000 }).toEqual({ publications: outcome === "commit" ? 1 : 0, active: 0, next: null }); + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); await context.service.shutdown(); } + }); + + it("retains the first accepted-response wake through idle drain without passive SQL", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, context.runId)); + let canRun = true; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => canRun); + try { + await worker.ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + canRun = false; + await db.transaction(async tx => { + await notifyChatRunMilestones(tx, { companyId: context.fixture.companyId, issueId: context.issue.id }); + await tx.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, context.runId)); + }); + await expect.poll(() => idleWorkSnapshot().active, { timeout: 5000 }).toBe(0); + expect(coordinator.nextWakeAt()).not.toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + canRun = true; worker.wake(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("reconstructs a future accepted-response deadline without early publication", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const at = new Date(Date.now() + 60_000); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: at }).where(eq(nativeRunFinalizations.runId, context.runId)); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + try { + await worker.ready; + expect(coordinator.nextWakeAt()).toBe(at.getTime()); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await vi.advanceTimersByTimeAsync(Math.max(0, at.getTime() - Date.now())); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("retains an accepted-response deadline that elapses during the locked recovery check", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const work = nativeResponsePresentationWork(db); + const [entry] = await work.load({ companyId: context.fixture.companyId }); + expect(entry?.id).toBe(context.runId); + vi.useFakeTimers({ toFake: ["Date"] }); + const due = new Date(Date.now() + 1000); + const transaction = db.transaction.bind(db); + const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + await transaction(async tx => { + await tx.update(nativeRunFinalizations).set({ nextAttemptAt: due }).where(eq(nativeRunFinalizations.runId, context.runId)); + }); + const result = await transaction(callback, options); + vi.setSystemTime(due.getTime() + 1); + return result; + }) as Db["transaction"]); + try { + const deferred = await work.check(entry); + intercept.mockRestore(); + expect(deferred).toEqual({ ...entry, dueAt: due.getTime() }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await expect(work.check(deferred!)).resolves.toBeNull(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + } finally { intercept.mockRestore(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("defers scheduled accepted-response recovery until the persisted retry is due", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const input = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId, onlyWhenDue: true }; + try { + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, context.runId)); + await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(false); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(0); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, context.runId)); + await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(true); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(1); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + } finally { await context.service.shutdown(); } + }); + + it.each(["pending", "failed"])("defers scheduled accepted-response recovery behind %s status effects", async deliveryState => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const input = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId, onlyWhenDue: true }; + try { + const [effect] = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, context.coordinator.decisionId!)); + expect(effect).toBeDefined(); + await db.update(statusDecisionEffects).set({ deliveryState }).where(eq(statusDecisionEffects.id, effect.id)); + await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(false); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(0); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await db.update(statusDecisionEffects).set({ deliveryState: "delivered" }).where(eq(statusDecisionEffects.id, effect.id)); + await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(true); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + } finally { await context.service.shutdown(); } + }); + it("leaves a contended accepted response retryable without reclassifying or partially publishing it", async () => { const context = await committedChatResponseRecoveryFixture("telegram"); let release!: () => void; diff --git a/server/src/__tests__/committed-chat-response-authority.test.ts b/server/src/__tests__/committed-chat-response-authority.test.ts new file mode 100644 index 00000000000..497cbaa4233 --- /dev/null +++ b/server/src/__tests__/committed-chat-response-authority.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { authorizeCommittedChatResponse, onCommittedChatResponseAuthorityRegistered, registerCommittedChatResponseAuthority } from "../services/durable-chat-wakeup.js"; + +describe("committed response authority lifecycle", () => { + it("reactivates only its database and unsubscribes without removing a replacement listener", () => { + const db = {} as Db, otherDb = {} as Db; + const first = vi.fn(), second = vi.fn(), other = vi.fn(); + const disposeFirst = onCommittedChatResponseAuthorityRegistered(db, first); + const disposeOther = onCommittedChatResponseAuthorityRegistered(otherDb, other); + const removeAuthority = registerCommittedChatResponseAuthority(db, async () => {}); + expect(first).toHaveBeenCalledTimes(1); + expect(other).not.toHaveBeenCalled(); + disposeFirst(); + const disposeSecond = onCommittedChatResponseAuthorityRegistered(db, second); + disposeFirst(); + const removeReplacement = registerCommittedChatResponseAuthority(db, async () => {}); + expect(first).toHaveBeenCalledTimes(1); + expect(second).toHaveBeenCalledTimes(1); + removeAuthority(); + expect(second).toHaveBeenCalledTimes(1); + disposeSecond(); disposeOther(); removeReplacement(); + const removeLast = registerCommittedChatResponseAuthority(db, async () => {}); + expect(second).toHaveBeenCalledTimes(1); + removeLast(); + }); + + it("retains the replacement authority when the old service shuts down", async () => { + const db = {} as Db; + const first = vi.fn(async () => {}), second = vi.fn(async () => {}); + const removeFirst = registerCommittedChatResponseAuthority(db, first); + const removeSecond = registerCommittedChatResponseAuthority(db, second); + try { + removeFirst(); + const input = { companyId: "company", issueId: "issue", agentId: "agent", runId: "run", resultId: "result" }; + await authorizeCommittedChatResponse(db, db, input); + expect(first).not.toHaveBeenCalled(); + expect(second).toHaveBeenCalledExactlyOnceWith(db, input); + } finally { removeSecond(); } + }); +}); diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index cd0df3664f8..15a54e1173b 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -8,6 +8,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ChatRunMilestoneScope = { companyId: string; issueId?: string }; export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeResponsePresentation, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify({ companyId: scope.companyId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 602a3d71c2a..a7eec6622b7 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 = { + nativeResponsePresentation: "native-response-presentation", nativePendingEffects: "native-pending-effects", unsafeWorkspaceExportRecovery: "unsafe-workspace-export-recovery", automaticCompletionReviews: "automatic-completion-reviews", diff --git a/server/src/services/durable-chat-wakeup.ts b/server/src/services/durable-chat-wakeup.ts index 62324a15ece..692312c3a20 100644 --- a/server/src/services/durable-chat-wakeup.ts +++ b/server/src/services/durable-chat-wakeup.ts @@ -21,6 +21,21 @@ const committedChatResponseAuthorities = new WeakMap< CommittedChatResponseAuthority >(); +const committedChatResponseAuthorityListeners = new WeakMap void>>(); + +export function onCommittedChatResponseAuthorityRegistered(db: Db, listener: () => void): () => void { + let listeners = committedChatResponseAuthorityListeners.get(db); + if (!listeners) { + listeners = new Set(); + committedChatResponseAuthorityListeners.set(db, listeners); + } + listeners.add(listener); + return () => { + listeners.delete(listener); + if (listeners.size === 0 && committedChatResponseAuthorityListeners.get(db) === listeners) committedChatResponseAuthorityListeners.delete(db); + }; +} + export class CommittedChatResponseAuthorizationError extends HttpError { constructor() { super( @@ -41,6 +56,7 @@ export function registerCommittedChatResponseAuthority( authority: CommittedChatResponseAuthority, ): () => void { committedChatResponseAuthorities.set(db, authority); + for (const listener of committedChatResponseAuthorityListeners.get(db) ?? []) listener(); return () => { if (committedChatResponseAuthorities.get(db) === authority) { committedChatResponseAuthorities.delete(db); diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts new file mode 100644 index 00000000000..bd260d1366e --- /dev/null +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -0,0 +1,91 @@ +import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { ChatRunMilestoneScope } from "../chat-run-milestone-notifications.js"; +import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wakeup.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; + +type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; + +export function nativeResponsePresentationWork(db: Db) { + function candidates(scope?: ChatRunMilestoneScope, after?: string, runId?: string) { + return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), + eq(nativeRunResults.runId, heartbeatRuns.id), eq(nativeRunResults.completionContractId, heartbeatRuns.completionContractId))) + .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId), + eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId), + eq(statusDecisions.runId, heartbeatRuns.id), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId))) + .where(and( + eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out"]), + isNotNull(heartbeatRuns.finishedAt), eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner), + eq(nativeRunResults.schemaStatus, "accepted"), + sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`, + sql`coalesce(${statusDecisions.decisionJson}->'externalChatReviewPresentation', 'null'::jsonb) in ('null'::jsonb, 'false'::jsonb, '0'::jsonb, '""'::jsonb)`, + sql`${nativeRunResults.resultJson} #>> '{result,schema}' = 'paperclip.run_result.v1'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,runTerminalState}' = 'succeeded'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`, + or( + and(eq(heartbeatRuns.status, "succeeded"), + sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`), + and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'yielded'`, + sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' in ('response_wake', 'same_agent', 'monitor')`), + ), + 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 ? and(eq(heartbeatRuns.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined, + after ? gt(heartbeatRuns.id, after) : undefined, runId ? eq(heartbeatRuns.id, runId) : undefined, + )).orderBy(asc(heartbeatRuns.id)).limit(100); + } + async function load(scope?: ChatRunMilestoneScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await candidates(scope, after); + for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].id; + } + throw new Error("Native response presentation discovery stopped"); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Native response presentation recovery stopped"); + const scope = { companyId: entry.companyId, issueId: entry.issueId }; + const [current] = await candidates(scope, undefined, entry.id); + if (!current) return null; + if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() }; + if (signal?.aborted) throw new Error("Native response presentation recovery stopped"); + const recoveryStartedAt = Date.now(); + await recoverCommittedNativeChatResponse(db, { ...scope, runId: entry.id, onlyWhenDue: true }); + const [remaining] = await candidates(scope, undefined, entry.id); + return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt + ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeResponsePresentation, + decodeScope: (hint: string): ChatRunMilestoneScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: ChatRunMilestoneScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId), + load: (scope, signal) => load(scope, signal, canRun), + check: (entry, signal) => { + if (!canRun()) throw new Error("Native response presentation recovery paused"); + return check(entry, signal); + }, + }, canRun); + const unsubscribe = onCommittedChatResponseAuthorityRegistered(db, () => worker.invalidate()); + return { ...worker, stop: unsubscribe }; + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index c9eeafb47f0..dbaad2dacee 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -37,6 +37,7 @@ import { nativeRunFinalizations, nativeRunResults, statusDecisions, + statusDecisionEffects, workAssessments, workspaceOperations, } from "@paperclipai/db"; @@ -857,7 +858,7 @@ export async function repairCommittedNativeChatResponse( export async function recoverCommittedNativeChatResponse( db: Db, - input: { companyId: string; issueId: string; runId: string }, + input: { companyId: string; issueId: string; runId: string; onlyWhenDue?: boolean }, ): Promise { let agentId: string | null = null; let materialized = false; @@ -920,6 +921,16 @@ export async function recoverCommittedNativeChatResponse( !coordinator.assessmentId ) return false; + if (input.onlyWhenDue) { + if (coordinator.nextAttemptAt && coordinator.nextAttemptAt.getTime() > Date.now()) return false; + const [pendingEffect] = await tx.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, input.companyId), + eq(statusDecisionEffects.issueId, input.issueId), + eq(statusDecisionEffects.decisionId, coordinator.decisionId), + inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + )).limit(1); + if (pendingEffect) return false; + } const [accepted] = await tx .select() .from(nativeRunResults) diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index 711b5d2aaf5..c6667cc47c7 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -90,5 +90,8 @@ export function registerScopedDeadlineWork( hasPending: async () => canRun() && dirty.size > 0, nextRunAt: next, }); - return worker; + return { ...worker, invalidate(hint = "*") { + dirty.add(hint); + worker.wake(); + } }; }