From d6c7c175892d91ccf88558f370fcfadac5c68318 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 21:02:26 -0500 Subject: [PATCH] Close native admission and stranded status producer edges Co-Authored-By: Paperclip --- doc/plans/2026-10-30-live-recovery-sleep.md | 19 ++++- .../src/__tests__/live-recovery-work.test.ts | 84 +++++++++++++++++-- .../active-run-watchdog/adapters/postgres.ts | 3 +- .../modules/run-dispatch/adapters/postgres.ts | 4 + .../wake-queue/adapters/postgres.test.ts | 4 +- .../modules/wake-queue/adapters/postgres.ts | 10 ++- .../adapters/queued-comment-postgres.ts | 8 +- .../modules/wake-queue/application/ports.ts | 3 +- .../wake-queue/application/use-cases.ts | 9 +- .../chat-channels/failed-run-retry.ts | 2 + .../chat-channels/slack/session-stop.ts | 2 + .../deferred-execution-notifications.ts | 1 + .../execution-control-notifications.ts | 4 +- server/src/services/heartbeat/queue.ts | 9 +- server/src/services/heartbeat/recovery.ts | 2 + .../src/services/heartbeat/run-lifecycle.ts | 4 +- server/src/services/issue-tree-control.ts | 6 ++ server/src/services/issues.ts | 12 ++- .../native-finalization-reconciler.ts | 10 ++- server/src/services/run-cost-accounting.ts | 3 + .../services/run-dispatch-notifications.ts | 1 + server/src/services/task-retry.ts | 8 ++ 22 files changed, 183 insertions(+), 25 deletions(-) diff --git a/doc/plans/2026-10-30-live-recovery-sleep.md b/doc/plans/2026-10-30-live-recovery-sleep.md index 10dfec0ff76..a7c075ed292 100644 --- a/doc/plans/2026-10-30-live-recovery-sleep.md +++ b/doc/plans/2026-10-30-live-recovery-sleep.md @@ -4,13 +4,13 @@ This is the final recovery slice after run dispatch, issue state recovery, and d | Duty | Durable truth and pending predicate | Producers | Deadline and exit | | --- | --- | --- | --- | -| Resultless native session | Native running/failed run; finalization `retryable_failure`; no result | Execution/finalization writes, checkpoint writes, restart claim writes, process identity events | Maximum saved retry/claim lease deadline. Fresh claim owner validates provider/session state. Accepted result, terminal failure, or explicit ownership gate removes the entry. | +| Resultless native session | Native running/failed run; finalization `retryable_failure`; no result | Execution/finalization writes, checkpoint writes, restart claim writes, process identity events, and existing run-dispatch/deferred-eligibility boundaries for agent, task, review, dependency and pause-hold changes | Maximum saved retry/claim lease deadline. Fresh claim owner validates provider/session state. Accepted result, terminal failure, or explicit ownership gate removes the entry. | | Native question cancellation | Cancellable run with committed `nativeQuestionCancellation` marker | Authorized task-completion interaction transaction | Immediate. Existing cancellation owner consumes exact company/run marker. Terminal status removes debt. | | Active execution | Running run | Run lifecycle, controller renewal/revocation, native authority and process evidence writes | Fresh native/legacy controller expiry, saved five-minute preparation grace, or bounded physical-process probe. Local execution promises retain execution ownership. Explicit evidence/manual gates park until their owning producer changes truth. | | Settled native process identity | Terminal native failure/cancellation with persisted process identity | Terminal execution writes, server process identity/stop receipts, environment settlement | Immediate inspection. Only matched local host/PID/group/birth or confirmed remote stop evidence clears identity. A live matched process or local execution owner keeps a bounded probe. Missing authority waits for committed evidence. Startup reconstructs rows independently of the old capped process Map. Successful release removes any obsolete compatibility Map entry. | | Orphan active environment lease | Active lease with NULL run reference or terminal same-company run | Environment owning transactions and indexed physical-resource peer notifications | Saved lease updatedAt plus five minutes, re-read before due work. Existing export, shutdown, and shared-resource fences remain. Exact-lease dispatch does not mutate timestamps for broad-page fairness. Successful CAS hands work to the existing pending-cleanup owner. | | Legacy shutdown settlement | Exact lease with pending shutdown marker | Begin, renew, finish, expiry settlement owning transactions | Exact saved marker deadline. Existing owner revalidates boot/controller, source, competing allocations and stop/copyback authority. Expiry records preserved workspace recovery; it never infers file export from process loss. | -| Stranded agent status | Ready agent with status running | Agent configuration/lifecycle/instruction settlement and run/resource transitions | Saved updatedAt plus five minutes, then actual surviving controller expiry when present. Fresh settlement rechecks live execution, native controller, working copy, and lease truth. Settled status removes debt. | +| Stranded agent status | Ready agent with status running | Agent configuration/lifecycle/instruction settlement, run/resource transitions, accounting acknowledgement, canonical wake status writes, and direct existing-receipt terminal writers | Saved updatedAt plus five minutes, then actual surviving controller expiry when present. Fresh settlement rechecks live execution, native controller, working copy, and lease truth. Settled status removes debt. | | Active output silence | Running run and coalesced last output/process start/run start/created timestamp | Actual output progress transaction, terminal source mutation, same-run terminal activity commit, watchdog decisions, execution state | Five-minute threshold, exact saved snooze/continue expiry, or physical termination probe. Fresh check re-reads output and decision. Existing watchdog alone owns completion evidence, process cleanup and fold. Suppressed/manual/blocked history does not poll. | All reconstruction queries use company/run/lease/agent scopes and keyset pages. Run and resource topics discard irrelevant fields before coalescing. The generic scheduler's pending/deadline reads use memory. Ordinary writes register intent before mutation in the same owning transaction. Nested commit, outer rollback and uncertain transaction reconciliation retain the shared database-work semantics. @@ -23,4 +23,17 @@ The real PostgreSQL matrix uses the actual heartbeat service and all recovery, d The legacy/controller process retry executes on real JavaScript and PostgreSQL clocks. Other deadline cases install their virtual clock before scheduler registration. Installing fake timers after a real scheduler timer has armed is not a valid deadline test. -The initial matrix passed12/12 in97.89s. Startup registration, scheduler-disabled and unrelated-owner failure contracts passed111/111. Final expanded receipts and current-head CI/review results are recorded in the PR. +The expanded production matrix passed23/23 in90.02s. Earlier final17 plus extracted recovery PostgreSQL contracts passed64/64 in102.08s. Existing native identity and watchdog authority suites passed44/44. Startup registration, scheduler-disabled and unrelated-owner failure contracts passed111/111. Current-head CI/review results are recorded in the PR. + + +## Final producer closure + +Native claim rejection has two meanings. A current task, assignment, dependency, pause hold, or agent gate is a deliberate wait. Scoped recovery parks that wait until the existing dispatch/eligibility notification boundary commits. A caught database failure keeps durable retry backoff and reaches the worker error callback. Explicit legacy aggregate callers retain their prior deferral behavior. + +Agent resume does not override a task that issue recovery deliberately marked Blocked while the agent was paused. The production test requires a new native-owner revalidation pass, preserves that task gate, and proves quiet again. The separate pause-hold release test proves actual native claim settlement. + +A run terminal commit activates resource recovery even if cleanup stops before any lease update. This fanout is in execution-control notification boundaries. Liveness-only renewal stays run-only. The watchdog's own terminal fold also publishes its exact run/resource scope. + +The stranded-agent predicate depends on final accounting and wake receipts as well as runs. `accountRunCost` publishes before acknowledgement, including stale acknowledgement cleanup, with no budget policy required. The canonical wake status writer publishes the known agent before mutation. Direct receipt writers in wake-queue cancellation/authorization, task Retry retirement/chat settlement, late chat authorization failure, Slack stop, tree cancellation and assignment cancellation do the same. Run-coupled queue/retry cancellation paths use execution-control notifications. Newly inserted already-terminal receipts do not remove a prior blocker and need no such signal. Assignment cancellation selects only actual unfinished receipts with the canonical idle proof; historical cancellation write semantics remain unchanged. + +Each producer regression starts with a worker already parked and no unrelated budget/resource transition. The final receipt controls require durable agent status settlement, no scheduler deadline/idle hold, and twenty minutes with zero SQL. The native gate and terminal-resource controls captured the original failure before the missing topic fanout was added. diff --git a/server/src/__tests__/live-recovery-work.test.ts b/server/src/__tests__/live-recovery-work.test.ts index 82cc2f87251..0d84fc546d8 100644 --- a/server/src/__tests__/live-recovery-work.test.ts +++ b/server/src/__tests__/live-recovery-work.test.ts @@ -2,12 +2,12 @@ import { randomUUID } from "node:crypto"; import os from "node:os"; import { spawn } from "node:child_process"; import { once } from "node:events"; -import { setImmediate as immediate } from "node:timers/promises"; +import { setImmediate as immediate, setTimeout as realDelay } from "node:timers/promises"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { agents, authUsers, companies, companyMemberships, createDb, closeLivenessDb, closeCheckpointDb, heartbeatRuns, heartbeatRunEvents, nativeRunFinalizations, environmentLeases, issueThreadInteractions, - issueRecoveryActions, issues, activityLog } from "@paperclipai/db"; + issueRecoveryActions, issues, activityLog, agentWakeupRequests } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -23,6 +23,11 @@ import { createHeartbeatRunOutput } from "../services/heartbeat/run-output.js"; import { recoveryService } from "../services/recovery/service.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { persistActivity } from "../services/activity-log.js"; +import { accountRunCost } from "../services/run-cost-accounting.js"; +import { createHeartbeatLifecycle } from "../services/heartbeat/run-lifecycle.js"; +import { createAgentLifecycle, configureAgentLifecycle } from "../services/agent-lifecycle.js"; +import { issueTreeControlService } from "../services/issue-tree-control.js"; +import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js"; import { issueService } from "../services/issues.js"; vi.setConfig({ testTimeout: 90_000 }); @@ -51,7 +56,7 @@ describe("residual recovery production matrix", () => { coordinator.register = (queue, task) => { deadlines.set(queue, () => task.nextRunAt?.() ?? null); return register(queue, { ...task, onSettled: hints => { timeline.push(["settled", queue, Date.now(), [...hints ?? []]]); task.onSettled?.(hints); }, run: signal => { timeline.push(["pass", queue, Date.now()]); const pass = Promise.resolve(task.run(signal)); passes.add(pass); - void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass; + void pass.then(() => { timeline.push(["finished", queue, Date.now()]); passes.delete(pass); }, () => passes.delete(pass)); return pass; } }); }; }); afterEach(async () => { @@ -101,7 +106,7 @@ describe("residual recovery production matrix", () => { const quietAt = Date.now(); if (vi.isFakeTimers()) await advance(31_000); else await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 35_000 }); - console.info("quiet-drain", JSON.stringify({ quietAt, now: Date.now(), active: idleWorkSnapshot().active, next: [...deadlines].map(([q, at]) => [q, at()]), timeline })); + console.info("quiet-drain", JSON.stringify({ quietAt, now: Date.now(), active: idleWorkSnapshot().active, next: [...deadlines].map(([q, at]) => [q, at()]), passes: Object.fromEntries([...deadlines.keys()].map(queue => [queue, timeline.filter(event => Array.isArray(event) && event[0] === "pass" && event[1] === queue).length])), settled: timeline.filter(event => Array.isArray(event) && event[0] === "settled") })); await vi.waitFor(() => { expect(coordinator.nextWakeAt(), JSON.stringify([...deadlines].map(([name, at]) => [name, at()]))).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); const queries: string[] = [], prior = db.$client.options.debug; db.$client.options.debug = (_connection, query) => { queries.push(query); }; @@ -303,6 +308,8 @@ describe("residual recovery production matrix", () => { controllerBootId: randomUUID(), controllerLeaseExpiresAt: expiry }).returning(); await db.insert(activityLog).values({ companyId: f.companyId, actorType: "agent", actorId: f.agent.id, agentId: f.agent.id, runId: target.id, entityType: "issue", entityId: source.id, action: "issue.updated", details: { status: "done" } }); + const broad = await recoveryService(db, { enqueueWakeup: heartbeat.wakeup }).scanSilentActiveRuns({ companyId: f.companyId }); + expect(broad.scanned).toBe(100); expect((await getRun(target.id)).status).toBe("running"); await start(); await advance(31_000); await vi.waitFor(async () => expect((await getRun(target.id)).status).toBe("succeeded")); expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "running"))).toHaveLength(suppressed.length); @@ -387,9 +394,10 @@ describe("residual recovery production matrix", () => { }); it("parks suppressed worktree recovery without probes and reconstructs it on the committed feature change", async () => { const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); - const settings = instanceSettingsService(db); + const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "live-recovery-worktree" }; + const settings = instanceSettingsService(db, { runtimeEnv }); await settings.updateExperimental({ enableWorktreeRunExecution: false }); - heartbeat = heartbeatService(db, { runtimeEnv: { PAPERCLIP_IN_WORKTREE: "true" } }); + heartbeat = heartbeatService(db, { runtimeEnv }); const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, status: "active", provider: "local", updatedAt: old }).returning(); await start(); expect((await getLease(lease.id)).status).toBe("active"); await quiet(); try { @@ -399,6 +407,70 @@ describe("residual recovery production matrix", () => { } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } await quiet(); }); + it.each(["agent", "hold"] as const)("revalidates parked native session work on ordinary %s eligibility", async gate => { + const f = await fixture(), tree = issueTreeControlService(db); + const lifecycleWorker = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete" }); + extraStops.push(() => lifecycleWorker.stop()); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id)); + const actor = { actorType: "user" as const, actorId: f.actorId, userId: f.actorId }; + const hold = gate === "hold" ? await tree.createHold(f.companyId, f.issue.id, { mode: "pause", actor }) : null; + if (gate === "agent") { await createAgentLifecycle(db).pauseAgent(f.agent.id); await lifecycleWorker.process(f.agent.id); } + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "failed", runtimeMode: "native", nativeIssueId: f.issue.id, contextSnapshot: { issueId: f.issue.id }, + finishedAt: new Date(), errorCode: "saved_native_failure" }).returning(); + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, f.issue.id)); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, + phase: "retryable_failure", nextAttemptAt: new Date(Date.now() - 1_000) }); + await start(); await quiet(); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0].phase).toBe("retryable_failure"); + const before = timeline.filter(event => Array.isArray(event) && event[0] === "finished" && event[1] === "native-process-sessions").length; + if (gate === "agent") { await createAgentLifecycle(db).resumeAgent(f.agent.id); await lifecycleWorker.process(f.agent.id); } + else await tree.releaseHold(f.companyId, f.issue.id, hold!.hold.id, { actor, reason: "Resume native recovery" }); + await advance(1_000); + const deadline = performance.now() + 5_000; + while (timeline.filter(event => Array.isArray(event) && event[0] === "finished" && event[1] === "native-process-sessions").length <= before && performance.now() < deadline) + await realDelay(5); + expect(timeline.filter(event => Array.isArray(event) && event[0] === "finished" && event[1] === "native-process-sessions").length).toBeGreaterThan(before); + const expectedPhase = gate === "agent" ? "retryable_failure" : "terminal_failure"; + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0].phase).toBe(expectedPhase); + if (gate === "agent") expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].status).toBe("blocked"); + expect((await getRun(run.id)).errorCode).toBe("saved_native_failure"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + await quiet(); + }); + it("discovers an active lease from the ordinary terminal run commit before any lease write", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "running", runtimeMode: "legacy", startedAt: new Date(), updatedAt: new Date(), + controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: new Date(Date.now() + 60 * 60_000) }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id, + status: "active", provider: "local", updatedAt: old }).returning(); + await start(); expect(deadlines.get("orphan-active-leases")!()).toBeNull(); + const ended = await terminalizeLegacyExecution({ db, run, status: "failed", patch: { finishedAt: new Date(), errorCode: "owner_process_failure" } }); + expect(ended?.status).toBe("failed"); expect((await getLease(lease.id)).status).toBe("active"); + await advance(1_000); + await vi.waitFor(async () => expect((await getLease(lease.id)).status).toBe("pending_cleanup")); + await quiet(); + }); + it.each(["accounting", "wake", "assignment_cancel"] as const)("settles parked agent status when its last %s receipt commits", async receipt => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + await db.update(agents).set({ status: "running", updatedAt: old }).where(eq(agents.id, f.agent.id)); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "succeeded", finishedAt: old, costAccountingPending: receipt === "accounting", + usageJson: { accountingProviderWorkStarted: false } }).returning(); + const [wake] = receipt !== "accounting" ? await db.insert(agentWakeupRequests).values({ companyId: f.companyId, + agentId: f.agent.id, runId: run.id, status: "claimed", source: "automation", reason: "fixture-last-receipt", + requestedByActorType: receipt === "assignment_cancel" ? "user" : "system", requestedByActorId: f.actorId, payload: { issueId: f.issue.id } }).returning() : []; + await start(); await quiet(); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("running"); + if (receipt === "accounting") expect(await accountRunCost(db, run.id)).toBe(true); + else if (receipt === "assignment_cancel") await issueService(db).update(f.issue.id, { assigneeAgentId: null, assigneeUserId: f.actorId }); + else await createHeartbeatLifecycle(db, {} as Parameters[1]) + .setWakeupStatus(wake.id, "completed", { finishedAt: new Date() }); + await advance(1_000); + await vi.waitFor(async () => expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("idle")); + await quiet(); + }); it("retains first producer debt during drain and warm standby, then resumes without duplicate cleanup", async () => { const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); draining = true; await start(); diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 151c57e7a16..2d892755d39 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -1,4 +1,4 @@ -import { notifyActiveRunSilence, notifyLiveRunRecovery } from "../../../services/live-run-recovery-notifications.js"; +import { notifyActiveRunSilence, notifyLiveRunRecovery, notifyLiveResourceRecovery } from "../../../services/live-run-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; @@ -349,6 +349,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch const transactionResult = await db.transaction(async (tx) => { await notifyLiveRunRecovery(tx, input.run); + await notifyLiveResourceRecovery(tx, input.run); await notifyInstructionCollection(tx, { companyId, runId: input.run.id }); await notifyExternalAgentSettlement(tx, { companyId, runId: input.run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id }); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 5acbfcf6884..cb59e363c1a 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "../../../services/execution-control-notifications.js"; import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js"; import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restore-recovery-state.js"; import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js"; @@ -687,6 +688,7 @@ export function createPostgresRunDispatchAdapter( | { applied: true; run: HeartbeatRun; postCommitEffects: PostCommitEffect[] } | { applied: false } > { + await notifyExecutionRunWork(tx, { companyId: input.companyId, runId: input.runId }); await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId }); await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); @@ -732,6 +734,7 @@ export function createPostgresRunDispatchAdapter( tx: Db, input: CancelSuppressedRetryInput, ): Promise { + await notifyExecutionRunWork(tx, { companyId: input.companyId, runId: input.runId }); await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); await notifyExternalAgentSettlement(tx, { companyId: input.companyId, runId: input.runId }); await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId }); @@ -1018,6 +1021,7 @@ export function createPostgresRunDispatchAdapter( now: Date, providerDispatchNotStarted = false, ): Promise { + await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 09fbb1749ac..cfb8919cc80 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -871,7 +871,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { expect(candidate?.id).toBe(wakeId); const agent = await ports.transaction.findInvokableAgent({ companyId, agentId: foreignAgentId }); expect(agent).toBeNull(); - const failed = await ports.transaction.failDeferredWake({ companyId, wakeId: candidate!.id, now: new Date() }); + const failed = await ports.transaction.failDeferredWake({ companyId, wakeId: candidate!.id, agentId: candidate!.agentId, now: new Date() }); expect(failed).toBe(true); return { outcome: { kind: "released" as const }, postCommitEffects: [] }; }); @@ -903,6 +903,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { async (_locked, ports) => { const cancelledUnderWrongCompany = await ports.transaction.cancelDeferredWake({ companyId: otherCompanyId, + agentId, wakeId, reason: "cross-company cancel attempt", now: new Date(), @@ -911,6 +912,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { const failedUnderWrongCompany = await ports.transaction.failDeferredWake({ companyId: otherCompanyId, + agentId, wakeId, now: new Date(), }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index e7a5f12f3f2..062b5bedb41 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "../../../services/live-run-recovery-notifications.js"; import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js"; import { notifyRunDispatch } from "../../../services/run-dispatch-notifications.js"; import { notifyExternalConversationSettlement } from "../../../services/external-conversation-settlement-notifications.js"; @@ -274,6 +275,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru throw error; } const now = new Date(); + await notifyStrandedAgentStatus(tx, { companyId, agentId: row.agentId }); await notifyChatInboundWakeNotices(tx, { companyId, issueId }); await tx .update(agentWakeupRequests) @@ -319,7 +321,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru return { liveNonSelfCommentIds, containedSelfAuthoredComment }; }, - async cancelDeferredWake({ companyId, wakeId, reason, now }) { + async cancelDeferredWake({ companyId, wakeId, agentId, reason, now }) { + await notifyStrandedAgentStatus(tx, { companyId, agentId }); await notifyChatInboundWakeNotices(tx, { companyId }); const rows = await tx .update(agentWakeupRequests) @@ -327,6 +330,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .where( and( eq(agentWakeupRequests.id, wakeId), + eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS), ), @@ -351,7 +355,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru return row ? toDeferredWakeCandidate(row) : null; }, - async failDeferredWake({ companyId, wakeId, now }) { + async failDeferredWake({ companyId, wakeId, agentId, now }) { + await notifyStrandedAgentStatus(tx, { companyId, agentId }); await notifyChatInboundWakeNotices(tx, { companyId }); const rows = await tx .update(agentWakeupRequests) @@ -364,6 +369,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .where( and( eq(agentWakeupRequests.id, wakeId), + eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS), ), diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index 184c799f82d..441a0c51271 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -1,3 +1,5 @@ +import { notifyStrandedAgentStatus } from "../../../services/live-run-recovery-notifications.js"; +import { notifyExecutionRunWork } from "../../../services/execution-control-notifications.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; @@ -52,7 +54,7 @@ export type QueuedCommentQueuePostgresAdapterDeps = { syncCommentExternalObjectsSafely(commentId: string, tx: Db): Promise; }; -function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentQueueTransaction { +function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps, wakeAgentId: string): QueuedCommentQueueTransaction { return { async updateCommentBody({ issueId, commentId, body, updatedAt }) { const updated = await tx @@ -100,6 +102,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async cancelWake({ wakeId, reason, now }) { + await notifyStrandedAgentStatus(tx, { companyId, agentId: wakeAgentId }); await notifyChatInboundWakeNotices(tx, { companyId, issueId }); await tx .update(agentWakeupRequests) @@ -108,6 +111,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async cancelQueueRun({ queueRunId, reason, now }) { + await notifyExecutionRunWork(tx, { companyId, runId: queueRunId }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); await notifyInstructionCollection(tx, { companyId, runId: queueRunId }); await notifyExternalAgentSettlement(tx, { companyId, runId: queueRunId }); @@ -308,7 +312,7 @@ export function createQueuedCommentIssueLockWriter(db: Db, deps: QueuedCommentQu .then((rows) => rows[0] ?? null) : null; - const transaction = buildTransaction(tx, companyId, input.issue.id, deps); + const transaction = buildTransaction(tx, companyId, input.issue.id, deps, wakeRow.agentId); const wake = toWakeRow(wakeRow); const queueRun = queueRunRow ? toRunRow(queueRunRow) : null; const activeRun = activeRunRow ? toRunRow(activeRunRow) : null; diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index 8827c82e4ca..66c8a631c15 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -123,6 +123,7 @@ export interface WakeQueueTransaction { cancelDeferredWake(input: { companyId: string; wakeId: string; + agentId: string; reason: string; now: Date; }): Promise; @@ -135,7 +136,7 @@ export interface WakeQueueTransaction { now: Date; }): Promise; /** Sets `status = 'failed'` guarded by the current `deferred_issue_execution` status. */ - failDeferredWake(input: { companyId: string; wakeId: string; now: Date }): Promise; + failDeferredWake(input: { companyId: string; wakeId: string; agentId: string; now: Date }): Promise; getPauseHoldFacts(input: { companyId: string; issueId: string; diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 069143cca94..79aada9b05f 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -191,6 +191,7 @@ async function runReleaseDrain( await ports.transaction.cancelDeferredWake({ companyId: run.companyId, wakeId: candidate.id, + agentId: candidate.agentId, reason: "Delegation closing note refers to completed child work", now: input.now, }); @@ -212,6 +213,7 @@ async function runReleaseDrain( await ports.transaction.cancelDeferredWake({ companyId: run.companyId, wakeId: candidate.id, + agentId: candidate.agentId, reason: "Deferred task messages now belong to the current assignee", now: input.now, }); @@ -260,6 +262,7 @@ async function runReleaseDrain( await ports.transaction.cancelDeferredWake({ companyId: run.companyId, wakeId: candidate.id, + agentId: candidate.agentId, reason: commentAction.selfAuthored ? "Deferred wake contained only comments authored by the finishing run" : "Queued messages were discarded before promotion", @@ -288,7 +291,7 @@ async function runReleaseDrain( }); if (wakeOutcome.kind === "fail_not_invokable") { - await ports.transaction.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, now: input.now }); + await ports.transaction.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, agentId: workingCandidate.agentId, now: input.now }); continue; } @@ -300,7 +303,7 @@ async function runReleaseDrain( if (wakeOutcome.kind === "cancel_pause_hold") { await ports.transaction.cancelDeferredWake({ companyId: run.companyId, - wakeId: workingCandidate.id, + wakeId: workingCandidate.id, agentId: workingCandidate.agentId, reason: "Deferred wake suppressed by active subtree pause hold", now: input.now, }); @@ -374,7 +377,7 @@ async function promoteDeferredWake( ) { await ports.transaction.cancelDeferredWake({ companyId: run.companyId, - wakeId: workingCandidate.id, + wakeId: workingCandidate.id, agentId: workingCandidate.agentId, reason: "Deferred execution wake no longer applies to a terminal task", now: input.now, }); diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index f8b8c3ebeab..ecb300c3204 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; import { notifyDeferredExecutionWork, notifyTaskRetrySettlement } from "../deferred-execution-notifications.js"; import { notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; @@ -1923,6 +1924,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti // Receipt existence proves the earlier wait, not current authority. // Retire only this unadmitted wake; an admitted run keeps its history. await db.transaction(async tx => { + await notifyStrandedAgentStatus(tx, { companyId: claimed.companyId, agentId: receipt.agentId }); await notifyChatInboundWakeNotices(tx, { companyId: claimed.companyId, issueId: String(claimed.payload.issueId), }); diff --git a/server/src/services/chat-channels/slack/session-stop.ts b/server/src/services/chat-channels/slack/session-stop.ts index ea922411c88..9468467255a 100644 --- a/server/src/services/chat-channels/slack/session-stop.ts +++ b/server/src/services/chat-channels/slack/session-stop.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "../../live-run-recovery-notifications.js"; import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js"; import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js"; @@ -585,6 +586,7 @@ export function createSlackSessionStop(db: Db, options: SlackSessionStopOptions) ["queued", "deferred_issue_execution"].includes(wakeup.status) && !issue.executionRunId ) { + await notifyStrandedAgentStatus(tx, { companyId: current.companyId, agentId: wakeup.agentId }); await notifyChatInboundWakeNotices(tx, { companyId: current.companyId, issueId: issue.id }); await tx .update(agentWakeupRequests) diff --git a/server/src/services/deferred-execution-notifications.ts b/server/src/services/deferred-execution-notifications.ts index b82cea725a5..89f7027e9fb 100644 --- a/server/src/services/deferred-execution-notifications.ts +++ b/server/src/services/deferred-execution-notifications.ts @@ -24,6 +24,7 @@ export async function notifyTaskRetrySettlement(tx: object, scope: DeferredExecu } export async function notifyDeferredExecutionEligibility(tx: object, scope?: { companyId: string }) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeProcessSessions, scope ? JSON.stringify({ companyId: scope.companyId }) : "*"); if (scope) { await notifyDeferredExecutionWork(tx, scope); await notifyNativeStatusWakeWork(tx, scope); diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index b79e0a7b24e..31aead54c7e 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -1,4 +1,4 @@ -import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js"; +import { notifyLiveRunRecovery, notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import type { Db } from "@paperclipai/db"; @@ -11,6 +11,7 @@ export async function notifyExecutionRunWork(transaction: object, scope?: Execut await notifyRunDispatch(transaction, scope); await notifyDeferredExecutionEligibility(transaction, scope); await notifyLiveRunRecovery(transaction, scope); + await notifyLiveResourceRecovery(transaction, scope); } for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { @@ -29,6 +30,7 @@ export function writeExecutionControl( if (scope) { await notifyDeferredExecutionEligibility(transaction, scope); await notifyLiveRunRecovery(transaction, scope); + await notifyLiveResourceRecovery(transaction, scope); } await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors); for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 962f15f327e..c0f0a9ebe25 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { writeDeferredExecutionWait } from "../deferred-execution-wait.js"; import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js"; import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; @@ -870,6 +871,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id }); await notifyExternalAgentSettlement(tx, { companyId: current.companyId, runId: current.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: current.companyId, runId: current.id }); + await notifyExecutionRunWork(tx, { companyId: current.companyId, agentId: current.agentId, runId: current.id }); await notifyUnsafeWorkspaceExportRunChange(tx, current); [terminal] = await tx .update(heartbeatRuns) @@ -996,6 +998,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); + await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); const [row] = await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), errorCode: "case_handoff_superseded", error: "This case handoff was cancelled or superseded.", updatedAt: new Date() }) @@ -1360,7 +1363,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); - await notifyUnsafeWorkspaceExportRunChange(tx, run); + await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id }); + await notifyUnsafeWorkspaceExportRunChange(tx, run); const [cancelled] = await tx .update(heartbeatRuns) .set({ @@ -1485,7 +1489,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); - await notifyUnsafeWorkspaceExportRunChange(tx, run); + await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id }); + await notifyUnsafeWorkspaceExportRunChange(tx, run); const [cancelled] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 089a4ab8f66..6a9eff739bb 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -2081,6 +2081,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD runtimeEnv.PAPERCLIP_INSTANCE_ID?.trim() || "paperclip-heartbeat", now, runIds: [...claimableNativeRunIds], + deferIneligible: !scope.runId, + onError: scope.onError, dispatch: (claim) => { const execution = executeRun(claim.runId, { nativeLeaseOwner: claim.leaseOwner, diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 9d3b30367d2..64d98ecd374 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; import { notifyRunDispatch } from "../run-dispatch-notifications.js"; import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; @@ -1418,12 +1419,13 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl ) { if (!wakeupRequestId) return; await db.transaction(async tx => { - const [current] = await tx.select({ companyId: agentWakeupRequests.companyId, status: agentWakeupRequests.status }) + const [current] = await tx.select({ companyId: agentWakeupRequests.companyId, agentId: agentWakeupRequests.agentId, status: agentWakeupRequests.status }) .from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeupRequestId)).for("update"); if (!current || current.status === "cancelled") return; if (current.status !== (patch?.status ?? status) || patch?.payload !== undefined || patch?.runId !== undefined) { await notifyExternalConversationSettlement(tx, { companyId: current.companyId }); } + await notifyStrandedAgentStatus(tx, { companyId: current.companyId, agentId: current.agentId }); await tx.update(agentWakeupRequests) .set({ status, ...patch, updatedAt: new Date() }) diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 2c05df61ceb..67c4e24187d 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -1222,6 +1223,11 @@ export function issueTreeControlService(db: Db) { if (issueIds.length === 0) return []; const now = new Date(); return db.transaction(async tx => { + const affected = await tx.select({ agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + isNull(agentWakeupRequests.runId), inArray(sql`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds), + )).for("update"); + for (const agentId of new Set(affected.map(wake => wake.agentId))) await notifyStrandedAgentStatus(tx, { companyId, agentId }); await notifyChatInboundWakeNotices(tx, { companyId }); return tx .update(agentWakeupRequests) diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 8955ace27c8..d494eb51b99 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,4 +1,5 @@ -import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; +import { notifyStrandedAgentStatus, notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; +import { finishedIdleWake } from "./idle-work-proof.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; @@ -11480,6 +11481,15 @@ export function issueService(db: Db) { // Invalidate human direction at the common assignment boundary, including // plugin/service writes that do not go through HTTP run cancellation. // Keep the requester attribution for audit; cancellation revokes its use. + const affectedWakes = await tx.select({ agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, receiptExisting.companyId), eq(agentWakeupRequests.requestedByActorType, "user"), + sql`(${sql.raw(finishedIdleWake("agent_wakeup_requests"))}) IS NOT TRUE`, + sql`coalesce(${agentWakeupRequests.payload}->>'issueId', ${agentWakeupRequests.payload}->>'taskId', + ${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'issueId', + ${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'taskId') = ${id}`, + )).for("update"); + for (const agentId of new Set(affectedWakes.map((wake: { agentId: string }) => wake.agentId))) + await notifyStrandedAgentStatus(tx, { companyId: receiptExisting.companyId, agentId }); await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId }); await tx.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date(), updatedAt: new Date() }) .where(and(eq(agentWakeupRequests.companyId, receiptExisting.companyId), diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index d24988d291c..913b6713e9c 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -80,6 +80,8 @@ export async function dispatchNativeSessionResumptions(input: { now?: Date; limit?: number; dispatch: (claim: NativeSessionResumeClaim) => void; + deferIneligible?: boolean; + onError?: (error: unknown) => void; }) { const claims = await claimNativeSessionResumptions(input); for (const claim of claims) input.dispatch(claim); @@ -98,6 +100,8 @@ export async function claimNativeSessionResumptions(input: { runIds?: string[]; now?: Date; limit?: number; + deferIneligible?: boolean; + onError?: (error: unknown) => void; }): Promise { const now = input.now ?? new Date(); const candidates = await input.db @@ -136,6 +140,7 @@ export async function claimNativeSessionResumptions(input: { for (const candidate of candidates) { const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`; let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null; + let claimFailed = false; const claimed = await input.db.transaction(async (tx) => { await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`); const [issue] = await tx.select().from(issues).where(and( @@ -351,6 +356,8 @@ export async function claimNativeSessionResumptions(input: { }); return true; }).catch((error: unknown) => { + claimFailed = true; + input.onError?.(error); terminalRunToEmit = null; logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId }, "failed to claim native session recovery candidate"); @@ -363,7 +370,7 @@ export async function claimNativeSessionResumptions(input: { void reportRunFailure(input.db, terminalRunToEmit); } if (claimed) claims.push({ runId: candidate.runId, leaseOwner }); - else await input.db.transaction(async tx => { + else if (claimFailed || input.deferIneligible !== false) await input.db.transaction(async tx => { await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`); await notifyLiveRunRecovery(tx, { companyId: candidate.companyId, runId: candidate.runId }); await tx.update(nativeRunFinalizations).set({ @@ -379,6 +386,7 @@ export async function claimNativeSessionResumptions(input: { or(isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)), )); }).catch((error: unknown) => { + input.onError?.(error); logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId }, "failed to defer native session recovery candidate"); }); diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 7a2efebdafd..b2f63d18360 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { usdToUnits, unitsToCents } from "@paperclipai/shared"; import { and, asc, eq, inArray, or, sql } from "drizzle-orm"; import { accountingRuntimeBaselines, budgetReservations, agentRuntimeState, agents, costEvents, heartbeatRuns, nativeRunFinalizations, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db"; @@ -38,6 +39,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService let [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update"); if (!run?.costAccountingPending || !terminalStatuses.includes(run.status)) return false; if (run.costAccountedAt) { + await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId }); await tx.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, run.id)); await budgetServiceInTransaction(tx, publications).evaluateAffectedScopes(run.companyId, run.agentId, text(object(object(run.usageJson).ledgerScope).projectId)); @@ -69,6 +71,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService if (!preProviderFailure && object(run.usageJson).accountingReceiptReady === false) return false; // Acknowledgement is part of the same transaction. Budget checks inside // this transaction must not mistake this receipt for missing accounting. + await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId }); await tx.update(heartbeatRuns).set({ costAccountingPending: false, costAccountedAt: new Date() }).where(eq(heartbeatRuns.id, run.id)); await tx.update(budgetReservations).set({ state: preProviderFailure ? "released" : "settled", settledAt: new Date() }) .where(and(eq(budgetReservations.companyId, run.companyId), eq(budgetReservations.runId, run.id), eq(budgetReservations.state, "held"))); diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index cba58357cf0..9661d2d72e0 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -6,6 +6,7 @@ export type RunDispatchScope = { companyId: string }; export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined); const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativeProcessSessions, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); } diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index ba5959b3eb6..a116aeba7fe 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { writeExecutionControl } from "./execution-control-notifications.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; @@ -183,6 +184,12 @@ export async function retireSupersededTaskRetry(db: Db, input: { sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${input.issueId}`), )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); if (latest?.id === input.sourceRunId) return false; + const [wake] = await db.select({ agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.requestId), + inArray(agentWakeupRequests.status, ["deferred_issue_execution", "queued", "claimed"]), + )).for("update"); + if (!wake) return false; + await notifyStrandedAgentStatus(db, { companyId: input.companyId, agentId: wake.agentId }); await notifyExternalConversationSettlement(db, input); const retired = await db.update(agentWakeupRequests).set({ status: "skipped", error: "task_retry_superseded", finishedAt: new Date(), updatedAt: new Date() }).where(and( @@ -375,6 +382,7 @@ export async function reconcileTaskChatRetryReceipts(db: Db, scope: { companyId? updatedAt: new Date(), }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"))); if (receipt?.runId || failed) await db.transaction(async tx => { + await notifyStrandedAgentStatus(tx, { companyId: wake.companyId, agentId: wake.agentId }); await notifyExternalConversationSettlement(tx, { companyId: wake.companyId, issueId: typeof wake.payload?.issueId === "string" ? wake.payload.issueId : undefined }); await write(tx);