diff --git a/doc/idle-execution-control.md b/doc/idle-execution-control.md new file mode 100644 index 00000000000..a18600124ec --- /dev/null +++ b/doc/idle-execution-control.md @@ -0,0 +1,71 @@ +# Execution-control workers and idle instances + +Execution-control maintenance uses the shared delivery coordinator instead of an +unconditional 15-second interval. This changes when recovery is inspected; it +does not change the authority, evidence, locking, idempotency, or cleanup rules +inside the six owning services. No table or migration is added. + +## Duties and durable state + +| Duty | Trigger and durable source | Remaining scheduled work | Quiet condition | +| --- | --- | --- | --- | +| Abandoned finalization | Committed control-deadline writes on `heartbeat_runs` and `native_run_finalizations` | Earliest non-null control deadline; a due row still present after recovery retries after 15 seconds | No control deadlines | +| Safe native replacement | Committed run transitions and reconciliation decisions; existing failed run/coordinator rows | Retry while the existing replacement predicate selects a candidate, including one awaiting provider termination or sandbox release | No replacement candidate, or a recorded successor/denial | +| Reconciled continuation | Committed recovery-action marking/resolution; `evidence.continuationDelivery` | Retry pending delivery after 15 seconds | No resolved action awaiting continuation delivery | +| Execution status | Committed run status writes; `execution_status_delivery_id` | Retry while a delivery marker remains | All markers acknowledged | +| Automatic disposition | Committed run/recovery-action changes; existing eligible recovery-action queries | Retry while a selected action still needs disposition or an obsolete conversation hold remains | No eligible action/hold; resolved history alone is quiet | +| Local login cleanup | Committed local login creation; existing `adapter_auth_sessions.expires_at` | Earliest local login expiry; retry overdue failed cleanup | No local login expiry remains | + +Each worker runs once on startup to restore durable obligations. Passes are +coalesced at the former 15-second resolution: a tenant's repeated writes cannot +force continuous global scans. This is a minimum spacing between requested +passes, not a recurring poll; an empty worker never arms that spacing by itself. +A newly discovered deadline within the spacing window runs at its end, retaining +the former maximum 15-second scheduling resolution. Future deadlines outside +that window retain their exact due time. Workers remain +independent: a slow replacement proof cannot stall status delivery. Batches and +failed obligations retain the existing 15-second retry budget; empty workers do +not repeatedly inspect the database. Login authorization still checks expiry on +use even before physical cleanup completes. + +## Producer contract + +`notifyExecutionRunWork(transaction)` signals the run transition's status, +finalization, replacement, and disposition consumers before the write. Existing +transaction owners call it directly. Standalone deadline and recovery-action +writes use `writeExecutionControl` to give the write a transaction boundary. +Nested transactions retain the existing outer-commit notification semantics. + +Producer coverage includes heartbeat status transitions, legacy terminalization, +native finalization and restart/failure settlement, replacement successors and +denials, recovery-action upserts/resolution, operator reconciliation, control +deadline creation/cancellation, and local login creation. Workers acknowledge +existing records rather than creating a second queue. The replacement and +disposition pending checks reuse the executing services' eligibility queries. + +This is the existing single live server / shared `createDb` owner signal model. +Independent SQL writers or separate replicas must integrate notifications; these +workers do not continuously discover unannounced external writes. Startup still +recovers committed rows after process loss. Uncertain commit handling remains in +`packages/db/src/work-signals.ts`. + +## Lifecycle and verification + +A worker consumes a requested generation only after its sweep and deadline +inspection succeed. Writes during a sweep, startup during drain, and a failure +before drain remain pending for resume. Passive waiting during drain performs no +worker SQL. Shutdown uses the coordinator to unsubscribe and await in-flight work. + +`server/src/__tests__/execution-control-work.test.ts` covers empty and retained +history with the real sweep implementations, a real producer inside an outer +commit/rollback, saved deadlines, deadline replacement/cancellation, durable +status retries, recovery continuations, independent duties, and drain/resume. +Database tests use embedded PostgreSQL by default. An isolated external test +instance can be supplied through `PAPERCLIP_EXECUTION_SLEEP_TEST_DATABASE_URL`; +the database must have a dedicated name ending in `_test` and is cleared by the +suite. Never use a tenant database. + +Worker quiet is not host sleep authorization. The existing idle-safety blockers +remain, and Cloud's external wake contract does not gain coverage for execution +or login deadlines through this change. Provider suspension and wake must be +qualified separately before relaxing those blockers. diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index 35db2511e4a..e77a8719c71 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -536,13 +536,13 @@ Longer polling intervals alone do not complete an item. retain subscription scope, expiry, and replay safety. Source: [events.ts](../../server/src/services/public-mcp/events.ts). -- [ ] **JOB-06 — Execution-control reconciliation.** Current: independent - **15s** recovery/delivery lanes, including when the main heartbeat is disabled. - Change: activate lanes from durable pending work and deadlines. **Complete - when:** empty lanes are quiet and abandoned finalizations, replacements, - continuations, status/disposition delivery, and login expiry still recover. - Sources: [index.ts](../../server/src/index.ts), - [execution-control-deadline.ts](../../server/src/services/execution-control-deadline.ts). +- [x] **JOB-06 — Execution-control reconciliation.** Implemented in + [PR #17094](https://github.com/paperclipai/paperclip/pull/17094): six independent + workers wake from committed writes, saved control/login deadlines, or concrete + unfinished obligations. Empty queues and completed history no longer cause + recurring SQL. Existing 15s retries remain for pending work; Cloud sleep + eligibility and external alarm coverage are unchanged. See the + [worker and producer contract](../idle-execution-control.md). - [ ] **JOB-07 — Plugin job scheduling.** Current: **30s** SQL even without jobs. Change: register nearest `nextRunAt` and signal registration/update/removal. diff --git a/server/src/__tests__/execution-control-work.test.ts b/server/src/__tests__/execution-control-work.test.ts new file mode 100644 index 00000000000..9d107b75dd7 --- /dev/null +++ b/server/src/__tests__/execution-control-work.test.ts @@ -0,0 +1,289 @@ +import { writeExecutionControl } from "../services/execution-control-notifications.js"; +import { reconcileSafeNativeReplacements } from "../services/native-runtime/native-safe-replacement.js"; +import { localAiLoginService } from "../services/local-ai-login.js"; +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, applyPendingMigrations } from "@paperclipai/db"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js"; +import { executionControlSchedule, registerExecutionControlWork } from "../services/execution-control-work.js"; +import { deliverExecutionStatuses } from "../services/execution-status-delivery.js"; +import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js"; +import { reconcileAbandonedExecutionControl } from "../services/execution-control-reconciliation.js"; +import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { markExecutionReconciliation, deliverReconciledExecutions, settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const coordinators: ReturnType[] = []; +afterEach(async () => { for (const worker of coordinators.splice(0)) await worker.stop(); vi.useRealTimers(); }); +function coordinator(owner: object, canRun = () => true, onError: (error: unknown) => void = vi.fn()) { + const c = createDeliveryWorkCoordinator({ owner, canRun, canReconcile: () => true, onError }); + coordinators.push(c); return c; +} +function fakeOwner() { + const execute = vi.fn(async () => [{ xid: "42" }]); + const owner = installDatabaseWorkSignals({ execute, async transaction(fn: (tx: any) => Promise): Promise { + return fn({ execute, transaction: owner.transaction }); + } }); + return owner; +} + +describe("execution-control lifecycle", () => { + it.each(["startup", "commit", "failure"])("retains %s work through drain without querying while paused", async kind => { + vi.useFakeTimers(); + const owner = fakeOwner(); let draining = kind === "startup"; + const errors = vi.fn(); const c = coordinator(owner, () => !draining, errors); + const run = vi.fn(async () => {}), next = vi.fn(async () => null); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next }]).ready; + if (kind !== "startup") { + if (kind === "failure") run.mockRejectedValueOnce(new Error("delivery unavailable")); + else draining = true; + await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus)); + await vi.advanceTimersByTimeAsync(15_000); + draining = true; + } + const reads = next.mock.calls.length, runs = run.mock.calls.length; + await vi.advanceTimersByTimeAsync(60_000); + expect(next).toHaveBeenCalledTimes(reads); expect(run).toHaveBeenCalledTimes(runs); + draining = false; + await vi.advanceTimersByTimeAsync(15_000); + expect(run).toHaveBeenCalledTimes(runs + 1); + expect(c.nextWakeAt()).toBeNull(); + }); + it("keeps a change arriving during a sweep and does not overlap it", async () => { + vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const run = vi.fn(async () => {}); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next: async () => null }]).ready; + run.mockImplementationOnce(() => gate); + await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus)); + await vi.advanceTimersByTimeAsync(15_000); + await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus)); + await vi.advanceTimersByTimeAsync(1); + expect(run).toHaveBeenCalledTimes(2); + release(); await vi.advanceTimersByTimeAsync(15_000); + expect(run).toHaveBeenCalledTimes(3); expect(c.nextWakeAt()).toBeNull(); + }); + it("coalesces tenant write bursts without repeatedly scanning global queues", async () => { + vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner); + const run = vi.fn(async () => {}), next = vi.fn(async () => null); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next }]).ready; + for (let i = 0; i < 100; i++) { + await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus, `company-${i % 2}`)); + await vi.advanceTimersByTimeAsync(10); + } + expect(run).toHaveBeenCalledTimes(1); expect(next).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(14_000); + expect(run).toHaveBeenCalledTimes(2); expect(next).toHaveBeenCalledTimes(2); + expect(c.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(60_000); + expect(run).toHaveBeenCalledTimes(2); + }); + it.each([false, true])("does not start sweeps or inspect deadlines once server shutdown begins (notification=%s)", async notification => { + vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner); let stopping = false; + const run = vi.fn(async () => {}), next = vi.fn(async () => Date.now() + 1000); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next }], () => !stopping).ready; + stopping = true; + if (notification) await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus)); + await vi.advanceTimersByTimeAsync(30_000); + expect(run).toHaveBeenCalledTimes(1); expect(next).toHaveBeenCalledTimes(1); + await c.stop(); expect(c.nextWakeAt()).toBeNull(); + }); + it("keeps other obligations responsive while one is waiting, and awaits shutdown", async () => { + vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner); + let release!: () => void; const gate = new Promise(resolve => { release = resolve; }); + const status = vi.fn(async () => {}); + const started = registerExecutionControlWork(c, [ + { queue: DELIVERY_QUEUES.executionReplacement, run: () => gate, next: async () => null }, + { queue: DELIVERY_QUEUES.executionStatus, run: status, next: async () => null }, + ]); + await vi.advanceTimersByTimeAsync(1); expect(status).toHaveBeenCalledTimes(1); + let stopped = false; const stopping = c.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1); expect(stopped).toBe(false); + release(); await started.ready; await stopping; expect(stopped).toBe(true); + }); +}); + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXECUTION_SLEEP_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("execution-control durable work", () => { + let db: ReturnType, observed: typeof db; + let temporary: Awaited> | undefined; + let calls: number, companyId: string, agentId: string, issueId: string; + const errors: unknown[] = []; + beforeAll(async () => { + if (external) await applyPendingMigrations(external); + else temporary = await startEmbeddedPostgresTestDatabase("execution-sleep-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 }); + }, 90000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + beforeEach(async () => { + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + companyId = randomUUID(); agentId = randomUUID(); issueId = randomUUID(); calls = 0; errors.length = 0; + await db.insert(companies).values({ id: companyId, name: "Idle execution", issuePrefix: "EXE" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "codex_local" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Execution", status: "in_progress", assigneeAgentId: agentId }); + observed = new Proxy(db, { get(target, key) { + if (!["select", "update", "insert", "delete", "execute", "transaction"].includes(String(key))) return Reflect.get(target, key); + return (...args: unknown[]) => { calls++; return Reflect.apply(Reflect.get(target, key), target, args); }; + } }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + async function settle() { + await vi.advanceTimersByTimeAsync(15_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 }); + await vi.advanceTimersByTimeAsync(1); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 }); + expect(errors).toEqual([]); + } + async function quiet(c: ReturnType) { + await settle(); expect(c.nextWakeAt()).toBeNull(); const before = calls; + await vi.advanceTimersByTimeAsync(90_000); expect(calls).toBe(before); + } + async function runRow(status = "running") { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status, runtimeMode: "legacy", contextSnapshot: { issueId } }).returning(); + return run!; + } + it("stops all six pending/deadline scans with empty queues and completed history", async () => { + await runRow("succeeded"); + const schedule = executionControlSchedule(observed); + const c = coordinator(db, undefined, error => { errors.push(error); }); + const topics = [DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionContinuation, + DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.localLoginCleanup]; + const runs = [() => reconcileAbandonedExecutionControl(observed), () => reconcileSafeNativeReplacements(observed), + () => deliverReconciledExecutions(observed, vi.fn(async () => null)), () => deliverExecutionStatuses(observed), + () => settleUnrecoverableExecutions(observed), () => localAiLoginService(observed).reapExpired()]; + await registerExecutionControlWork(c, Object.values(schedule).map((next, i) => ({ queue: topics[i]!, run: runs[i]!, next }))).ready; + await quiet(c); + }); + it("bounds real producer bursts with another company's completed history present", async () => { + const otherCompany = randomUUID(), otherAgent = randomUUID(); + await db.insert(companies).values({ id: otherCompany, name: "Other tenant", issuePrefix: "OTH" }); + await db.insert(agents).values({ id: otherAgent, companyId: otherCompany, name: "Other worker", adapterType: "codex_local" }); + await db.insert(heartbeatRuns).values({ companyId: otherCompany, agentId: otherAgent, status: "succeeded" }); + const run = await runRow(); const publish = vi.fn(); const c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, + run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status }]).ready; + const before = calls; + for (let i = 0; i < 20; i++) { + await terminalizeLegacyExecution({ db, run, status: "succeeded", fromStatuses: ["running", "succeeded"] }); + await vi.advanceTimersByTimeAsync(10); + } + expect(calls).toBe(before); expect(publish).not.toHaveBeenCalled(); + await settle(); expect(publish).toHaveBeenCalledTimes(1); + expect(publish.mock.calls[0]?.[0]).toMatchObject({ companyId, payload: { runId: run.id } }); + await quiet(c); + }); + it("wakes status delivery through the real producer only after outer commit, then becomes quiet", async () => { + const run = await runRow(); const publish = vi.fn(); const c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, + run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status }]).ready; + await quiet(c); + await db.transaction(async tx => { + await terminalizeLegacyExecution({ db: tx as unknown as typeof db, run, status: "succeeded", patch: { finishedAt: new Date() } }); + await vi.advanceTimersByTimeAsync(1); expect(publish).not.toHaveBeenCalled(); + }); + await settle(); expect(publish).toHaveBeenCalledTimes(1); + await quiet(c); + const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(saved?.status).toBe("succeeded"); expect(saved?.executionStatusDeliveryId).toBeNull(); + }); + it("rolls back a nested producer without delivering or leaving periodic work", async () => { + const run = await runRow(), publish = vi.fn(); const c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, + run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status }]).ready; + await expect(db.transaction(async tx => { + await terminalizeLegacyExecution({ db: tx as unknown as typeof db, run, status: "succeeded" }); + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(c); expect(publish).not.toHaveBeenCalled(); + }); + it("restores an exact control deadline on restart without querying before it", async () => { + const run = await runRow("succeeded"), due = new Date(Date.now() + 60_000); + await db.update(heartbeatRuns).set({ executionControlDeadlineAt: due }).where(eq(heartbeatRuns.id, run.id)); + const c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionFinalization, + run: () => reconcileAbandonedExecutionControl(observed), next: executionControlSchedule(observed).finalization }]).ready; + expect(c.nextWakeAt()).toBe(due.getTime()); const before = calls; + await vi.advanceTimersByTimeAsync(55_000); expect(calls).toBe(before); + await vi.advanceTimersByTimeAsync(5000); await settle(); await quiet(c); + }); + it("retains replacement retries for an unresolved failed execution, excludes completed/denied history", async () => { + const run = await runRow("failed"); + await db.update(heartbeatRuns).set({ nativeIssueId: issueId }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId, issueId, phase: "terminal_failure", failureCode: "provider_transport_failed" }); + const schedule = executionControlSchedule(observed); + expect(await schedule.replacement()).toBeGreaterThan(Date.now()); + await db.update(nativeRunFinalizations).set({ failureDetail: { replacementDenied: "operator_reconciled" } }).where(eq(nativeRunFinalizations.runId, run.id)); + expect(await schedule.replacement()).toBeNull(); + }); + it("a recovery-action producer wakes automatic disposition and leaves resolved history quiet", async () => { + const run = await runRow("failed"); const c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionDisposition, + run: () => settleUnrecoverableExecutions(observed), next: executionControlSchedule(observed).disposition }]).ready; + await quiet(c); + const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", cause: "execution_finalization_deadline_exceeded", fingerprint: run.id, + returnOwnerAgentId: agentId, evidence: { runId: run.id }, nextAction: "Recover execution" }); + await settle(); + expect(await issueRecoveryActionService(db).getActiveForIssue(companyId, issueId)).toBeNull(); + expect(action.id).toBeTruthy(); await quiet(c); + }); + it("retries a failed broadcast only while its durable marker remains, including after restart", async () => { + const run = await runRow(); + const publish = vi.fn(() => { throw new Error("receiver unavailable"); }); + let c = coordinator(db, undefined, error => { errors.push(error); }); + const duty = { queue: DELIVERY_QUEUES.executionStatus, + run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status }; + await registerExecutionControlWork(c, [duty]).ready; + await terminalizeLegacyExecution({ db, run, status: "succeeded" }); + await settle(); expect(publish).toHaveBeenCalledTimes(1); + const before = calls; + await vi.advanceTimersByTimeAsync(10_000); expect(calls).toBe(before); + await c.stop(); + publish.mockImplementation(() => undefined); + c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [duty]).ready; + expect(publish).toHaveBeenCalledTimes(2); await quiet(c); + }); + it("reschedules an earlier deadline and disarms a cancelled deadline", async () => { + const run = await runRow("succeeded"); const c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionFinalization, + run: () => reconcileAbandonedExecutionControl(observed), next: executionControlSchedule(observed).finalization }]).ready; + const due = new Date(Date.now() + 120_000), earlier = new Date(Date.now() + 60_000); + for (const at of [due, earlier]) { + await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns) + .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id))); + await settle(); expect(c.nextWakeAt()).toBe(at.getTime()); + } + await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns) + .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id))); + await quiet(c); + }); + it("delivers a reconciled continuation when marking and resolving commit together", async () => { + const run = await runRow("failed"), successor = await runRow("queued"); + const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", cause: "execution_finalization_deadline_exceeded", fingerprint: run.id, + returnOwnerAgentId: agentId, evidence: { runId: run.id }, nextAction: "Recover execution" }); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, recoveryActionId: action.id, previousRunId: run.id } }) + .where(eq(heartbeatRuns.id, successor.id)); + const wake = vi.fn(async () => successor), c = coordinator(db, undefined, error => { errors.push(error); }); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionContinuation, + run: () => deliverReconciledExecutions(observed, wake), next: executionControlSchedule(observed).continuation }]).ready; + await db.transaction(async tx => { + await markExecutionReconciliation(tx as unknown as typeof db, action, + { runId: run.id, providerStopped: true, actionOutcome: "not_performed", outcomeEvidence: "Verified the provider stopped without performing an external action." }, "operator"); + await issueRecoveryActionService(tx).resolveActiveForIssue({ companyId, sourceIssueId: issueId, actionId: action.id, status: "resolved", outcome: "restored" }); + await vi.advanceTimersByTimeAsync(1); expect(wake).not.toHaveBeenCalled(); + }); + await settle(); expect(wake).toHaveBeenCalledTimes(1); await quiet(c); + const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successor.id)); + expect(saved?.retryOfRunId).toBe(run.id); + }); + +}); diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index b4faf3b74d1..d0d98e6a52c 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -8297,19 +8297,28 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const lease = originalWatch(...args); return { ...lease, stop() { stopped(); lease.stop(); } }; }); - const originalUpdate = db.update.bind(db); + const originalTransaction = db.transaction.bind(db); const cleanupFailure = vi.fn(() => { throw new Error("fixture_deadline_cleanup_failure"); }); - const update = vi.spyOn(db, "update").mockImplementation(((table: typeof heartbeatRuns) => { - const builder = originalUpdate(table); - const originalSet = builder.set.bind(builder); - builder.set = ((values: Record) => { - if (table === heartbeatRuns && Object.keys(values).length === 1 && values.executionControlDeadlineAt === null) { - return { where: async () => cleanupFailure() }; + const transaction = vi.spyOn(db, "transaction").mockImplementation((operation, config) => + originalTransaction(async tx => { + const originalUpdate = tx.update.bind(tx); + const update = vi.spyOn(tx, "update").mockImplementation(((table: typeof heartbeatRuns) => { + const builder = originalUpdate(table); + const originalSet = builder.set.bind(builder); + builder.set = ((values: Record) => { + if (table === heartbeatRuns && Object.keys(values).length === 1 && values.executionControlDeadlineAt === null) { + return { where: async () => cleanupFailure() }; + } + return originalSet(values); + }) as typeof builder.set; + return builder; + }) as typeof tx.update); + try { + return await operation(tx); + } finally { + update.mockRestore(); } - return originalSet(values); - }) as typeof builder.set; - return builder; - }) as typeof db.update); + }, config)); const heartbeat = heartbeatService(db); try { await heartbeat.resumeQueuedRuns(); @@ -8319,7 +8328,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(adapterExecutionControls.has(runId)).toBe(false); expect((await heartbeat.getRun(runId))?.status).not.toBe("running"); } finally { - update.mockRestore(); + transaction.mockRestore(); watcher.mockRestore(); } }); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 0246d509ff3..e1151c9c899 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -106,6 +106,7 @@ describe("issueRecoveryActionService", () => { }); const fakeDb = { + async transaction(fn: (tx: unknown) => Promise): Promise { return fn(fakeDb); }, select: vi.fn(() => makeSelectQuery(selectResults.shift() ?? [])), update: vi.fn(() => ({ set: vi.fn(() => ({ diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 96d94ef248a..db22ce7b2c2 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -790,16 +790,9 @@ describe("startServer feedback export wiring", () => { } }); - it("reconciles native replacements at startup and on the execution-control interval", async () => { + it("registers native replacement recovery without an execution-control interval", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); - let executionControlTick: (() => void) | undefined; - const setIntervalSpy = vi.spyOn(globalThis, "setInterval").mockImplementation((( - callback: () => void, - interval: number, - ) => { - if (interval === EXECUTION_RECONCILIATION_INTERVAL_MS) executionControlTick = callback; - return 1 as unknown as ReturnType; - }) as typeof setInterval); + const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); try { await startServer(); await new Promise((resolve) => setImmediate(resolve)); @@ -808,11 +801,8 @@ describe("startServer feedback export wiring", () => { expect.any(Date), { verifyStoppedSession: expect.any(Function) }, ); - - expect(executionControlTick).toBeDefined(); - executionControlTick?.(); - await new Promise((resolve) => setImmediate(resolve)); - expect(reconcileSafeNativeReplacements).toHaveBeenCalledTimes(2); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.executionReplacement)).toBe(true); + expect(setIntervalSpy.mock.calls.some(([, interval]) => interval === EXECUTION_RECONCILIATION_INTERVAL_MS)).toBe(false); } finally { setIntervalSpy.mockRestore(); } diff --git a/server/src/index.ts b/server/src/index.ts index 1645b0d7bcb..a7b0d17fabe 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -39,7 +39,7 @@ import { deliverExecutionStatuses } from "./services/execution-status-delivery.j import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js"; import { reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js"; import { reconcileAbandonedExecutionControl } from "./services/execution-control-reconciliation.js"; -import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./services/execution-control-deadline.js"; +import { executionControlSchedule, registerExecutionControlWork } from "./services/execution-control-work.js"; import { connectionIntentDeliveryService } from "./services/connection-intent-delivery.js"; import { existsSync, readFileSync, rmSync } from "node:fs"; import type { Server, RequestListener } from "node:http"; @@ -1205,30 +1205,6 @@ async function startServerWithDatabaseTeardown( await Promise.allSettled([...heartbeatSchedulerInFlight]); } }; - const executionControlSweepsInFlight = new Set(); - const executionControlSweeps = [ - ["finalization", () => reconcileAbandonedExecutionControl(db)], - ["replacement", () => heartbeat ? reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }) : undefined], - ["reconciliation_delivery", () => heartbeat ? deliverReconciledExecutions(db, heartbeat.wakeup) : undefined], - ["status_delivery", () => deliverExecutionStatuses(db)], - ["automatic_disposition", () => settleUnrecoverableExecutions(db)], - ["local_ai_login_cleanup", () => localAiLoginService(db).reapExpired()], - ] as const; - const sweepExecutionControl = () => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped || isWarmStandby()) return; - // Independent durable queues must not block one another. Each queue remains - // single-flight; a later sweep observes committed transitions from its peers. - for (const [queue, work] of executionControlSweeps) { - if (executionControlSweepsInFlight.has(queue)) continue; - executionControlSweepsInFlight.add(queue); - trackHeartbeatSchedulerWork(Promise.resolve().then(async () => { await work(); }) - .catch(err => logger.error({ err, queue }, "execution control reconciliation failed")) - .finally(() => { executionControlSweepsInFlight.delete(queue); })); - } - }; - const executionControlInterval = setInterval(sweepExecutionControl, EXECUTION_RECONCILIATION_INTERVAL_MS); - executionControlInterval.unref?.(); - sweepExecutionControl(); const startHeartbeatSchedulerInterval = (callback: () => void) => { heartbeatSchedulerInterval = setInterval(() => { if (!isWarmStandby() && !isIdleTaskDrainActive()) callback(); @@ -1346,6 +1322,18 @@ async function startServerWithDatabaseTeardown( resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction), }); const deliveryWork = app.locals.deliveryWork as ReturnType; + const executionSchedule = executionControlSchedule(db); + registerExecutionControlWork(deliveryWork, [ + { queue: DELIVERY_QUEUES.executionFinalization, run: () => reconcileAbandonedExecutionControl(db), next: executionSchedule.finalization }, + { queue: DELIVERY_QUEUES.executionStatus, run: () => deliverExecutionStatuses(db), next: executionSchedule.status }, + { queue: DELIVERY_QUEUES.executionDisposition, run: () => settleUnrecoverableExecutions(db), next: executionSchedule.disposition }, + { queue: DELIVERY_QUEUES.localLoginCleanup, run: () => localAiLoginService(db).reapExpired(), next: executionSchedule.localLogin }, + ...(heartbeat ? [ + { queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement }, + { queue: DELIVERY_QUEUES.executionContinuation, run: () => deliverReconciledExecutions(db, heartbeat.wakeup), next: executionSchedule.continuation }, + ] : []), + ], () => !heartbeatSchedulerStopped); + const budgetEnforcementWorker = registerBudgetEnforcementWork(db, deliveryWork, { cancelWorkForScope: accountingHeartbeat.cancelBudgetScopeWork, }); @@ -2092,7 +2080,6 @@ async function startServerWithDatabaseTeardown( unsubscribeChatCompletions(); await deliveryWork.stop(); unregisterTerminalWorkspaceProbe?.(); - clearInterval(executionControlInterval); if (heartbeatSchedulerInterval) { clearInterval(heartbeatSchedulerInterval); heartbeatSchedulerInterval = null; diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index b1c7c62062a..26f737b39cd 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,12 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + executionFinalization: "execution-finalization", + executionReplacement: "execution-replacement", + executionContinuation: "execution-continuation", + executionStatus: "execution-status", + executionDisposition: "execution-disposition", + localLoginCleanup: "local-login-cleanup", budgetEnforcement: "budget-enforcement", accounting: "cost-accounting", agentLifecycle: "agent-lifecycle", diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts new file mode 100644 index 00000000000..fb59deafbe8 --- /dev/null +++ b/server/src/services/execution-control-notifications.ts @@ -0,0 +1,20 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js"; + +export async function notifyExecutionRunWork(transaction: object) { + for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, + DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition]) { + await notifyDeliveryWork(transaction, queue); + } +} + +export function writeExecutionControl( + db: Db | Parameters[0]>[0], + queue: DeliveryQueue | readonly DeliveryQueue[], + write: (tx: Db) => PromiseLike, +): Promise { + return db.transaction(async transaction => { + for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); + return write(transaction as unknown as Db); + }); +} diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 72266731dff..5b3a7aa3735 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { randomUUID } from "node:crypto"; import { logger } from "../middleware/logger.js"; import { and, eq, isNotNull, lte, sql } from "drizzle-orm"; @@ -181,6 +182,7 @@ export async function reconcileAbandonedExecutionControl( updatedAt: now, }) .where(eq(nativeRunFinalizations.runId, run.id)); + await notifyExecutionRunWork(tx); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/execution-control-work.ts b/server/src/services/execution-control-work.ts new file mode 100644 index 00000000000..1c7a1a6d376 --- /dev/null +++ b/server/src/services/execution-control-work.ts @@ -0,0 +1,98 @@ +import { and, asc, eq, isNotNull, sql } from "drizzle-orm"; +import { adapterAuthSessions, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { type DeliveryQueue } from "./delivery-work-notifications.js"; +import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js"; +import { obsoleteExecutionConversationHold, unresolvedExecutionDispositions } from "./execution-recovery-resolution.js"; +import { nativeReplacementPending } from "./native-runtime/native-safe-replacement.js"; + +export function registerExecutionControlWork( + coordinator: ReturnType, + duties: ReadonlyArray<{ + queue: DeliveryQueue; + run: () => Promise; + next: () => Promise; + }>, + canRun: () => boolean = () => true, +) { + const workers = duties.map(duty => { + let requested = 1; + let reconciled = 0; + let next: number | null = null; + let earliestRunAt = 0; + return coordinator.register(duty.queue, { + retryMs: EXECUTION_RECONCILIATION_INTERVAL_MS, + pausePassiveMaintenance: true, + onSettled: () => { requested += 1; }, + run: async () => { + if (!canRun()) { requested += 1; next = null; return; } + const now = Date.now(); + if (now < earliestRunAt) { + if (reconciled !== requested || (next !== null && next <= now)) next = earliestRunAt; + return; + } + const generation = ++requested; + earliestRunAt = now + EXECUTION_RECONCILIATION_INTERVAL_MS; + await duty.run(); + const deadline = await duty.next(); + next = deadline === null ? null : Math.max(deadline, earliestRunAt); + reconciled = generation; + }, + hasPending: async () => reconciled !== requested, + nextRunAt: () => next, + }); + }); + return { ready: Promise.all(workers.map(worker => worker.ready)) }; +} + +export function executionControlSchedule(db: Db) { + const retry = () => Date.now() + EXECUTION_RECONCILIATION_INTERVAL_MS; + const deadline = (dates: Array) => { + const at = dates.flatMap(date => date ? [date.getTime()] : []); + if (!at.length) return null; + const earliest = Math.min(...at); + return earliest > Date.now() ? earliest : retry(); + }; + return { + async finalization() { + const [native, legacy] = await Promise.all([ + db.select({ at: nativeRunFinalizations.controlDeadlineAt }).from(nativeRunFinalizations) + .where(isNotNull(nativeRunFinalizations.controlDeadlineAt)).orderBy(asc(nativeRunFinalizations.controlDeadlineAt)).limit(1), + db.select({ at: heartbeatRuns.executionControlDeadlineAt }).from(heartbeatRuns) + .where(isNotNull(heartbeatRuns.executionControlDeadlineAt)).orderBy(asc(heartbeatRuns.executionControlDeadlineAt)).limit(1), + ]); + return deadline([native[0]?.at, legacy[0]?.at]); + }, + async replacement() { + const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .where(nativeReplacementPending()).limit(1); + return rows.length ? retry() : null; + }, + async continuation() { + const rows = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.status, "resolved"), + sql`${issueRecoveryActions.evidence}->>'continuationDelivery' = 'pending'`, + )).limit(1); + return rows.length ? retry() : null; + }, + async status() { + const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(isNotNull(heartbeatRuns.executionStatusDeliveryId)).limit(1); + return rows.length ? retry() : null; + }, + async disposition() { + const [conversation, execution] = await Promise.all([ + db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(obsoleteExecutionConversationHold()).limit(1), + unresolvedExecutionDispositions(db, 1), + ]); + return conversation.length || execution.length ? retry() : null; + }, + async localLogin() { + const rows = await db.select({ at: adapterAuthSessions.expiresAt }).from(adapterAuthSessions).where(and( + eq(adapterAuthSessions.connectionMethod, "local_subscription"), isNotNull(adapterAuthSessions.expiresAt), + )).orderBy(asc(adapterAuthSessions.expiresAt)).limit(1); + return deadline([rows[0]?.at]); + }, + }; +} diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index 3323bceb93e..e56141565b9 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -1,3 +1,5 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; import { randomUUID } from "node:crypto"; import { canonicalExecutionUuid } from "./execution-identity.js"; @@ -201,38 +203,43 @@ export async function markExecutionReconciliation( throw conflict("The authorized chat retry owner is no longer valid."); } } - await db - .update(nativeRunFinalizations) - .set({ - failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "operator_reconciled" })}::jsonb`, - }) - .where( - and( - eq(nativeRunFinalizations.companyId, action.companyId), - eq(nativeRunFinalizations.runId, decision.runId), - ), - ); - await db - .update(issueRecoveryActions) - .set({ - evidence: { - ...action.evidence, - automaticRecovery: undefined, - executionReconciliation: { - ...decision, - actorId, - recordedAt: new Date().toISOString(), + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition); + await tx + .update(nativeRunFinalizations) + .set({ + failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "operator_reconciled" })}::jsonb`, + }) + .where( + and( + eq(nativeRunFinalizations.companyId, action.companyId), + eq(nativeRunFinalizations.runId, decision.runId), + ), + ); + await tx + .update(issueRecoveryActions) + .set({ + evidence: { + ...action.evidence, + automaticRecovery: undefined, + executionReconciliation: { + ...decision, + actorId, + recordedAt: new Date().toISOString(), + }, + continuationDelivery: options?.workspaceRepairOnly ? "not_requested" : deliveryOwner ? "delegated" : "pending", + ...(deliveryOwner ? { continuationDeliveryOwner: deliveryOwner } : {}), }, - continuationDelivery: options?.workspaceRepairOnly ? "not_requested" : deliveryOwner ? "delegated" : "pending", - ...(deliveryOwner ? { continuationDeliveryOwner: deliveryOwner } : {}), - }, - }) - .where( - and( - eq(issueRecoveryActions.companyId, action.companyId), - eq(issueRecoveryActions.id, action.id), - ), - ); + }) + .where( + and( + eq(issueRecoveryActions.companyId, action.companyId), + eq(issueRecoveryActions.id, action.id), + ), + ); + }); } export async function deliverReconciledExecutions( @@ -337,60 +344,8 @@ export async function deliverReconciledExecutions( } } -/** - * Failed execution is a system responsibility, not a user questionnaire. After - * automatic recovery is ruled out, preserve evidence and stop without replay. - * This is NOT evidence that an external action succeeded or never happened. - * The resolved record retains a dispatch hold until actual evidence clears it. - */ -export async function settleUnrecoverableExecutions( - db: Db, - now = new Date(), - options: { failpoint?: (phase: "persisted") => void } = {}, -) { - // Fold obsolete conversation holds without waking historical work on upgrade. - // Keep their evidence and record the policy change in the task's activity log. - const obsoleteConversationHold = and( - conversationRecoveryActionPredicate(), - sql`not (${issueRecoveryActions.evidence} ? 'systemRecovery')`, - or( - inArray(issueRecoveryActions.status, ["active", "escalated"]), - sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, - ), - ); - await db.transaction(async tx => { - const foldable = await tx.select().from(issueRecoveryActions).where(obsoleteConversationHold) - .limit(25).for("update", { skipLocked: true }); - for (const candidate of foldable) { - if (await getConversationOwnershipBlocker(tx as unknown as Db, candidate.companyId, candidate.sourceIssueId)) continue; - const [action] = await tx.update(issueRecoveryActions).set({ - status: "resolved", - outcome: "cancelled", - resolvedAt: now, - updatedAt: now, - nextAction: "Automatic attempts stopped. Send a new message to continue the conversation.", - resolutionNote: "Conversation continuation does not replay prior tool calls.", - wakePolicy: null, - monitorPolicy: null, - evidence: sql`case when ${issueRecoveryActions.evidence} ? 'automaticRecovery' - then jsonb_set(${issueRecoveryActions.evidence}, '{automaticRecovery,replay}', '"conversation_continuation"'::jsonb) - else ${issueRecoveryActions.evidence} end`, - }).where(and(obsoleteConversationHold, eq(issueRecoveryActions.id, candidate.id))).returning(); - if (!action) continue; - await persistActivity(tx as unknown as Db, { - companyId: action.companyId, - actorType: "system", - actorId: "execution-recovery", - action: "issue.execution_recovery_settled", - entityType: "issue", - entityId: action.sourceIssueId, - details: { recoveryActionId: action.id, outcome: "cancelled", continuation: "conversation" }, - }); - } - }); - // Filter eligibility before applying the batch limit. A queue of sessions - // awaiting replacement must not starve settled incidents behind it. - const candidates = await db +export function unresolvedExecutionDispositions(db: Db, limit: number) { + return db .select({ action: issueRecoveryActions }) .from(issueRecoveryActions) .innerJoin( @@ -437,7 +392,67 @@ export async function settleUnrecoverableExecutions( or coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') <> '')`, ), ) - .limit(25); + .limit(limit); +} + +export function obsoleteExecutionConversationHold() { + return and( + conversationRecoveryActionPredicate(), + sql`not (${issueRecoveryActions.evidence} ? 'systemRecovery')`, + or( + inArray(issueRecoveryActions.status, ["active", "escalated"]), + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, + ), + ); +} + +/** + * Failed execution is a system responsibility, not a user questionnaire. After + * automatic recovery is ruled out, preserve evidence and stop without replay. + * This is NOT evidence that an external action succeeded or never happened. + * The resolved record retains a dispatch hold until actual evidence clears it. + */ +export async function settleUnrecoverableExecutions( + db: Db, + now = new Date(), + options: { failpoint?: (phase: "persisted") => void } = {}, +) { + // Fold obsolete conversation holds without waking historical work on upgrade. + // Keep their evidence and record the policy change in the task's activity log. + const obsoleteConversationHold = obsoleteExecutionConversationHold(); + await db.transaction(async tx => { + const foldable = await tx.select().from(issueRecoveryActions).where(obsoleteConversationHold) + .limit(25).for("update", { skipLocked: true }); + for (const candidate of foldable) { + if (await getConversationOwnershipBlocker(tx as unknown as Db, candidate.companyId, candidate.sourceIssueId)) continue; + const [action] = await tx.update(issueRecoveryActions).set({ + status: "resolved", + outcome: "cancelled", + resolvedAt: now, + updatedAt: now, + nextAction: "Automatic attempts stopped. Send a new message to continue the conversation.", + resolutionNote: "Conversation continuation does not replay prior tool calls.", + wakePolicy: null, + monitorPolicy: null, + evidence: sql`case when ${issueRecoveryActions.evidence} ? 'automaticRecovery' + then jsonb_set(${issueRecoveryActions.evidence}, '{automaticRecovery,replay}', '"conversation_continuation"'::jsonb) + else ${issueRecoveryActions.evidence} end`, + }).where(and(obsoleteConversationHold, eq(issueRecoveryActions.id, candidate.id))).returning(); + if (!action) continue; + await persistActivity(tx as unknown as Db, { + companyId: action.companyId, + actorType: "system", + actorId: "execution-recovery", + action: "issue.execution_recovery_settled", + entityType: "issue", + entityId: action.sourceIssueId, + details: { recoveryActionId: action.id, outcome: "cancelled", continuation: "conversation" }, + }); + } + }); + // Filter eligibility before applying the batch limit. A queue of sessions + // awaiting replacement must not starve settled incidents behind it. + const candidates = await unresolvedExecutionDispositions(db, 25); for (const { action: candidate } of candidates) { const runId = candidate.evidence.runId; if (typeof runId !== "string") continue; @@ -586,6 +601,7 @@ export async function settleUnrecoverableExecutions( replay: "not_authorized", }, }); + await notifyExecutionRunWork(tx); await tx .update(heartbeatRuns) .set({ executionStatusDeliveryId: randomUUID() }) diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index c449167dd00..a33d421a24f 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { canSelectFreshWorkspace, type ContinuationReadiness } from "./fresh-workspace-continuation.js"; import { readTaskRetryIntent } from "./task-retry.js"; import { readSystemRecovery } from "./system-recovery-state.js"; @@ -356,16 +357,19 @@ export async function admitExplicitNativeContinuation(input: { ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}), ...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId, previousRunId: previous.id, recordedAt: new Date().toISOString() }; - for (const runId of cancelledStartupIds) { - await db.update(nativeRunFinalizations).set({ - phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null, - controlDeadlineAt: null, updatedAt: new Date(), - }).where(and(eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId))); - await db.update(heartbeatRuns).set({ - ...(nativeSources.some(run => run.id === runId) ? { nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date() } : {}), - executionControlDeadlineAt: null, - }).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); - } + if (cancelledStartupIds.size) await db.transaction(async tx => { + await notifyExecutionRunWork(tx); + for (const runId of cancelledStartupIds) { + await tx.update(nativeRunFinalizations).set({ + phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null, + controlDeadlineAt: null, updatedAt: new Date(), + }).where(and(eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId))); + await tx.update(heartbeatRuns).set({ + ...(nativeSources.some(run => run.id === runId) ? { nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date() } : {}), + executionControlDeadlineAt: null, + }).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); + } + }); if (nativeSources.length) await db.update(nativeRunFinalizations).set({ failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "explicit_user_continuation" })}::jsonb`, updatedAt: new Date(), diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index b571364d22d..537510275da 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,5 @@ +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { writeExecutionControl } from "./execution-control-notifications.js"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; import { promptInjectionPreventionService } from "./prompt-injection-prevention.js"; import { createHeartbeatAgentStatus } from "./heartbeat/agent-status.js"; @@ -3657,10 +3659,10 @@ export function heartbeatService( let latestRun = await getRun(run.id).catch(() => null); try { if (latestRun && isHeartbeatRunTerminalStatus(latestRun.status)) { - await db + await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx .update(heartbeatRuns) .set({ executionControlDeadlineAt: null }) - .where(eq(heartbeatRuns.id, run.id)); + .where(eq(heartbeatRuns.id, run.id))); } nativeOwnershipHeld = nativeOwnershipHeld || diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index fbebed7f69f..0ee088a4d51 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import os from "node:os"; import { DEFERRED_WAKE_CONTEXT_KEY, @@ -1011,12 +1012,12 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl } async function writeRunStatus(status: string, companyId: string | undefined, write: (tx: Db) => Promise) { - if (!isHeartbeatRunTerminalStatus(status)) return write(db); // Register on the outer transaction before the terminal write. An unknown // commit cannot strand newly-actionable accounting after the live event. return db.transaction(async transaction => { const tx = transaction as unknown as Db; - await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId); + if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId); + await notifyExecutionRunWork(tx); return write(tx); }); } diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index bba22d1442f..cc304a6afba 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -1,3 +1,5 @@ +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { writeExecutionControl } from "../execution-control-notifications.js"; import { AsyncLocalStorage } from "node:async_hooks"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { parseNativeSessionGoalControl } from "./scheduling.js"; @@ -933,7 +935,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe })); }); } await ownership?.assertHeld(); - await db + await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx .update(heartbeatRuns) .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) }) .where( @@ -941,7 +943,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running"), ), - ); + )); const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded"; await recordWorkspaceFinalize(workspaceFinalizeStatus); if (adapterResult.nativeFinalization) { @@ -1009,7 +1011,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe && !(nativeExecution && "access" in nativeExecution.workspace && nativeExecution.workspace.access === "none")) { input.effects.onProviderResourceDisposition("stop_and_retain"); } - await db + await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx .update(heartbeatRuns) .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) }) .where( @@ -1017,7 +1019,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running"), ), - ); + )); if ( issueRef && context.resumeSessionGoalHeartbeat === true && diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 5351722b198..04e2c19a6d1 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -1,3 +1,5 @@ +import { writeExecutionControl } from "./execution-control-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { and, desc, eq, inArray, isNull, isNotNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { heartbeatRuns, issueRecoveryActions, workspaceOperations } from "@paperclipai/db"; @@ -318,6 +320,8 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ): Promise { try { const created = await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation); const [superseded] = await tx .update(issueRecoveryActions) .set({ @@ -381,10 +385,10 @@ export function issueRecoveryActionService(db: DbOrTransaction) { if (existing.cause === "native_workspace_finalization_owner_unverified" && input.cause === existing.cause && input.fingerprint === existing.fingerprint && !existing.evidence?.workspaceOwnerStop && existing.nextAction !== input.nextAction) { - const [updated] = await db.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now }) + const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now }) .where(and(eq(issueRecoveryActions.id, existing.id), inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]), - sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning(); + sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning()); if (updated) return toReadModel(updated); } return existing; @@ -404,7 +408,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { existing.attemptCount, Math.min(nextAttemptCount, effectiveMaxAttempts), ); - const [exhausted] = await db + const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx .update(issueRecoveryActions) .set({ status: "escalated", @@ -452,13 +456,13 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ]), ), ) - .returning(); + .returning()); if (!exhausted) { return retryUpsertSourceScoped(input, retryCount); } return toReadModel(exhausted); } - const [updated] = await db + const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx .update(issueRecoveryActions) .set({ recoveryIssueId: input.preserveExistingOwner @@ -517,7 +521,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]), ), ) - .returning(); + .returning()); if (!updated) { return retryUpsertSourceScoped(input, retryCount); } @@ -525,10 +529,10 @@ export function issueRecoveryActionService(db: DbOrTransaction) { } try { - const [created] = await db + const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx .insert(issueRecoveryActions) .values(buildInsertValues(input, ownerType, now)) - .returning(); + .returning()); return toReadModel(created!); } catch (error) { if (!isUniqueRecoveryActionConflict(error)) throw error; @@ -565,7 +569,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { predicates.push(eq(issueRecoveryActions.fingerprint, input.fingerprint)); } - const [updated] = await dbOrTx + const [updated] = await writeExecutionControl(dbOrTx, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx .update(issueRecoveryActions) .set({ status: input.status, @@ -575,7 +579,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { updatedAt: now, }) .where(and(...predicates)) - .returning(); + .returning()); return updated ? toReadModel(updated) : null; } diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 54fe6184b35..f1b16bd0e49 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { hasRequiredWorkspaceRecovery, preserveWorkspaceRestoreRecoveryMetadata, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js"; import { hasUnrestoredRemoteWorkspace, preserveLegacyWorkspaceRestoreSources, type LegacyWorkspaceRestoreSource } from "./legacy-workspace-restore-recovery.js"; import { isPreDispatchReviewWaitVerified } from "./pre-dispatch-review-wait.js"; @@ -108,6 +109,7 @@ export async function terminalizeLegacyExecution(input: { resultJson: preserveWorkspaceRestoreRecoveryMetadata(current.resultJson, input.recordRestoreFailureOnly ? { ...current.resultJson, ...patch.resultJson } : patch.resultJson), }; + await notifyExecutionRunWork(tx); let [updated] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/local-ai-login.ts b/server/src/services/local-ai-login.ts index 71482abc4cb..2108a916baf 100644 --- a/server/src/services/local-ai-login.ts +++ b/server/src/services/local-ai-login.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { realpathSync } from "node:fs"; import { mkdir, rm, writeFile } from "node:fs/promises"; @@ -120,6 +121,7 @@ export function localAiLoginService(db: Db) { const expiresAt = new Date(Date.now() + ATTEMPT_DURATION_MS); await prepareHome(id, intent.provider); try { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.localLoginCleanup); await tx.insert(adapterAuthSessions).values({ id, publicSessionId: id, companyId, environmentId: environment.id, adapterType, diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index 76abcdefd31..c8f31024920 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js"; @@ -55,6 +56,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea let retained = !retainFiles || alreadyRetained ? run : await recordRequiredWorkspaceRestoreFailure(tx as unknown as Db, run, { workspaceRestoreFailure: "restore_failed" }, source); if (!retained || retainFiles && !hasRequiredWorkspaceRecovery(retained.resultJson)) return; if (options.allowPreterminal && coordinator.phase === "observed") { + await notifyExecutionRunWork(tx); const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null, recoveryState: "blocked", failureCode: coordinator.failureCode ?? "native_continuation_requires_reconciliation", updatedAt: new Date(), diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index fb05f82252a..2e3229fe011 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; import { activeNativeExecutionRunIds, NATIVE_PREPARATION_FENCING_VERSION } from "./native-active-runs.js"; import { @@ -855,6 +856,7 @@ export async function claimNativeRestartRecoveries(input: { recycledProviderPids: providerProcesses.recycledPids, stderrTail: row.run.stderrExcerpt, }); + await notifyExecutionRunWork(tx); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index cf17721d832..61f97aefb8b 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.js"; import { assertNativeCaseHandoffResult } from "./native-case-handoff.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; @@ -420,6 +421,7 @@ async function recordRetryableFailure(input: { : "native_finalization_retry_exhausted" : input.failureCode; if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); + await notifyExecutionRunWork(tx); await tx .update(nativeRunFinalizations) .set({ @@ -705,6 +707,7 @@ async function projectCommittedRun(input: { const [updatedRun] = await input.db.transaction(async transaction => { const tx = transaction as unknown as Db; await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); + await notifyExecutionRunWork(tx); return tx .update(heartbeatRuns) .set({ @@ -1617,13 +1620,12 @@ export async function finalizeNativeRun(input: { }) .where(eq(heartbeatRuns.id, run.id)) .returning(); - const [updatedRun] = input.projectRunStatus - ? await input.db.transaction(async transaction => { - const tx = transaction as unknown as Db; - await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); - return projectRun(tx); - }) - : await projectRun(input.db); + const [updatedRun] = await input.db.transaction(async transaction => { + const tx = transaction as unknown as Db; + if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); + await notifyExecutionRunWork(tx); + return projectRun(tx); + }); if ( input.projectRunStatus && !alreadyEmittedByCommittedDecision && diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index 929cc58c510..03109f56def 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { logger } from "../../middleware/logger.js"; import { createHash, randomUUID } from "node:crypto"; import { stat } from "node:fs/promises"; @@ -34,6 +35,20 @@ const processAlive = (pid: number | null) => { } }; +export function nativeReplacementPending() { + return and( + eq(heartbeatRuns.status, "failed"), + eq(nativeRunFinalizations.phase, "terminal_failure"), + inArray( + nativeRunFinalizations.failureCode, + ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"], + ), + isNull(nativeRunFinalizations.resultId), + sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`, + sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`, + ); +} + /** Retryable outbox: terminal_failure plus a new-format failure code is the durable candidate. */ export async function reconcileSafeNativeReplacements( db: Db, @@ -55,17 +70,7 @@ export async function reconcileSafeNativeReplacements( eq(nativeRunFinalizations.runId, heartbeatRuns.id), ) .where( - and( - eq(heartbeatRuns.status, "failed"), - eq(nativeRunFinalizations.phase, "terminal_failure"), - inArray( - nativeRunFinalizations.failureCode, - ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"], - ), - isNull(nativeRunFinalizations.resultId), - sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`, - sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`, - ), + nativeReplacementPending(), ) .limit(25); let scheduled = 0; @@ -260,6 +265,7 @@ export async function reconcileSafeNativeReplacements( current.failureDetail?.replacementDenied ) return; + await notifyExecutionRunWork(tx); await tx .update(nativeRunFinalizations) .set({ @@ -401,6 +407,7 @@ export async function reconcileSafeNativeReplacements( idempotencyKey: `native-safe-replacement:${run.id}`, }) .returning(); + await notifyExecutionRunWork(tx); await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: run.companyId, diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 67423e01554..682186ee2f3 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -9460,13 +9460,24 @@ describe("native session bounded recovery", () => { ); const transact = db.transaction.bind(db); const recoveryWriteAttempt = vi.fn(); - db.transaction = (async (operation) => { - if (state.execute.mock.calls.length > 0) { - recoveryWriteAttempt(); - if (ancillaryFailure === "recovery_write") throw secondaryFailure; - } - return transact(operation); - }) as typeof db.transaction; + db.transaction = (async (operation) => transact(async tx => operation(new Proxy(tx, { + get(target, key) { + if (key !== "update") return Reflect.get(target, key); + return (table: unknown) => { + const query = Reflect.apply(target.update, target, [table]); + return new Proxy(query, { get(update, property) { + if (property !== "set") return Reflect.get(update, property); + return (values: Record) => { + if (table === nativeRunFinalizations && values.phase === "terminal_failure") { + recoveryWriteAttempt(); + if (ancillaryFailure === "recovery_write") throw secondaryFailure; + } + return update.set(values); + }; + } }); + }; + }, + })))) as typeof db.transaction; const updateIssue = vi.fn(async () => null); const service = vi .spyOn(issueServiceModule, "issueService") diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 73ae25afb56..60d6e9a1d4a 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,6 @@ +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { writeExecutionControl } from "../execution-control-notifications.js"; +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { createRunUsageRecorder } from "../usage-receipts.js"; import { applyObservabilityEnvironment, type ProviderObservabilityLaunch } from "@paperclipai/adapter-utils/agent-observability"; import { museRunnerBroker } from "../muse-runner-broker.js"; @@ -9226,7 +9229,7 @@ async function executePaperclipNativeSessionWithinScope( }; const stoppedLeaseRenewal = leaseRenewal.stop().catch(() => undefined); try { - await input.db + await writeExecutionControl(input.db, DELIVERY_QUEUES.executionFinalization, tx => tx .update(nativeRunFinalizations) .set({ controlDeadlineAt: new Date( @@ -9238,7 +9241,7 @@ async function executePaperclipNativeSessionWithinScope( eq(nativeRunFinalizations.runId, input.execution.binding.runId), eq(nativeRunFinalizations.leaseOwner, leaseOwner), ), - ); + )); const failedAtMs = Date.now(); const executionFailureMessage = redactIdentityText( @@ -9280,7 +9283,7 @@ async function executePaperclipNativeSessionWithinScope( typeof stopIntent.intentAuditId === "string" && stopIntent.intentAuditId.length > 0 && ["pending", "acknowledged"].includes(String(stopIntent.dispatchState)); if (stoppedRun && (pendingStartupStop || hasAcknowledgedNativeStopIntent(stoppedRun) || hasAcknowledgedNativeReassignmentStopIntent(stoppedRun))) { - const [settled] = await input.db.update(nativeRunFinalizations).set({ + const [settled] = await writeExecutionControl(input.db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: null, updatedAt: new Date(), @@ -9290,7 +9293,7 @@ async function executePaperclipNativeSessionWithinScope( eq(nativeRunFinalizations.leaseOwner, leaseOwner), eq(nativeRunFinalizations.attempt, attempt), isNull(nativeRunFinalizations.resultId), - )).returning({ runId: nativeRunFinalizations.runId }); + )).returning({ runId: nativeRunFinalizations.runId })); if (settled) { await stoppedLeaseRenewal; if (warmSessionId !== null && lifecyclePolicy.mode === "warm") { @@ -9377,6 +9380,7 @@ async function executePaperclipNativeSessionWithinScope( ), ) .for("update"); + await notifyExecutionRunWork(tx); const updated = await tx .update(nativeRunFinalizations) .set({ diff --git a/tests/e2e/chat-adapters-ui-providers.spec.ts b/tests/e2e/chat-adapters-ui-providers.spec.ts index 399a966d1bd..bdde190f2d7 100644 --- a/tests/e2e/chat-adapters-ui-providers.spec.ts +++ b/tests/e2e/chat-adapters-ui-providers.spec.ts @@ -180,6 +180,7 @@ test.describe.serial("native chat adapter UI", () => { await expect(page.getByLabel("App configuration access token", { exact: true })).toHaveCount(0); await expect(page.getByLabel("Signing Secret", { exact: true })).toHaveCount(0); await page.reload(); + await expect(page.getByRole("heading", { name: "Send a message to your agent", exact: true })).toBeVisible(); expect(mock.slackCreations).toBe(1); expect(mock.slackInstallations).toBe(1); mock.setWebhookVerified(); await expect(page.getByRole("heading", { name: "Success", exact: true })).toBeVisible();