diff --git a/doc/idle-decision-continuations.md b/doc/idle-decision-continuations.md new file mode 100644 index 00000000000..8146e4ff653 --- /dev/null +++ b/doc/idle-decision-continuations.md @@ -0,0 +1,106 @@ +# Decision continuation delivery while idle + +A terminal decision can owe an agent wake. Its existing row stores that debt: +`continuationPolicy = wake_origin_agent`, `metadata.continuationPending = true`, +and either expired/cancelled status or a decided status with succeeded/partial/ +failed execution. Open decisions and running effects are not ready to deliver. +This migration adds no queue table and changes no wake payload. + +## Ownership and producers + +`decisions.ts` owns decision creation, authorization, signatures, effects, and +state changes. Its ten update sites now use a narrow transaction-aware write +helper before the SQL statement. These cover the running claim; ordinary effect +completion; archive-proposal signature, manifest, and execution failures; archive +success; TTL rejection on use; cancel; direct dismissal; and background expiry. +The helper acquires intent in the existing outer transaction or its savepoint. +It does not include effect execution, provider calls, activity publication, or +agent wake callbacks in a new database transaction. + +Creation and bundle creation insert open decisions, so they do not create ready +continuation debt. Acknowledgement only clears debt and belongs to the delivery +owner; it does not signal itself. The production decision services are created +by server startup and routes with the application's DB handle. Independent SQL +writers and additional active processes require a separate notification and +ownership contract. This change does not make wake-producing service methods +safe to call inside an uncommitted caller transaction; only their SQL mutation +boundary has the existing outer-commit guarantee. + +`decision-continuations.ts` owns the readiness predicate, delivery, and recovery. +It reads at most 100 IDs per page and advances past failed candidates before +reporting failure. This avoids starving later rows behind a failing first page. +A partial ID index contains ready pending continuations only. The predicate uses +fixed SQL literals so a prepared generic plan can still use that index. + +The shared DB package exposes an opaque work-scope identity. Raw DB object +identity is insufficient because dedicated and liveness child pools already +share their owner's signals. The key is shared by those children and nested +transactions; independent roots have different keys. No connection credentials, +private transaction state, or decision policy are exposed. Existing signal +consumers and transactions that do not register work keep their behavior. + +## Foreground, recovery, and failure + +Foreground delivery remains immediate. All service instances sharing the DB +owner coalesce a simultaneous wake for the same decision into one promise. +Delivery re-reads durable readiness after claiming that local flight, so a late +foreground call cannot resend a continuation already acknowledged by recovery. +The promise is removed on success or failure. This is process-local coalescing, +not a new durable claim or an exactly-once guarantee. + +Effect completion requires the row to remain running. A competing completion +cannot restore pending metadata after another caller has completed and delivered +it. Effects retain their own existing durable claims and authorization checks. +The delivery worker never executes effects. + +A wake failure leaves pending metadata intact. A failed acknowledgement also +leaves debt, so a later attempt can send the wake again. A crash after accepting +the wake but before acknowledgement retains the existing at-least-once contract. +An uncertain producer transaction remains held until the shared work-signal +protocol establishes settlement; one empty scan cannot clear that uncertainty. + +## Lifecycle and scope + +Startup reconstructs pending delivery. Committed changes wake the worker through +the existing coordinator. Failures or remaining ready debt retain a 30-second +retry. Open/running decisions, no decisions, and delivered history have no +recurring continuation scan or deadline. Changes during a pass retain another +request. Drain/standby defer dispatch without consuming that request; shutdown +waits for admitted delivery. A future process-local retry is not a Cloud alarm. + +Enabled-heartbeat startup registers this worker before the combined decision +sweep. The main tick skips that sweep's independent continuation scan. Its public +API retains combined behavior by default. Expiry or effect recovery can still +deliver their newly produced continuations immediately. Disabled-heartbeat +startup keeps its existing behavior and does not dispatch this new worker. + +Decision expiry, invalidated-target scanning, interrupted effect recovery, and +retention remain separate migrations. Existing Cloud safety blockers are not +relaxed. This establishes worker quiet under the named conditions, not deployed +host/database suspension or measured fleet savings. + +## Verification + +The selected run passes 92 tests: 49 real-PostgreSQL decision cases, 39 startup +cases, and four database work-signal unit cases. Fifteen new continuation cases +cover real decide/cancel/dismiss/expiry producers, running-to-terminal completion, +foreground contention, dedicated child pools, stale caller re-read, restart, +failed acknowledgement, outer rollback/commit, first write during drain, failed +retry through drain, changes during delivery, shutdown, and paging past a failed +candidate. Quiet checks observe no SQL calls across an hour of controlled time. +Nine pre-existing embedded-PostgreSQL signal tests are skipped locally because +the embedded runtime is unavailable; CI must run that supported environment. +Workspace typecheck and build pass. The final literal-predicate adjustment also +passes all 49 decision tests. + +Migration 0382 applies and replays without error. A rolled-back synthetic +40,000-row history with one pending continuation uses the partial index under +`force_generic_plan`, reads two shared buffers, and executes in 0.015 ms on the +local PostgreSQL server. This is query evidence, not a production latency claim. + +The first test selection had 90 passes and one failure: the injected PostgreSQL +acknowledgement error is wrapped by Drizzle, but the assertion expected its +message at the top level. The assertion now checks the original cause. No +production retry, hold, or timeout was weakened. A recursive unit-fixture type +initially prevented migration generation; an explicit fixture type fixed it. +Full CI and fresh review remain required before PR readiness. 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 b40dd9e523a..f7733fcd0f7 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -15,7 +15,7 @@ Do not mark an item done from a timer removal or a green check alone. | 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | [#17443](https://github.com/paperclipai/paperclip/pull/17443), all checks passed, Greptile 5/5 at `36844a458b` | | 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending | | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | -| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; notification delivery: [#17475](https://github.com/paperclipai/paperclip/pull/17475), local verification passed, Greptile 5/5 at `692e3ee02c`, CI running | +| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; notification delivery: [#17475](https://github.com/paperclipai/paperclip/pull/17475), all checks passed (Paperclip Review skipped), Greptile 5/5 at `692e3ee02c` | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | | 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), checks passed at `d41590ac0e`; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), all checks passed, Greptile 5/5 at `c5c53fc43f`; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), all checks passed, Greptile 5/5 at `ce6a260a83`; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), all checks passed, Greptile 5/5 at `7a1cf03e62`; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), all checks passed, Greptile 5/5 at `85003a7ac4`; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), all checks passed, Greptile 5/5 at `0956883a84` | @@ -369,3 +369,26 @@ unavailable. Slice 08 review update: Greptile is 5/5 on `692e3ee02c` with no actionable findings. The latest CI snapshot has 10 successful checks and 44 in progress, with no failures. This is a live CI wait, not a readiness or completion claim. + + +## Slice 09 implementation: decision continuations + +The continuation owner and worker are implemented on `codex/sleep-09-decision-continuations`. +See [decision continuations](../idle-decision-continuations.md) for producer, +transaction, concurrency, failure, and test evidence. The existing row is durable +truth; the only DB API addition is an opaque shared work-scope identity for local +coalescing across child pools. No new queue table or external wake exemption is +added. Terminal effect settlement is fenced to running state so a competing +caller cannot re-arm an acknowledged continuation. A concurrent partial index +bounds pending selection independently of retained history. + +The selected run passes 92 tests; nine older embedded-Postgres cases are locally +skipped, with the new lifecycle tests running against real external PostgreSQL. +Workspace typecheck/build pass. Migration apply/replay and a 40,000-row prepared +query probe pass. Initial fixture type/error-shape failures are recorded in the +behavior document. PR creation and full CI/review are next. + +Parent #17475 at `692e3ee02c` now has passing checks and Greptile 5/5; the extra +Paperclip Review check was skipped. Fresh master is `fe85fd7ac4`, still at +migration 0377. The stack remains open and unmerged. The full inventory above +remains the completion scope. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 30c09ed5550..b4a2818acc6 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -507,6 +507,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0378_mean_scream.sql", "0379_narrow_the_executioner.sql", "0381_open_madame_web.sql", + "0382_chemical_madame_web.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index d65d39527ac..0702f8c5336 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -45,4 +45,4 @@ export { issueRelations } from "./schema/issue_relations.js"; export { issueReferenceMentions } from "./schema/issue_reference_mentions.js"; export * from "./schema/index.js"; -export { signalDatabaseWork, subscribeDatabaseWork, databaseWorkPending, reconcileDatabaseWork } from "./work-signals.js"; +export { databaseWorkScopeKey, signalDatabaseWork, subscribeDatabaseWork, databaseWorkPending, reconcileDatabaseWork } from "./work-signals.js"; diff --git a/packages/db/src/migrations/0382_chemical_madame_web.sql b/packages/db/src/migrations/0382_chemical_madame_web.sql new file mode 100644 index 00000000000..ad828b44e2c --- /dev/null +++ b/packages/db/src/migrations/0382_chemical_madame_web.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "decisions_pending_continuation_idx" ON "decisions" USING btree ("id") WHERE "decisions"."continuation_policy" = 'wake_origin_agent' AND "decisions"."metadata" ->> 'continuationPending' = 'true' AND ("decisions"."status" IN ('expired', 'cancelled') OR ("decisions"."status" = 'decided' AND "decisions"."execution_status" IN ('succeeded', 'partial', 'failed'))); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0377_snapshot.json b/packages/db/src/migrations/meta/0382_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0377_snapshot.json rename to packages/db/src/migrations/meta/0382_snapshot.json index 6e89edc3a9b..7f91405ae3e 100644 --- a/packages/db/src/migrations/meta/0377_snapshot.json +++ b/packages/db/src/migrations/meta/0382_snapshot.json @@ -1,6 +1,6 @@ { - "id": "a0fcd9b8-f125-4985-8795-f567466b9aef", - "prevId": "7832df38-fa1b-4687-bd06-eccbc29bd28c", + "id": "5e8d042e-d697-4753-9582-7d0c042ef893", + "prevId": "afdc5e70-19b3-416d-917a-beb9fdc1e2e1", "version": "7", "dialect": "postgresql", "tables": { @@ -24030,6 +24030,22 @@ "method": "btree", "with": {} }, + "decisions_pending_continuation_idx": { + "name": "decisions_pending_continuation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"continuation_policy\" = 'wake_origin_agent' AND \"decisions\".\"metadata\" ->> 'continuationPending' = 'true' AND (\"decisions\".\"status\" IN ('expired', 'cancelled') OR (\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" IN ('succeeded', 'partial', 'failed')))", + "concurrently": false, + "method": "btree", + "with": {} + }, "decisions_bundle_idx": { "name": "decisions_bundle_idx", "columns": [ @@ -52787,6 +52803,66 @@ "method": "btree", "with": {} }, + "tool_action_requests_pending_expiry_idx": { + "name": "tool_action_requests_pending_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_executing_deadline_idx": { + "name": "tool_action_requests_executing_deadline_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'executing'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_approved_idx": { + "name": "tool_action_requests_approved_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'approved'", + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_invocation_idx": { "name": "tool_action_requests_invocation_idx", "columns": [ @@ -55197,6 +55273,21 @@ "method": "btree", "with": {} }, + "tool_gateway_sessions_expiry_idx": { + "name": "tool_gateway_sessions_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_gateway_sessions_run_idx": { "name": "tool_gateway_sessions_run_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 9b6437c673a..ba1ac84f9f8 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2661,6 +2661,13 @@ "when": 1792863626931, "tag": "0381_open_madame_web", "breakpoints": true + }, + { + "idx": 382, + "version": "7", + "when": 1792871748836, + "tag": "0382_chemical_madame_web", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/decisions.ts b/packages/db/src/schema/decisions.ts index e2f668c40e6..18b18d0c704 100644 --- a/packages/db/src/schema/decisions.ts +++ b/packages/db/src/schema/decisions.ts @@ -69,6 +69,8 @@ export const decisions = pgTable( table.status, table.expiresAt, ), + pendingContinuationIdx: index("decisions_pending_continuation_idx").on(table.id) + .where(sql`${table.continuationPolicy} = 'wake_origin_agent' AND ${table.metadata} ->> 'continuationPending' = 'true' AND (${table.status} IN ('expired', 'cancelled') OR (${table.status} = 'decided' AND ${table.executionStatus} IN ('succeeded', 'partial', 'failed')))`), bundleIdx: index("decisions_bundle_idx").on(table.bundleId), originIssueIdx: index("decisions_origin_issue_idx").on(table.originIssueId), companyIdempotencyUq: uniqueIndex("decisions_company_idempotency_uq") diff --git a/packages/db/src/work-signals.test.ts b/packages/db/src/work-signals.test.ts index 13bd4b8552a..b14740bd38e 100644 --- a/packages/db/src/work-signals.test.ts +++ b/packages/db/src/work-signals.test.ts @@ -3,7 +3,7 @@ import { drizzle } from "drizzle-orm/postgres-js"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { createDb, withDedicatedDbConnection } from "./client.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js"; -import { databaseWorkPending, installDatabaseWorkSignals, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js"; +import { databaseWorkScopeKey, databaseWorkPending, installDatabaseWorkSignals, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js"; const support = await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("database work signals", () => { @@ -187,3 +187,23 @@ describe("work signal query boundaries", () => { await expect(signalDatabaseWork(connection, "bad")).rejects.toThrow("inside a transaction"); }); }); + + +describe("database work scope identity", () => { + type TestConnection = { execute: () => Promise; transaction: (callback: (tx: TestConnection) => unknown) => Promise }; + function connection(): TestConnection { + return { execute: vi.fn(async () => []), transaction: async callback => callback(connection()) }; + } + it("shares owner identity through child pools and nested transactions, but isolates independent owners", async () => { + const first = installDatabaseWorkSignals(connection()); + const second = installDatabaseWorkSignals(connection()); + const child = installDatabaseWorkSignals(connection(), first); + const key = databaseWorkScopeKey(first); + expect(databaseWorkScopeKey(second)).not.toBe(key); + expect(databaseWorkScopeKey(child)).toBe(key); + await child.transaction(async tx => { + expect(databaseWorkScopeKey(tx)).toBe(key); + await tx.transaction(async nested => { expect(databaseWorkScopeKey(nested)).toBe(key); }); + }); + }); +}); diff --git a/packages/db/src/work-signals.ts b/packages/db/src/work-signals.ts index 4abb2570a5e..cf914398e5d 100644 --- a/packages/db/src/work-signals.ts +++ b/packages/db/src/work-signals.ts @@ -5,7 +5,7 @@ import { sql } from "drizzle-orm"; // an XID, so unrelated transactions keep their existing SQL behavior. type Listener = (event: "intent" | "settled", hints?: ReadonlySet) => void; type Transaction = { topics: Set; hints: Map>; xid?: Promise; uncertain: boolean }; -type Scope = { listeners: Map>; transactions: Set; root: Connection }; +type Scope = { key: object; listeners: Map>; transactions: Set; root: Connection }; type Context = { scope: Scope; transaction?: Transaction }; type Connection = { transaction: (...args: any[]) => Promise; execute: (...args: any[]) => Promise }; const contexts = new WeakMap(); @@ -19,7 +19,7 @@ function publish(scope: Scope, topic: string, event: "intent" | "settled" = "set /** Installed once by createDb; dedicated child pools share their owner's scope. */ export function installDatabaseWorkSignals(db: T, owner?: object): T { - const scope = (owner && contexts.get(owner)?.scope) || { listeners: new Map(), transactions: new Set(), root: db }; + const scope = (owner && contexts.get(owner)?.scope) || { key: {}, listeners: new Map(), transactions: new Set(), root: db }; install(db, { scope }); return db; } @@ -112,3 +112,7 @@ export function subscribeDatabaseWork(owner: object, topic: string, listener: Li } return () => { listeners.delete(listener); }; } + +export function databaseWorkScopeKey(connection: object): object { + return contexts.get(connection)?.scope.key ?? connection; +} diff --git a/server/src/__tests__/decisions-service.test.ts b/server/src/__tests__/decisions-service.test.ts index 4b418e355f1..f235469e19c 100644 --- a/server/src/__tests__/decisions-service.test.ts +++ b/server/src/__tests__/decisions-service.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; -import { eq, inArray } from "drizzle-orm"; +import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { activityLog, @@ -11,6 +11,7 @@ import { companies, companyMemberships, createDb, + withDedicatedDbConnection, decisionEffectExecutions, decisionRetention, decisions, @@ -26,7 +27,15 @@ import { decisionService } from "../services/decisions.js"; import { hashAttentionArchiveManifest } from "../services/decision-retention.js"; import type { AttentionArchiveManifestEntry, AttentionArchiveTargetSnapshot } from "@paperclipai/shared"; -const support = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { decisionContinuationService } from "../services/decision-continuations.js"; +import { registerDecisionContinuationWork } from "../services/decision-continuation-work.js"; +import { writeDecisionChange } from "../services/decision-work-notifications.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISIONS_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describePg = support.supported ? describe : describe.skip; describePg("decisionService", () => { @@ -41,9 +50,11 @@ describePg("decisionService", () => { let decidedByUserId: string; let wakes: Array>; + let stopWorker: (() => Promise) | undefined; + beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decisions-"); - db = createDb(tempDb.connectionString); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decisions-"); + db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }, 20_000); beforeEach(async () => { @@ -69,13 +80,14 @@ describePg("decisionService", () => { }); afterEach(async () => { + await stopWorker?.(); stopWorker = undefined; delete process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE; delete process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS; delete process.env.PAPERCLIP_AGENT_JWT_SECRET; await db.delete(decisionEffectExecutions); await db.delete(decisionTargetIssues); await db.delete(decisions); await db.delete(decisionRetention); await db.delete(activityLog); await db.delete(issueComments); await db.delete(issueRelations); await db.delete(heartbeatRuns); await db.delete(issues); await db.delete(agents); await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies); }); - afterAll(async () => tempDb?.cleanup()); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); const agentActor = () => ({ type: "agent" as const, companyId, agentId, runId, source: "agent_jwt" as const, onBehalfOfUserId: originResponsibleUserId, onBehalfOfMemberships: [{ companyId, membershipRole: "member", status: "active" }] }); @@ -656,4 +668,210 @@ describePg("decisionService", () => { expect(result).toMatchObject({ status: "decided", chosenOptionId: "dismissed" }); expect(wakes).toEqual([{ companyId, agentId, issueId: originIssueId, decisionId: created.id, outcome: "decided" }]); }); + describe("continuation worker", () => { + let coordinator: ReturnType | undefined; + let draining: boolean; + const errors: unknown[] = []; + const wake = vi.fn(async (_input: Record) => {}); + const workerService = () => decisionService(db, { wakeOriginAgent: wake }); + + beforeEach(async () => { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + draining = false; errors.length = 0; wake.mockReset(); + stopWorker = async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists decision_ack_fail on decisions`); + await db.execute(sql`drop function if exists decision_ack_fail()`); + expect(errors).toEqual([]); + }; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerDecisionContinuationWork(coordinator, workerService(), () => !draining).ready; + } + async function quiet() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(60 * 60_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + } + async function pending(id: string) { + expect((await service().get(id))?.metadata.continuationPending).toBe(true); + } + async function delivered(id: string) { + expect((await service().get(id))?.metadata.continuationPending).toBe(false); + } + const markReady = (id: string, source = db) => writeDecisionChange(source, async tx => tx.update(decisions) + .set({ status: "cancelled", metadata: { continuationPending: true } }).where(eq(decisions.id, id))); + + it.each(["decide", "cancel", "dismiss", "expire"] as const)("reactivates from real %s and leaves delivered history quiet", async action => { + const row = await createCommentDecision(); + await start(); await quiet(); + const svc = workerService(); + if (action === "decide") await svc.decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() }); + if (action === "cancel") await svc.cancel(row.id, { actorType: "agent", actorId: agentId }); + if (action === "dismiss") await svc.dismiss(row.id, decidedByUserId, boardActor()); + if (action === "expire") { await expireDecisionNow(row.id); await svc.sweepExpired(new Date(), { includeContinuations: false }); } + await quiet(); await delivered(row.id); + expect(wake).toHaveBeenCalledTimes(1); + expect(wake.mock.calls[0][0]).toMatchObject({ decisionId: row.id, companyId, agentId, issueId: originIssueId }); + }); + + it("stays quiet while open or running, then wakes when effect completion commits", async () => { + const row = await createCommentDecision(); + await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes", + decidedByUserId, metadata: { continuationPending: true } }).where(eq(decisions.id, row.id)); + await start(); await quiet(); expect(wake).not.toHaveBeenCalled(); + await workerService().decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() }); + await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); + }); + + it("coalesces simultaneous foreground replay and worker delivery without replaying effects", async () => { + const row = await createCommentDecision(); + await start(); + await Promise.all(Array.from({ length: 3 }, () => workerService().decide({ + id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor(), + }).catch(error => { if (error.message !== "decision_already_resolved") throw error; }))); + await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); + }); + + it("shares an in-flight wake with a dedicated child pool and rereads after completion", async () => { + const row = await createCommentDecision(); await markReady(row.id); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + wake.mockImplementationOnce(async () => gate); + const first = decisionContinuationService(db, wake).deliver(row.id); + const childWake = vi.fn(async () => {}); + let child: Promise; + try { + await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1)); + child = withDedicatedDbConnection(db, async connection => { + const attempt = decisionContinuationService(connection, childWake).deliver(row.id); + expect(attempt).toBe(first); + await attempt; + }); + await vi.advanceTimersByTimeAsync(100); + } finally { release(); } + await first; await child!; + await decisionContinuationService(db, childWake).deliver(row.id); + expect(childWake).not.toHaveBeenCalled(); await delivered(row.id); + }); + + it("reconstructs pending delivery at restart without replaying decision effects", async () => { + const row = await createCommentDecision(); + wake.mockRejectedValueOnce(new Error("wake unavailable")); + await expect(workerService().decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() })).rejects.toThrow("wake unavailable"); + await pending(row.id); await start(); await quiet(); await delivered(row.id); + expect(wake).toHaveBeenCalledTimes(2); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); + }); + + it("retries a failed acknowledgement with the existing at-least-once wake contract", async () => { + const row = await createCommentDecision(); await start(); + await db.execute(sql`create function decision_ack_fail() returns trigger language plpgsql as $$ begin + if NEW.metadata ->> 'continuationPending' = 'false' then raise exception 'ack unavailable'; end if; + return NEW; end $$`); + await db.execute(sql`create trigger decision_ack_fail before update on decisions for each row execute function decision_ack_fail()`); + await expect(workerService().cancel(row.id, { actorType: "agent", actorId: agentId })).rejects.toMatchObject({ cause: { message: "ack unavailable" } }); + await pending(row.id); + await vi.advanceTimersByTimeAsync(60_000); + await db.execute(sql`drop trigger decision_ack_fail on decisions`); + await quiet(); await delivered(row.id); + expect(wake.mock.calls.length).toBeGreaterThan(1); + for (const error of errors) expect(error).toMatchObject({ cause: { message: "ack unavailable" } }); + errors.length = 0; + }); + + it("waits for an outer transaction, ignores rollback, and recovers the committed write", async () => { + const row = await createCommentDecision(); await start(); await quiet(); + await expect(db.transaction(async tx => { + await markReady(row.id, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(60_000); + expect(wake).not.toHaveBeenCalled(); throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); expect((await service().get(row.id))?.status).toBe("open"); + await db.transaction(async tx => { + await markReady(row.id, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(60_000); expect(wake).not.toHaveBeenCalled(); + }); + await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("retains a first write during drain without scanning and delivers after resume", async () => { + const row = await createCommentDecision(); await start(); await quiet(); draining = true; + await markReady(row.id); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); expect(wake).not.toHaveBeenCalled(); + draining = false; await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("retains a failed wake through drain and resumes its retry", async () => { + const row = await createCommentDecision(); await start(); + wake.mockImplementationOnce(async () => { draining = true; throw new Error("wake unavailable"); }); + await markReady(row.id); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors).toHaveLength(1); }); + await pending(row.id); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; errors.length = 0; await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(2); + }); + + it("preserves changes arriving during delivery and stops after shutdown", async () => { + const firstRow = await createCommentDecision(), secondRow = await createCommentDecision(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + wake.mockImplementationOnce(async () => { await markReady(secondRow.id); await gate; }); + await markReady(firstRow.id); + const starting = start(); + try { await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1)); } + finally { release(); } + await starting; await quiet(); await delivered(firstRow.id); await delivered(secondRow.id); + expect(wake).toHaveBeenCalledTimes(2); + await coordinator!.stop(); coordinator = undefined; + await vi.advanceTimersByTimeAsync(60_000); expect(wake).toHaveBeenCalledTimes(2); + }); + + it("awaits an admitted wake before completing shutdown", async () => { + const row = await createCommentDecision(); await markReady(row.id); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + wake.mockImplementationOnce(async () => gate); + const starting = start(); + let stopped = false; + let stopping: Promise; + try { + await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1)); + stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(60_000); + expect(stopped).toBe(false); + } finally { release(); } + await starting; await stopping!; coordinator = undefined; + await delivered(row.id); expect(stopped).toBe(true); + }); + + it("pages past a failed first candidate and preserves its retry", async () => { + const seed = await createCommentDecision(); + await db.delete(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, seed.id)); + await db.delete(decisions).where(eq(decisions.id, seed.id)); + const ids = Array.from({ length: 101 }, (_, i) => `10000000-0000-4000-8000-${String(i).padStart(12, "0")}`); + await db.insert(decisions).values(ids.map(id => ({ ...seed, id, status: "cancelled", metadata: { continuationPending: true } }))); + wake.mockImplementation(async input => { if (input.decisionId === ids[0]) throw new Error("first unavailable"); }); + await start(); + expect(wake).toHaveBeenCalledTimes(101); await delivered(ids.at(-1)!); await pending(ids[0]); + expect(errors).toHaveLength(1); errors.length = 0; wake.mockImplementation(async () => {}); + await quiet(); await delivered(ids[0]); expect(wake).toHaveBeenCalledTimes(102); + }); + }); + }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 09a2e87ec47..c07136633d7 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -5,7 +5,7 @@ import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import * as idleLocalWork from "../services/idle-local-work.js"; import { idleBackupWakeMarker, markIdleBackupWakeRequired } from "../services/idle-database-backup.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; -import { reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; +import { decisionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; import { runDatabaseBackup } from "@paperclipai/db"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; @@ -340,6 +340,7 @@ vi.mock("../services/index.js", () => ({ })), decisionService: vi.fn(() => ({ sweepExpired: vi.fn(async () => ({ expired: 0 })), + sweepContinuations: vi.fn(async () => ({ retry: false })), })), decisionRetentionService: vi.fn(() => ({ autoArchive: vi.fn(async () => 0), @@ -855,6 +856,10 @@ describe("startServer feedback export wiring", () => { expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.customImageExpiry)).toBe(true); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(true); + const decisions = vi.mocked(decisionService).mock.results.at(-1)!.value; + expect(decisions.sweepContinuations).toHaveBeenCalledTimes(1); + expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false }); } finally { setIntervalSpy.mockRestore(); } @@ -891,6 +896,7 @@ describe("startServer feedback export wiring", () => { expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled(); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(false); } finally { setIntervalSpy.mockRestore(); } diff --git a/server/src/index.ts b/server/src/index.ts index 2c69e3d7a18..2b47ad4d553 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerDecisionContinuationWork } from "./services/decision-continuation-work.js"; import { registerDecisionArchiveDeliveryWork } from "./services/decision-archive-delivery-work.js"; import { registerToolReviewProjectionWork } from "./services/tool-review-projection-work.js"; import { registerToolReviewInterruptionWork } from "./services/tool-review-interruption-work.js"; @@ -1800,7 +1801,9 @@ async function startServerWithDatabaseTeardown( logger.warn({ ...toolHealthSweep }, "startup tool connection health sweep found failing connections"); } }); - await step("decision_expiry", () => decisionExecutor.sweepExpired()); + await step("decision_continuation_worker", () => registerDecisionContinuationWork(deliveryWork, decisionExecutor, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); + await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false })); const loginSchedule = loginCleanupSchedule(db, config.heartbeatSchedulerIntervalMs); await step("login_cleanup_workers", () => registerLoginCleanupWork(deliveryWork, [ { queue: DELIVERY_QUEUES.adapterLoginCleanup, run: () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult), next: loginSchedule.adapter }, @@ -1821,7 +1824,7 @@ async function startServerWithDatabaseTeardown( trackHeartbeatSchedulerWork((async () => { if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleRoutineStatusReconcile(); - trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired().catch((err: unknown) => { + trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false }).catch((err: unknown) => { logger.error({ err }, "decision expiry sweep failed"); })); trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => { diff --git a/server/src/services/decision-continuation-work.ts b/server/src/services/decision-continuation-work.ts new file mode 100644 index 00000000000..78dfbb6eef2 --- /dev/null +++ b/server/src/services/decision-continuation-work.ts @@ -0,0 +1,20 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import type { decisionService } from "./decisions.js"; + +export function registerDecisionContinuationWork( + coordinator: ReturnType, + service: Pick, "sweepContinuations">, + canRun: () => boolean, +) { + let next: number | null = null; + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionContinuation, { + retryMs: 30_000, canRun, + run: async () => { + const result = await service.sweepContinuations(); + next = result.retry ? Date.now() + 30_000 : null; + }, + next: async () => next, + }); +} diff --git a/server/src/services/decision-continuations.ts b/server/src/services/decision-continuations.ts new file mode 100644 index 00000000000..d703273560f --- /dev/null +++ b/server/src/services/decision-continuations.ts @@ -0,0 +1,63 @@ +import { and, asc, eq, gt, sql } from "drizzle-orm"; +import { databaseWorkScopeKey, decisions, type Db } from "@paperclipai/db"; + +export type DecisionContinuationWake = (input: { + companyId: string; agentId: string; issueId: string; decisionId: string; + outcome: "decided" | "expired" | "cancelled"; +}) => Promise; + +const flights = new WeakMap>>(); +const ready = sql`${decisions.continuationPolicy} = 'wake_origin_agent' + AND ${decisions.metadata} ->> 'continuationPending' = 'true' + AND (${decisions.status} IN ('expired', 'cancelled') OR + (${decisions.status} = 'decided' AND ${decisions.executionStatus} IN ('succeeded', 'partial', 'failed')))`; + +export function decisionContinuationService(db: Db, wake: DecisionContinuationWake) { + const scope = databaseWorkScopeKey(db); + let pending = flights.get(scope); + if (!pending) flights.set(scope, pending = new Map()); + const inFlight = pending; + + function deliver(id: string): Promise { + const existing = inFlight.get(id); + if (existing) return existing; + const attempt = (async () => { + const [decision] = await db.select().from(decisions).where(and(eq(decisions.id, id), ready)); + if (!decision) return; + await wake({ companyId: decision.companyId, agentId: decision.originAgentId, + issueId: decision.originIssueId, decisionId: id, + outcome: decision.status === "expired" ? "expired" : decision.status === "cancelled" ? "cancelled" : "decided" }); + await db.update(decisions).set({ + metadata: sql`coalesce(${decisions.metadata}, '{}'::jsonb) || jsonb_build_object( + 'continuationPending', false, 'continuationDeliveredAt', ${new Date().toISOString()}::text)`, + updatedAt: new Date(), + }).where(and(eq(decisions.id, id), ready)); + })().finally(() => { inFlight.delete(id); }); + inFlight.set(id, attempt); + return attempt; + } + + async function sweep(maxRows = Infinity) { + let cursor: string | undefined; + let attempted = 0; + const errors: unknown[] = []; + while (attempted < maxRows) { + const limit = Math.min(100, maxRows - attempted); + const rows = await db.select({ id: decisions.id }).from(decisions) + .where(and(ready, cursor ? gt(decisions.id, cursor) : undefined)) + .orderBy(asc(decisions.id)).limit(limit); + for (const row of rows) { + try { await deliver(row.id); } catch (error) { errors.push(error); } + } + attempted += rows.length; + if (rows.length < limit) break; + cursor = rows.at(-1)!.id; + } + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Decision continuation delivery failed"); + const remaining = await db.select({ id: decisions.id }).from(decisions).where(ready).limit(1); + return { retry: remaining.length > 0 }; + } + + return { deliver, sweep }; +} diff --git a/server/src/services/decision-work-notifications.ts b/server/src/services/decision-work-notifications.ts new file mode 100644 index 00000000000..af1ac08487f --- /dev/null +++ b/server/src/services/decision-work-notifications.ts @@ -0,0 +1,9 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export function writeDecisionChange(db: Db, write: (transaction: Db) => Promise): Promise { + return db.transaction(async transaction => { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionContinuation); + return write(transaction as unknown as Db); + }); +} diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index c9a57063548..280ccee9e4c 100644 --- a/server/src/services/decisions.ts +++ b/server/src/services/decisions.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { and, asc, count, desc, eq, gt, gte, inArray, lte, or, sql } from "drizzle-orm"; +import { and, asc, count, desc, eq, gt, gte, inArray, lte, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db"; import { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared"; @@ -14,10 +14,13 @@ import { } from "./issues.js"; import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js"; +import { decisionContinuationService, type DecisionContinuationWake } from "./decision-continuations.js"; +import { writeDecisionChange } from "./decision-work-notifications.js"; + type Snapshot = { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; updatedAt: string; descendantCount?: number; descendantIds?: string[]; childCount?: number; attentionArchive?: unknown }; -type Wake = (input: { companyId: string; agentId: string; issueId: string; decisionId: string; outcome: "decided" | "expired" | "cancelled" }) => Promise; -export type DecisionServiceOptions = { wakeOriginAgent: Wake }; + +export type DecisionServiceOptions = { wakeOriginAgent: DecisionContinuationWake }; const DAY = 86_400_000; function targetIds(options: DecisionOption[]) { @@ -118,6 +121,7 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) { } export function decisionService(db: Db, options: DecisionServiceOptions) { + const continuations = decisionContinuationService(db, options.wakeOriginAgent); const authz = authorizationService(db); let targetSweepCursor: string | null = null; type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null; @@ -555,8 +559,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { options: decision.options, targetSnapshots: decision.targetSnapshots as Record, }), decision.signedSpec)) { - await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature" } }) - .where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature" } }) + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } if (option.id === "archive") { @@ -566,8 +570,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { ); const manifestHash = manifest ? hashAttentionArchiveManifest(manifest) : null; if (!manifest || manifestHash !== metadata.manifestHash) { - await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch" } }) - .where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch" } }) + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } const originActor: AuthorizationActor = { @@ -588,50 +592,30 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { }); } catch (error) { const message = error instanceof Error ? error.message : "archive_proposal_failed"; - await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message } }) - .where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message } }) + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } } - await db.update(decisions).set({ + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "succeeded", updatedAt: new Date(), metadata: { ...metadata, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) }, - }).where(eq(decisions.id, decision.id)); + }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } for (let index = 0; index < option.effects.length; index += 1) await executeEffect(decision, option.effects[index]!, index, userActor, decision.decidedByUserId, run?.responsibleUserId ?? null); const rows = await db.select().from(decisionEffectExecutions).where(eq(decisionEffectExecutions.decisionId, decision.id)); const successful = rows.filter((row) => row.status === "executed").length; const status = rows.every((row) => row.status === "executed") ? "succeeded" : successful ? "partial" : "failed"; - await db.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata, - ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata, + ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } - async function deliverContinuation( - decision: typeof decisions.$inferSelect, - outcome: "decided" | "expired" | "cancelled", - ) { - const metadata = decision.metadata as Record; - if (decision.continuationPolicy !== "wake_origin_agent" || metadata.continuationPending !== true) return; - // Delivery is intentionally at-least-once: a crash after wakeup but before - // this acknowledgement can retry, while a crash before wakeup cannot lose - // the continuation. Heartbeat wakeups coalesce concurrent queued work. - await options.wakeOriginAgent({ companyId: decision.companyId, agentId: decision.originAgentId, - issueId: decision.originIssueId, decisionId: decision.id, outcome }); - await db.update(decisions).set({ - metadata: sql`coalesce(${decisions.metadata}, '{}'::jsonb) || jsonb_build_object( - 'continuationPending', false, - 'continuationDeliveredAt', ${new Date().toISOString()}::text - )`, - updatedAt: new Date(), - }).where(and(eq(decisions.id, decision.id), sql`${decisions.metadata} ->> 'continuationPending' = 'true'`)); - } - async function resumeDecision(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) { const result = decision.executionStatus === "running" ? await runEffects(decision, userActor) : await outcome(decision.id); - await deliverContinuation(result, "decided"); + await continuations.deliver(result.id); return outcome(decision.id); } @@ -651,25 +635,25 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { } if (current.status !== "open") throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); if (current.expiresAt <= new Date()) { - const [expired] = await db.update(decisions).set({ status: "expired", updatedAt: new Date(), metadata: { ...metadata, expiredReason: "ttl", + const [expired] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "expired", updatedAt: new Date(), metadata: { ...metadata, expiredReason: "ttl", ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) - .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"), lte(decisions.expiresAt, new Date()))).returning(); + .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"), lte(decisions.expiresAt, new Date()))).returning()); if (expired) { await logActivity(db, { companyId: expired.companyId, actorType: "system", actorId: "decision-expiry-sweeper", agentId: expired.originAgentId, runId: expired.originRunId, action: "decision.expired", entityType: "decision", entityId: expired.id, details: { expiredReason: "ttl" } }); - await deliverContinuation(expired, "expired"); + await continuations.deliver(expired.id); } throw conflict("decision_expired", { code: "decision_expired" }); } if (!current.options.some((option) => option.id === input.optionId)) throw unprocessable("Unknown optionId"); const values = input.inputValues ?? {}; for (const field of current.inputs ?? []) { const value = values[field.id] ?? ""; if (field.required && !value.trim()) throw unprocessable(`Input ${field.id} is required`); if (field.maxLength && value.length > field.maxLength) throw unprocessable(`Input ${field.id} is too long`); } - const [claimed] = await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: input.optionId, inputValues: values, + const [claimed] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: input.optionId, inputValues: values, decidedByUserId: input.decidedByUserId, decidedAt: new Date(), updatedAt: new Date(), metadata: { ...metadata, decideIdempotencyKey: input.idempotencyKey ?? null, ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}), ...(input.dismissed ? { dismissed: true, dismissReason: input.dismissReason ?? null } : {}) } }) - .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning(); + .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning()); if (!claimed) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, claimed.originRunId)).then((rows) => rows[0] ?? null); await logActivity(db, { companyId: claimed.companyId, actorType: "system", actorId: "decision-executor", agentId: claimed.originAgentId, runId: claimed.originRunId, @@ -683,14 +667,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { const current = await get(id); if (!current) throw notFound("Decision not found"); if (actor.actorType === "agent" && actor.actorId !== current.originAgentId) throw forbidden("Only the origin agent may cancel"); if (current.status === "cancelled") { - await deliverContinuation(current, "cancelled"); + await continuations.deliver(current.id); return (await get(id))!; } - const [updated] = await db.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata, - ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(); + const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata, + ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning()); if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); await logActivity(db, { companyId: updated.companyId, actorType: actor.actorType, actorId: actor.actorId, runId: actor.runId, action: "decision.cancelled", entityType: "decision", entityId: id }); - await deliverContinuation(updated, "cancelled"); + await continuations.deliver(updated.id); return (await get(id))!; } @@ -699,14 +683,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { if (!verifyDecisionSpec(spec({ id: current.id, options: current.options, targetSnapshots: current.targetSnapshots as Record }), current.signedSpec)) throw forbidden("Decision signature verification failed"); const empty = current.options.find((option) => option.effects.length === 0); if (empty) return decide({ id, optionId: empty.id, decidedByUserId: userId, userActor, dismissed: true, dismissReason: reason }); - const [updated] = await db.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId, + const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId, decidedAt: new Date(), updatedAt: new Date(), metadata: { ...current.metadata, dismissed: true, dismissReason: reason ?? null, - ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(); + ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning()); if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-executor", agentId: updated.originAgentId, runId: updated.originRunId, responsibleUserIdOverride: userId, action: "decision.dismissed", entityType: "decision", entityId: updated.id, details: { chosenOptionId: "dismissed", decidedByUserId: userId, dismissed: true } }); - await deliverContinuation(updated, "decided"); + await continuations.deliver(updated.id); return outcome(id); } @@ -718,7 +702,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { return { ...bundle, decisions: created }; }); } - async function sweepExpired(now = new Date()) { + async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean } = {}) { const configuredBatchSize = Number(process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE ?? 100); const batchSize = Number.isFinite(configuredBatchSize) ? Math.max(1, Math.trunc(configuredBatchSize)) @@ -737,17 +721,9 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { const recovered = await runEffects(decision, await recoveryActor(decision.companyId, decision.decidedByUserId)); if (recovered.executionStatus === "running") continue; resumed += 1; - await deliverContinuation(recovered, "decided"); - } - const pendingContinuations = await db.select().from(decisions) - .where(and(eq(decisions.continuationPolicy, "wake_origin_agent"), - sql`${decisions.metadata} ->> 'continuationPending' = 'true'`, - or(inArray(decisions.status, ["expired", "cancelled"]), and(eq(decisions.status, "decided"), - inArray(decisions.executionStatus, ["succeeded", "partial", "failed"]))))) - .orderBy(asc(decisions.updatedAt)).limit(batchSize); - for (const decision of pendingContinuations) { - await deliverContinuation(decision, decision.status === "expired" ? "expired" : decision.status === "cancelled" ? "cancelled" : "decided"); + await continuations.deliver(recovered.id); } + if (options.includeContinuations !== false) await continuations.sweep(batchSize); const ttlRows = await db.select().from(decisions) .where(and(eq(decisions.status, "open"), lte(decisions.expiresAt, now))) .orderBy(asc(decisions.expiresAt)).limit(batchSize); @@ -774,13 +750,13 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { targets.some((target) => snapshots[target.id]?.status !== "done"); if (!targetGone && !targetsCompleted && decision.expiresAt >= now) continue; const reason = targetGone ? "target_gone" : targetsCompleted ? "target_completed" : "ttl"; - const [updated] = await db.update(decisions).set({ status: "expired", updatedAt: now, metadata: { ...decision.metadata, expiredReason: reason, - ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.status, "open"))).returning(); + const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "expired", updatedAt: now, metadata: { ...decision.metadata, expiredReason: reason, + ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.status, "open"))).returning()); if (!updated) continue; expired += 1; await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-expiry-sweeper", agentId: updated.originAgentId, runId: updated.originRunId, action: "decision.expired", entityType: "decision", entityId: updated.id, details: { expiredReason: reason } }); - await deliverContinuation(updated, "expired"); } + await continuations.deliver(updated.id); } return { expired, resumed }; } - return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired }; + return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired, sweepContinuations: continuations.sweep }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 6fde2487eca..99ed29bb409 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + decisionContinuation: "decision-continuation", decisionArchiveDelivery: "decision-archive-delivery", toolReviewProjection: "tool-review-projection", toolReviewInterruption: "tool-review-interruption",