From 5a7378af563e70cb41c877fdbc7d04d49aad687d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 22:04:58 -0600 Subject: [PATCH] fix(heartbeat): re-drive a queued run whose dispatch signal was lost after a restart (#20986) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The heartbeat queue starts queued runs. The run-dispatch worker re-runs the queue only when a write sends a "queued run dispatch" signal. > - A follow-up run can queue while its predecessor still holds its sandbox lease. The claim then declines the run without a log line, and the run stays `queued`. > - After a controller restart, nothing sent a new signal once that lease was settled. The run stayed `queued` forever. The issue had no Retry, and the reconciler counted the queued run as a valid next action. > - This pull request sends a dispatch signal from the native finalization writes that can clear a predecessor's cleanup gate, logs why a claim was declined, and bounds the wait. The dispatch worker re-drives an undispatched queued run after 60 s as a backstop, and the ownership reconciler reports and re-drives one still queued after 5 minutes. > - The benefit is that a follow-up starts within seconds of its predecessor settling, a lost signal delays it by about a minute instead of forever, and a run that still cannot start is visible. ## Linked Issues or Issue Description Refs #20909, #20935, #20953, #20970 **What happened?** Restart-chaos wave `w-20261111T223714-s9512` (master `6154b9485e` + #20907 + #20909, Daytona warm, Codex, W2, CR1 with a 360 s hold, `--ephemeral-server`, seed 9512), task t02. Run `0b95b1cd` crossed the outage with an accepted result. The next controller finalized it at 22:47:59Z. Its continuation run `1ad79418` was queued at 22:48:00Z (`retryOfRunId` = `0b95b1cd`, wake `d967700c`, source `native_status_decision`). The run stayed `queued` with `claimed_at = null` until the wave ended 19 minutes later. The agent stayed `running` with no running run. The task had no Retry. The grade was `R1:no_rest_within_budget` and U1 `no_owner_no_action ... queued 485s`. **Root cause** `claimQueuedRun` (`server/src/services/heartbeat/queue.ts`, `lockIssueExecutionClaim`) declines a queued run while `hasPendingRunCleanup` (`server/src/services/execution-blocker.ts`) is true for its predecessor. After the restart, the predecessor's warm `reuse_by_environment` lease was still `active` when the continuation queued, so the first claim was declined. The decline is correct, but nothing owned the retry. `registerRunDispatchWork` (`server/src/services/heartbeat/dispatch-work.ts`) has `next: async () => null`: the queue is only re-run by a later dispatch signal. At wave end the lease was `retained` with `cleanupStatus: success`, which does not block a claim. A dispatch pass in that state starts the run (verified locally). So no pass ran after the gate cleared. The ownership reconciler listed the run as `queued_run` with owner `run_dispatch` (always "alive"), and I7 treats any queued run as a next action, so nothing reported it. This is not the #20935 throttle. The wave base predates #20935, and a throttled intent keeps its wake intent `queued` with a `nextAllowedAt` deadline. Here the run row existed and its claim was never retried. On current master #20935 parks this specific `response_wake` self-wake, but any continuation, comment wake or Retry that queues behind a settling predecessor after a restart has the same gap. The red test uses a `same_agent` continuation, which master still enqueues. **Steps to reproduce** `server/src/__tests__/heartbeat-native-restart-suspension.test.ts`, "starts a continuation that queued behind its predecessor's sandbox release after a graceful restart". ## What Changed - Event-driven dispatch. A lease release already signals queued-run dispatch (`environmentsSvc.releaseLease`, `writeEnvironmentCleanup`, stop-and-retain settlement). I enumerated every `environment_leases` status writer: each one that moves a lease out of a blocking state already signals. The writers that do not signal only acquire a lease or set `pendingCleanupInFlight`, so they add a block and do not clear one. The native coordinator side had no signal, so: - `notifyNativeFinalizationWork` (the shared notifier of coordinator commits and terminal transitions) now also signals queued-run dispatch for the company. - The coordinator writes that clear `leaseOwner` or end the phase outside that notifier now signal in the same transaction: `releaseLapsedCoordinatorOwner`, native startup cleanup, observed-workspace preservation, `record_finalization_error` in the interaction bridge, the wake-queue terminal recovery, and the reconciler's ambiguous-state terminal failure. - Not changed: three non-transactional pre-dispatch closes in `run-completion.ts` and `agent-execution-stop.ts`. Their callers queue or settle the successor after the write, so the successor's own enqueue starts the queue. - `claimQueuedRun` now logs the reason when it declines a queued run at the issue execution claim (`predecessor_cleanup_pending`, `predecessor_still_executing`, `issue_execution_owned_by_another_run`, case handoff, task chat cleanup), when a predecessor is still settling in this process, and when a temporary budget hold keeps it queued. - New `server/src/services/queued-run-dispatch-liveness.ts`: `listUndispatchedQueuedRuns` reads queued runs in active companies whose ready, non-paused agent has no running run. It reads every page with a keyset cursor until the set is exhausted, so older held runs cannot hide a newer stranded run. `createQueuedRunRedriveSchedule` re-drives each such run 60 s after it was created, then every 2, 4, 8 minutes, capped at 15 minutes. Run ids that leave the set are dropped. - `server/src/services/heartbeat/dispatch-work.ts`: the queued-run dispatch worker's `next` now plans those re-drives. It re-reads only the companies the pass just dispatched, keeps the deadlines of every other company, and runs no query when the pass dispatched nothing. Only an instance-wide pass (startup or explicit wake) reads every company. It adds each due company to the dispatch scopes and returns the next due time. - `packages/db`: new partial index `heartbeat_runs_queued_created_idx (company_id, created_at, id) WHERE status = 'queued'` (migration `0470_heartbeat_runs_queued_created_index.sql`, `CREATE INDEX CONCURRENTLY IF NOT EXISTS`, registered as a top-level migration in `client.ts`). It returns `null` (no deadline, no SQL until a signal) when no run is undispatched or scheduling is suppressed. Every re-drive goes through the normal `startNextQueuedRunForAgent` and `claimQueuedRun` gates, so it never bypasses a hold, budget, pause or cleanup gate. - `server/src/services/heartbeat.ts`: wires `listUndispatchedQueuedRuns` into the worker (two lines). - `server/src/services/recovery/ownership-invariant-report.ts`: new check. A queued run still undispatched 5 minutes after creation is an I7 violation (`kind: queued_run`, `reason: queued_run_not_dispatched`), written to the activity log once. A younger one sets the report's next due time. - `server/src/services/native-runtime/native-finalization-reconciler.ts`: `applyOwnershipReportBackstops` sends a queued-run dispatch signal once, for each company with a newly found violation. Repeat passes do not signal again, so a held run does not wake the report every 30 s. - `tests/restart-chaos/README.md`: records the wave and the fix in the CR results table. The existing CR1 cell (#20909) and the enforced U1 check already cover this class in the harness. ## Verification - RED on master (`23909077ad`): - `heartbeat-native-restart-suspension.test.ts` "starts a continuation that queued behind its predecessor's sandbox release after a graceful restart": the continuation is `queued` behind the predecessor's `active` warm lease. The run-dispatch worker is registered, the lease then settles to `retained`/`success` with no dispatch signal, and the run stays queued: `AssertionError: expected null not to be null` after 75 s. - `ownership-reconciliation-report.test.ts` "reports and re-drives a queued run that no dispatch started while its agent had nothing running": `expected [] to deeply equal [ ObjectContaining{…} ]` (no violation for a run queued 19 minutes). - `ownership-reconciliation-report.test.ts` "signals queued-run dispatch when finalization work settles or a finished coordinator's lapsed owner is released": `expected [] to include 'settled'`. - GREEN on this branch: all pass. The restart test has two cases. When the lease is released through `releaseLease`, the continuation starts within 5 s. When the lease is settled by a write with no signal, the backstop starts it within 75 s. A new report test puts three older held runs ahead of a stranded run with a page size of 2, finds the stranded run, and checks that the backstop signals once. - `queued-run-dispatch-liveness.test.ts`: schedule bound, doubling, cap, and going quiet. - Focused suites pass (22 files, 495 tests): recovery-deadline-acceptance, live-recovery-work, heartbeat-run-dispatch-lifecycle, heartbeat-run-dispatch-work, deferred-execution-receipt-work, heartbeat-native-cleanup-admission, the ownership kinds suites, and others. `heartbeat-native-restart-suspension` and `ownership-reconciliation-report` pass in full. `tsc --noEmit` for the server passes. ## Risks - A queued run that is held on purpose (for example a temporary budget hold or a case handoff waiting for release) is now re-checked at most every 15 minutes after the first minute. Each re-check is one indexed query and the normal claim gates. Such a run is also reported once as an I7 violation. - An instance with an undispatched queued run keeps a dispatch deadline, so it does not idle while that run waits. An instance with no such run is unchanged. - Schema: one additive partial index, built concurrently. No data change. State saved by an older version is read as is. Nothing runs before `turn.start`. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, test runs). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- packages/db/src/client.ts | 1 + ...70_heartbeat_runs_queued_created_index.sql | 1 + ...{0465_snapshot.json => 0470_snapshot.json} | 102 +++++++++++++-- packages/db/src/migrations/meta/_journal.json | 7 ++ packages/db/src/schema/heartbeat_runs.ts | 2 + .../heartbeat-held-comment-wake.test.ts | 2 +- ...eartbeat-native-restart-suspension.test.ts | 72 ++++++++++- .../ownership-reconciliation-report.test.ts | 113 ++++++++++++++++- .../modules/wake-queue/adapters/postgres.ts | 3 +- server/src/services/heartbeat.ts | 3 + .../services/heartbeat/dispatch-work.test.ts | 37 ++++++ .../src/services/heartbeat/dispatch-work.ts | 32 ++++- server/src/services/heartbeat/queue.ts | 32 +++-- .../native-finalization-notifications.ts | 2 + .../native-finalization-reconciler.ts | 9 ++ .../native-interaction-bridge.ts | 2 + .../native-observed-workspace-preservation.ts | 2 + .../native-runtime/native-startup-cleanup.ts | 2 + .../queued-run-dispatch-liveness.test.ts | 52 ++++++++ .../services/queued-run-dispatch-liveness.ts | 119 ++++++++++++++++++ .../recovery/ownership-invariant-report.ts | 29 +++++ .../services/workspace-recovery-stop-proof.ts | 2 + tests/restart-chaos/README.md | 1 + 23 files changed, 604 insertions(+), 23 deletions(-) create mode 100644 packages/db/src/migrations/0470_heartbeat_runs_queued_created_index.sql rename packages/db/src/migrations/meta/{0465_snapshot.json => 0470_snapshot.json} (99%) create mode 100644 server/src/services/heartbeat/dispatch-work.test.ts create mode 100644 server/src/services/queued-run-dispatch-liveness.test.ts create mode 100644 server/src/services/queued-run-dispatch-liveness.ts diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 6c30c94fecd..ee080641142 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -630,6 +630,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0459_environment_lease_backup_scope_index.sql", "0460_workspace_activity_context_index.sql", "0461_created_from_legacy_origin_run_index.sql", + "0470_heartbeat_runs_queued_created_index.sql", "0469_zippy_photon.sql", diff --git a/packages/db/src/migrations/0470_heartbeat_runs_queued_created_index.sql b/packages/db/src/migrations/0470_heartbeat_runs_queued_created_index.sql new file mode 100644 index 00000000000..760231bbdc6 --- /dev/null +++ b/packages/db/src/migrations/0470_heartbeat_runs_queued_created_index.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_queued_created_idx" ON "heartbeat_runs" USING btree ("company_id","created_at","id") WHERE "heartbeat_runs"."status" = 'queued'; diff --git a/packages/db/src/migrations/meta/0465_snapshot.json b/packages/db/src/migrations/meta/0470_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0465_snapshot.json rename to packages/db/src/migrations/meta/0470_snapshot.json index 951e3c6deea..5cc01ce9202 100644 --- a/packages/db/src/migrations/meta/0465_snapshot.json +++ b/packages/db/src/migrations/meta/0470_snapshot.json @@ -1,6 +1,6 @@ { - "id": "38dbf2b9-da26-4c3a-b813-f37d0c19cba2", - "prevId": "e2a678da-05cc-4316-a40e-d85a69be01e3", + "id": "f0e17e47-6217-4080-b6b6-9d83b33e9ced", + "prevId": "ac25715a-74c6-4988-8525-63ad14781c74", "version": "7", "dialect": "postgresql", "tables": { @@ -1417,6 +1417,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "activity_log_company_handled_for_created_idx": { + "name": "activity_log_company_handled_for_created_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"details\" ->> 'handledForUserId')", + "asc": true, + "isExpression": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "(\"activity_log\".\"details\" ->> 'handledForUserId') IS NOT NULL", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -16340,6 +16368,18 @@ "primaryKey": false, "notNull": false }, + "default_decision_owner_user_id": { + "name": "default_decision_owner_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "budget_owner_user_id": { + "name": "budget_owner_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "require_board_approval_for_new_agents": { "name": "require_board_approval_for_new_agents", "type": "boolean", @@ -35606,6 +35646,34 @@ "method": "btree", "with": {} }, + "heartbeat_runs_queued_created_idx": { + "name": "heartbeat_runs_queued_created_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"status\" = 'queued'", + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -36235,6 +36303,12 @@ "primaryKey": false, "notNull": false }, + "fingerprint": { + "name": "fingerprint", + "type": "text", + "primaryKey": false, + "notNull": false + }, "created_at": { "name": "created_at", "type": "timestamp with time zone", @@ -51034,6 +51108,18 @@ "primaryKey": false, "notNull": false }, + "environment_id": { + "name": "environment_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "decision_owner_user_id": { + "name": "decision_owner_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "target_date": { "name": "target_date", "type": "date", @@ -51058,12 +51144,6 @@ "primaryKey": false, "notNull": false }, - "environment_id": { - "name": "environment_id", - "type": "uuid", - "primaryKey": false, - "notNull": false - }, "git_authorization_grant_id": { "name": "git_authorization_grant_id", "type": "uuid", @@ -62996,6 +63076,12 @@ "notNull": true, "default": false }, + "time_zone": { + "name": "time_zone", + "type": "text", + "primaryKey": false, + "notNull": false + }, "created_at": { "name": "created_at", "type": "timestamp with time zone", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index b3d5bd28d99..c418c97f419 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3277,6 +3277,13 @@ "when": 1794428870590, "tag": "0469_zippy_photon", "breakpoints": true + }, + { + "idx": 470, + "version": "7", + "when": 1794449318551, + "tag": "0470_heartbeat_runs_queued_created_index", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index b10c7ca8d61..44fcd369ae7 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -176,6 +176,8 @@ export const heartbeatRuns = pgTable( companyWakeCommentIdx: index("heartbeat_runs_company_wake_comment_idx") .on(table.companyId, sql`(${table.contextSnapshot}->>'wakeCommentId')`) .concurrently(), + queuedCreatedIdx: index("heartbeat_runs_queued_created_idx").on(table.companyId, table.createdAt, table.id) + .where(sql`${table.status} = 'queued'`).concurrently(), costAccountingPendingIdx: index("heartbeat_runs_cost_accounting_pending_idx").on(table.updatedAt, table.id).where(sql`${table.costAccountingPending} = true`), costDataQualityIdx: index("heartbeat_runs_cost_data_quality_idx") .on(table.companyId, sql`coalesce(${table.finishedAt}, ${table.createdAt}) desc`, table.id) diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index ea76358028e..b80b100bb65 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -656,7 +656,7 @@ describePostgres("saved intents while another run's accounting is pending", () = it("keeps a reconciler pass due when the backstop cannot save a message", async () => { const report = { violations: [{ rule: "I4_exactly_once_intent", kind: "issue_comment", recordId: randomUUID(), - companyId: "not-a-company-id", runId: null, issueId: "not-an-issue-id", reason: "user_comment_without_wake" }] }; + companyId: "not-a-company-id", runId: null, issueId: "not-an-issue-id", reason: "user_comment_without_wake" }], newViolations: [] }; expect(await applyOwnershipReportBackstops(db, report as never)).toEqual({ retryNeeded: true }); }); diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index b32671995b8..5b27b408f10 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -17,12 +17,13 @@ import type { NativeRunIdentity, NativeSession, PrpEvent } from "@paperclipai/pa import { randomUUID } from "node:crypto"; import * as controllerProcess from "../services/controller-process-identity.js"; import { and, eq, sql } from "drizzle-orm"; -import { activityLog, agents, createDb, agentWakeupRequests, authUsers, budgetReservations, companyMemberships, completionContracts, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db"; +import { activityLog, agents, environments, createDb, agentWakeupRequests, authUsers, budgetReservations, companyMemberships, completionContracts, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db"; import { describe, expect, it, vi } from "vitest"; import { subscribeDatabaseWork } from "@paperclipai/db/work-signals"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { heartbeatService } from "../services/heartbeat.js"; +import { environmentService } from "../services/environments.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { claimNativeRestartRecoveries, classifyNativeRunnerRecoveryEvidence, currentNativeControllerIdentity } from "../services/native-runtime/native-restart-recovery.js"; import type { NativeSessionBackend } from "@paperclipai/paperclip-runner"; @@ -1766,4 +1767,73 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut expect(outcome).toEqual({ settled: false, reason: "owner_changed_since_proof" }); expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); }, 60_000); + + it.each([ + { settledBy: "lease release", signalled: true, startWithinMs: 5_000 }, + { settledBy: "unsignalled write", signalled: false, startWithinMs: 75_000 }, + ])("starts a continuation that queued behind its predecessor's sandbox release after a graceful restart ($settledBy)", async ({ settledBy, signalled, startWithinMs }) => { + const f = await seedRecoverableRun(`ACCEPTED-CONTINUE-THEN-SIGTERM-${settledBy}`, true); + const [contract] = await f.db.select().from(completionContracts).where(eq(completionContracts.issueId, f.issueId)); + await f.db.update(heartbeatRuns).set({ completionContractId: contract.id, + completionContractSha256: contract.canonicalSha256 }).where(eq(heartbeatRuns.id, f.runId)); + const controller = await currentNativeControllerIdentity(); + await f.db.update(nativeRunFinalizations).set({ + leaseOwner: `${f.native.runnerInstanceId}:${randomUUID()}`, leaseExpiresAt: new Date(Date.now() + 20 * 60_000), + controllerBootId: controller.bootId, controllerHostHash: (await controllerProcess.currentControllerProcess()).hostHash, + controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: 1, + }).where(eq(nativeRunFinalizations.runId, f.runId)); + const [prepared] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const identity = { companyId: fixture.current.companyId, issueId: f.issueId, runId: f.runId, + agentId: prepared.agentId, sessionId: prepared.nativeSessionId! }; + const port = new PaperclipControlPlanePort(f.db, { ...identity, completionContractId: prepared.completionContractId!, + completionContractSha256: prepared.completionContractSha256!, sourceInstanceId: prepared.runnerInstanceId!, + controlPlaneSourceInstanceId: "continue-after-restart-control" }); + await port.openRun({ ...CONTROL_PLANE_CONFORMANCE_OPEN, identity, sourceInstanceId: prepared.runnerInstanceId! }); + const { revision, criterionIds } = f.native.completionContract; + await port.completeRun({ + result: { ...CONTROL_PLANE_CONFORMANCE_RESULT, reportedWorkDisposition: "yielded", + continuation: { kind: "same_agent", summary: "Continue with phase two.", idempotencyKey: `continue-${f.runId}` }, + completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: revision, + criteria: criterionIds.map((criterionId) => ({ criterionId, status: "satisfied" as const, evidenceRefs: ["event:2"] })) } }, + terminal: { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, reportedWorkDisposition: "yielded" }, + callerResultId: "continue-after-restart-result", + }); + await f.db.insert(workspaceOperations).values({ companyId: fixture.current.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, phase: "workspace_finalize", status: "running", cwd: fixture.current.runtimeRoot }); + const [environment] = await f.db.insert(environments).values({ + name: `Warm sandbox ${settledBy}`, driver: "sandbox", config: { provider: "daytona", reuseLease: true } }).returning(); + const [warmLease] = await f.db.insert(environmentLeases).values({ environmentId: environment!.id, + companyId: fixture.current.companyId, issueId: f.issueId, heartbeatRunId: f.runId, status: "active", + leasePolicy: "reuse_by_environment", provider: "daytona", providerLeaseId: `sandbox-${randomUUID()}` }).returning(); + + await heartbeatService(f.db).drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]); + const providerDispatch = vi.fn(() => { throw new Error("the continuation does not reach a provider here"); }); + const next = heartbeatService(f.db, { nativeSessionBackendFactory: providerDispatch }); + try { + await next.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + await reconcileNativeFinalizations(f.db, [f.runId]); + await next.dispatchPendingNativeStatusWakeups(); + } finally { await next.drainActiveRunExecutions(); } + const successors = () => f.db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.agentId, prepared.agentId), eq(heartbeatRuns.retryOfRunId, f.runId))); + const [queued] = await successors(); + expect(queued).toMatchObject({ status: "queued", startedAt: null }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]?.status).toBe("succeeded"); + + if (!signalled) await f.db.update(heartbeatRuns).set({ createdAt: new Date(Date.now() - 2 * 60_000) }) + .where(eq(heartbeatRuns.id, queued!.id)); + const coordinator = createDeliveryWorkCoordinator({ owner: f.db, canRun: () => true, onError: () => undefined }); + const later = heartbeatService(f.db, { nativeSessionBackendFactory: providerDispatch }); + try { + await later.registerRunDispatchWork(coordinator, () => true).ready; + expect((await successors())[0]?.status).toBe("queued"); + const settledAt = Date.now(); + if (signalled) await environmentService(f.db).releaseLease(warmLease!.id, "retained", { cleanupStatus: "success" }); + else await f.db.update(environmentLeases).set({ status: "retained", cleanupStatus: "success" }) + .where(eq(environmentLeases.id, warmLease!.id)); + await expect.poll(async () => (await successors())[0]?.startedAt ?? null, { timeout: startWithinMs, interval: 250 }) + .not.toBeNull(); + expect(Date.now() - settledAt).toBeLessThan(startWithinMs); + } finally { await coordinator.stop(); await later.drainActiveRunExecutions(); } + }, 150_000); }); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index fa8573c50d2..84372b632f4 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -26,7 +26,13 @@ import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.j import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { currentControllerBootId, currentProcessSpaceHash } from "../services/controller-process-identity.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; -import { registerOwnershipReconciliationWork } from "../services/native-runtime/native-finalization-reconciler.js"; +import { applyOwnershipReportBackstops, registerOwnershipReconciliationWork } from "../services/native-runtime/native-finalization-reconciler.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { listUndispatchedQueuedRuns } from "../services/queued-run-dispatch-liveness.js"; +import { releaseLapsedCoordinatorOwner } from "../services/workspace-recovery-stop-proof.js"; +import { subscribeDatabaseWork } from "@paperclipai/db/work-signals"; import { createOwnershipReportState, OWNERSHIP_INVARIANT_VIOLATION_ACTION, @@ -961,6 +967,111 @@ describe("report-only ownership reconciliation", () => { expect(report.nextDueAt?.getTime()).toBe(updatedAt.getTime() + 5 * 60_000); }); + it("reports and re-drives a queued run that no dispatch started while its agent had nothing running", async () => { + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Continuation", status: "in_progress", assigneeAgentId: agentId }); + const [stranded] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "queued", invocationSource: "automation", + contextSnapshot: { issueId }, createdAt: minutesAgo(19) }).returning(); + const signals: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.queuedRunDispatch, event => signals.push(event)); + try { + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.kind === "queued_run")).toEqual([ + expect.objectContaining({ rule: "I7_visible_next_action", recordId: stranded!.id, runId: stranded!.id, + issueId, companyId, reason: "queued_run_not_dispatched" })]); + await applyOwnershipReportBackstops(db, report); + expect(signals).toContain("settled"); + } finally { unsubscribe(); } + }); + + it("finds a stranded run behind older queued rows past the first page, and re-drives it only once", async () => { + const otherCompany = randomUUID(); + const otherAgent = randomUUID(); + await db.insert(companies).values({ id: otherCompany, name: "Held queue", issuePrefix: "HLD" }); + await db.insert(agents).values({ id: otherAgent, companyId: otherCompany, name: "Held", role: "engineer", adapterType: "codex_local" }); + await db.insert(heartbeatRuns).values([40, 39, 38].map(minutes => + ({ companyId: otherCompany, agentId: otherAgent, status: "queued", invocationSource: "automation", createdAt: minutesAgo(minutes) }))); + const [stranded] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "queued", invocationSource: "automation", + createdAt: minutesAgo(19) }).returning(); + const state = createOwnershipReportState(); + const first = await runOwnershipReconciliationPass(db, { state, limitPerKind: 2 }); + expect(first.violations.filter(violation => violation.kind === "queued_run").map(violation => violation.recordId)) + .toContain(stranded!.id); + const signals: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.queuedRunDispatch, event => signals.push(event)); + try { + await applyOwnershipReportBackstops(db, first); + expect(signals).toContain("settled"); + signals.length = 0; + const second = await runOwnershipReconciliationPass(db, { state, limitPerKind: 2 }); + expect(second.violations.filter(violation => violation.kind === "queued_run")).toHaveLength(4); + await applyOwnershipReportBackstops(db, second); + expect(signals).toEqual([]); + } finally { unsubscribe(); } + }); + + it("does not report queued runs a worktree runtime will never start: before its cutoff or while execution is off", async () => { + const copied = randomUUID(); + await db.insert(heartbeatRuns).values({ id: copied, companyId, agentId, status: "queued", invocationSource: "automation", + createdAt: minutesAgo(19) }); + const queuedRunViolations = async () => (await runOwnershipReconciliationPass(db, + { state: createOwnershipReportState(), record: false })).violations.filter(violation => violation.kind === "queued_run"); + vi.stubEnv("PAPERCLIP_IN_WORKTREE", "true"); + vi.stubEnv("PAPERCLIP_INSTANCE_ID", "ownership-worktree"); + const settings = instanceSettingsService(db); + try { + expect(await queuedRunViolations()).toEqual([]); + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + expect(await queuedRunViolations()).toEqual([]); + expect(await listUndispatchedQueuedRuns(db, { createdAfter: minutesAgo(1) })).toEqual([]); + expect((await listUndispatchedQueuedRuns(db)).map(run => run.runId)).toEqual([copied]); + } finally { + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + vi.unstubAllEnvs(); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR", orphanSpool); + } + }); + + it("signals queued-run dispatch when finalization work settles or a finished coordinator's lapsed owner is released", async () => { + const issueId = randomUUID(); + const runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Settling", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", invocationSource: "automation", + runtimeMode: "native", nativeIssueId: issueId, finishedAt: minutesAgo(2) }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "committed", + leaseOwner: "native-finalizer:lapsed", leaseExpiresAt: minutesAgo(1) }); + const signals: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.queuedRunDispatch, event => signals.push(event)); + try { + await db.transaction(tx => notifyNativeFinalizationWork(tx, { companyId, runId, issueId })); + expect(signals).toContain("settled"); + signals.length = 0; + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + expect(await db.transaction(tx => releaseLapsedCoordinatorOwner(tx, { companyId, runId }, coordinator, new Date()))).toBe(true); + expect(signals).toContain("settled"); + } finally { unsubscribe(); } + }); + + it("does not report a queued run behind a running run, on a paused agent, or still inside its grace window", async () => { + const otherAgent = randomUUID(); + const pausedAgent = randomUUID(); + await db.insert(agents).values([ + { id: otherAgent, companyId, name: "Busy", role: "engineer", adapterType: "codex_local" }, + { id: pausedAgent, companyId, name: "Paused", role: "engineer", adapterType: "codex_local", status: "paused" }, + ]); + await db.insert(heartbeatRuns).values([ + { companyId, agentId: otherAgent, status: "running", invocationSource: "manual" }, + { companyId, agentId: otherAgent, status: "queued", invocationSource: "automation", createdAt: minutesAgo(30) }, + { companyId, agentId: pausedAgent, status: "queued", invocationSource: "automation", createdAt: minutesAgo(30) }, + ]); + const createdAt = minutesAgo(1); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "queued", invocationSource: "automation", createdAt }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.kind === "queued_run")).toEqual([]); + expect(report.nextDueAt?.getTime()).toBeLessThanOrEqual(createdAt.getTime() + 5 * 60_000); + }); + it("retries a violation whose activity row failed to save", async () => { await mkdir(orphanSpool, { recursive: true }); const laterCompany = randomUUID(); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 6b507505e9b..d85bc0712eb 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,6 +1,6 @@ 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 { notifyQueuedRunDispatch, notifyRunDispatch } from "../../../services/run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js"; import { notifyExternalConversationSettlement } from "../../../services/external-conversation-settlement-notifications.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; @@ -871,6 +871,7 @@ async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: ReleaseRun, iss } } if (!existing.length) { + await notifyQueuedRunDispatch(tx, { companyId: issue.companyId }); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index aa92629d756..eb94fc0cb15 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -14,6 +14,7 @@ import { registerProviderLimitResumeWork, type PrepareChatRetry } from "./native import { reconcileTaskChatRetryReceipts } from "./task-retry.js"; import { registerIssueRecoveryWork } from "./recovery/work.js"; import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js"; +import { listUndispatchedQueuedRuns } from "./queued-run-dispatch-liveness.js"; import { NATIVE_RETAINED_OWNER_WAIT } from "./native-runtime/native-controller-recovery.js"; import { describeFailedAttachment } from "./native-runtime/remote-warm-session-recovery.js"; import { createProviderLimitWait } from "./heartbeat/provider-limit-wait.js"; @@ -4599,6 +4600,8 @@ export function heartbeatService( promoteDueScheduledRetries, nextScheduledRetryAt, resumeQueuedRunsAfterDrain, + listUndispatchedQueuedRuns: async scope => listUndispatchedQueuedRuns(db, + { ...scope, createdAfter: await getWorktreeExecutionCutoff() }), schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, }, () => canRun() && readTaskDrain(new Date()) === null), promoteDueScheduledRetries, diff --git a/server/src/services/heartbeat/dispatch-work.test.ts b/server/src/services/heartbeat/dispatch-work.test.ts new file mode 100644 index 00000000000..e929cd79820 --- /dev/null +++ b/server/src/services/heartbeat/dispatch-work.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it, vi } from "vitest"; +import { createDb } from "@paperclipai/db"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { registerRunDispatchWork } from "./dispatch-work.js"; +import type { UndispatchedQueuedRun } from "../queued-run-dispatch-liveness.js"; + +describe("queued-run dispatch re-drive", () => { + it("keeps the companies to re-read when the undispatched-run read fails, and re-drives after the retry", async () => { + const owner = createDb("postgres://paperclip:unused@127.0.0.1:1/unused", { maxConnections: 1 }); + const stranded: UndispatchedQueuedRun = { runId: "run-1", companyId: "company-1", agentId: "agent-1", issueId: null, + createdAt: new Date(Date.now() - 10 * 60_000) }; + const reads: Array = []; + const list = vi.fn(async (scope?: { companyId: string }) => { + reads.push(scope?.companyId); + if (reads.length === 1) throw new Error("transient read failure"); + return [stranded]; + }); + const resumed: Array = []; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: error => errors.push(error) }); + try { + await registerRunDispatchWork(coordinator, { + promoteDueScheduledRetries: async () => undefined, + nextScheduledRetryAt: async () => null, + schedulingSuppressed: async () => false, + resumeQueuedRunsAfterDrain: async scope => { resumed.push(scope?.companyId); }, + listUndispatchedQueuedRuns: list, + }, () => true).ready; + await expect.poll(() => resumed, { timeout: 45_000, interval: 250 }).toContain("company-1"); + expect(reads.slice(0, 2)).toEqual([undefined, undefined]); + expect(errors).toEqual([expect.objectContaining({ message: "transient read failure" })]); + } finally { + await coordinator.stop(); + await owner.$client.end({ timeout: 1 }); + } + }, 60_000); +}); diff --git a/server/src/services/heartbeat/dispatch-work.ts b/server/src/services/heartbeat/dispatch-work.ts index 9c95e123f2f..56c6ee4b7b8 100644 --- a/server/src/services/heartbeat/dispatch-work.ts +++ b/server/src/services/heartbeat/dispatch-work.ts @@ -2,6 +2,7 @@ import type { RunDispatchScope } from "../run-dispatch-notifications.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { registerDeadlineWork } from "../deadline-work.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { createQueuedRunRedriveSchedule, type QueuedRunRedrivePlan, type UndispatchedQueuedRun } from "../queued-run-dispatch-liveness.js"; export function registerRunDispatchWork( coordinator: ReturnType, @@ -9,6 +10,7 @@ export function registerRunDispatchWork( promoteDueScheduledRetries(): Promise; nextScheduledRetryAt(): Promise; resumeQueuedRunsAfterDrain(scope?: RunDispatchScope): Promise; + listUndispatchedQueuedRuns?(scope?: RunDispatchScope): Promise; schedulingSuppressed(): Promise; }, canRun: () => boolean, @@ -26,6 +28,8 @@ export function registerRunDispatchWork( next: () => suppressed ? Promise.resolve(null) : owner.nextScheduledRetryAt(), }); const scopes = new Set(["*"]); + const redrive = createQueuedRunRedriveSchedule(); + let dispatched: "all" | Set = new Set(); const queue = registerDeadlineWork(coordinator, DELIVERY_QUEUES.queuedRunDispatch, { retryMs: 30_000, notificationCoalesceMs: 1_000, @@ -41,14 +45,38 @@ export function registerRunDispatchWork( return; } try { - await owner.resumeQueuedRunsAfterDrain(requested[index] === "*" ? undefined : JSON.parse(requested[index]) as RunDispatchScope); + const scope = requested[index] === "*" ? undefined : JSON.parse(requested[index]) as RunDispatchScope; + await owner.resumeQueuedRunsAfterDrain(scope); + if (!scope) dispatched = "all"; + else if (dispatched !== "all") dispatched.add(scope.companyId); } catch (error) { for (const hint of requested.slice(index)) scopes.add(hint); throw error; } } }, - next: async () => null, + next: async () => { + if (!owner.listUndispatchedQueuedRuns) return null; + const list = owner.listUndispatchedQueuedRuns; + const refreshed = dispatched; + dispatched = new Set(); + const keep = () => { + if (refreshed === "all" || dispatched === "all") dispatched = "all"; + else for (const companyId of refreshed) dispatched.add(companyId); + }; + let plan: QueuedRunRedrivePlan; + try { + if (await owner.schedulingSuppressed()) { keep(); return null; } + const runs = refreshed === "all" ? await list() + : (await Promise.all([...refreshed].map(companyId => list({ companyId })))).flat(); + plan = redrive.refresh(refreshed, runs); + } catch (error) { + keep(); + throw error; + } + for (const companyId of plan.companyIds) scopes.add(JSON.stringify({ companyId })); + return plan.nextAt; + }, }); return { retries, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 030a3abbf18..307edfc74d9 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -987,7 +987,11 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // accepted work. Cancelling the unstarted run would drop the user's // message, Retry, or recovery it carries; keep it queued. Settling // accounting re-notifies run dispatch. A reached hard stop is the board's. - if (!run.startedAt && !(await isBoardOwnedBudgetBlock(db, run.companyId, budgetBlock))) return null; + if (!run.startedAt && !(await isBoardOwnedBudgetBlock(db, run.companyId, budgetBlock))) { + logger.info({ runId: run.id, agentId: run.agentId, companyId: run.companyId, reason: "budget_hold_settling", + scopeType: budgetBlock.scopeType }, "claimQueuedRun: queued run kept queued behind a temporary budget hold"); + return null; + } // Record which budget stops this queued run, so Raise budget and resume // continues it (budget-resume-continuation.ts). Saved before the cancel // so a crash between them cannot lose it; the resume matches it only @@ -1055,7 +1059,12 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, )) .limit(1); - if (settlingOwner) return null; + if (settlingOwner) { + logger.info({ runId: run.id, agentId: run.agentId, companyId: run.companyId, issueId, + reason: "predecessor_settling_in_process", predecessorRunId: settlingOwner.id }, + "claimQueuedRun: queued run declined at the issue execution claim"); + return null; + } } if (issueId) { const activePauseHold = await treeControlSvc.getActivePauseHoldGate( @@ -1197,8 +1206,13 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende context.wakeReason !== "source_scoped_recovery_action"; const [handoff] = issueId ? await tx.select().from(caseHandoffs).where(and(eq(caseHandoffs.companyId, run.companyId), eq(caseHandoffs.issueId, issueId), inArray(caseHandoffs.state, ["awaiting_release", "pending_dispatch"]))).limit(1) : []; - if (handoff && (handoff.wakeupRequestId !== run.wakeupRequestId || handoff.state === "awaiting_release")) + const declined = (reason: string, predecessorRunId?: string) => { + logger.info({ runId: run.id, agentId: run.agentId, companyId: run.companyId, issueId, reason, + ...(predecessorRunId ? { predecessorRunId } : {}) }, "claimQueuedRun: queued run declined at the issue execution claim"); return { ownsIssue, blocked: true }; + }; + if (handoff && (handoff.wakeupRequestId !== run.wakeupRequestId || handoff.state === "awaiting_release")) + return declined("case_handoff_pending"); if (caseHandoffReceipt) { const [current] = await tx.select({ state: caseHandoffs.state, version: caseHandoffs.version, currentVersion: cases.handoffVersion, workIssueId: cases.workIssueId, @@ -1209,18 +1223,18 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (!current || !["pending_dispatch", "delivered"].includes(current.state) || current.version !== current.currentVersion || current.workIssueId !== issueId || current.toAgentId !== run.agentId || current.wakeupRequestId !== run.wakeupRequestId || !ownsIssue || (!run.startedAt && current.appliedIssueStatusVersion !== owner?.statusVersion)) - return { ownsIssue, blocked: true }; + return declined("case_handoff_superseded"); } if (ownsIssue && ["native_safe_replacement", "native_provider_overloaded"].includes(run.scheduledRetryReason ?? "") && owner.checkoutRunId && owner.checkoutRunId !== run.id) { - return { ownsIssue, blocked: true }; + return declined("issue_checked_out_by_another_run", owner.checkoutRunId); } if (run.scheduledRetryReason === "native_provider_overloaded" && owner?.executionRunId && owner.executionRunId !== run.id && owner.executionRunId !== run.retryOfRunId) { - return { ownsIssue, blocked: true }; + return declined("issue_execution_owned_by_another_run", owner.executionRunId); } if (ownsIssue && issueId && await taskChatRetrySourceCleanupPending(tx, run, issueId)) { - return { ownsIssue, blocked: true }; + return declined("task_chat_retry_source_cleanup_pending"); } const executionOwnerId = run.scheduledRetryReason === "native_provider_overloaded" ? run.retryOfRunId : ownsIssue ? owner?.executionRunId : null; @@ -1240,10 +1254,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // another controller. Local absence alone is not a release receipt. if (!isHeartbeatRunTerminalStatus(previous?.status) || liveRunExecutions.has(previousRunId)) { - return { ownsIssue, blocked: true }; + return declined("predecessor_still_executing", previousRunId); } if (await hasPendingRunCleanup(tx, run.companyId, previousRunId)) { - return { ownsIssue, blocked: true }; + return declined("predecessor_cleanup_pending", previousRunId); } } return { ownsIssue, blocked: false }; diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index 623c4b23d41..cae576bd2bb 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -12,6 +12,7 @@ import { notifyNativeStatusReassessment } from "./native-status-reassessment-not import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { notifyQueuedRunDispatch } from "./run-dispatch-notifications.js"; export type NativeFinalizationWorkScope = { companyId: string; runId?: string; issueId?: string }; @@ -31,6 +32,7 @@ export async function notifyNativeFinalizationWork(transaction: object, scope: N await notifyNativeResponsePresentation(transaction, scope); await notifyNativeReviewResponse(transaction, scope); } + await notifyQueuedRunDispatch(transaction, { companyId: scope.companyId }); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativePendingEffects, JSON.stringify({ companyId: scope.companyId, runId: scope.runId })); } diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index c22b0389015..489113814a0 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -29,6 +29,8 @@ import { type OwnershipReport, } from "../recovery/ownership-invariant-report.js"; import { deferredWorkIntentWait, promoteDeferredWorkIntents, saveUndeliveredUserComments } from "../work-intent-outbox.js"; +import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js"; +import { QUEUED_RUN_NOT_DISPATCHED } from "../queued-run-dispatch-liveness.js"; import { createHash, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, not, notInArray, or, sql, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -242,6 +244,7 @@ export async function claimNativeSessionResumptions(input: { const failureCode = originalFailureCode; await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, row.run.companyId); await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); + await notifyQueuedRunDispatch(tx, { companyId: row.run.companyId }); await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, @@ -866,6 +869,12 @@ export async function applyOwnershipReportBackstops(db: Db, report: OwnershipRep await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId, trigger: "reconciler_backstop" })) .catch(error => { retryNeeded = true; logger.warn({ err: error, companyId }, "deferred intent backstop promotion failed"); }); } + const undispatched = new Set(report.newViolations.filter(violation => violation.reason === QUEUED_RUN_NOT_DISPATCHED) + .flatMap(violation => violation.companyId ? [violation.companyId] : [])); + for (const companyId of undispatched) { + await db.transaction(tx => notifyQueuedRunDispatch(tx, { companyId })) + .catch(error => { retryNeeded = true; logger.warn({ err: error, companyId }, "queued run dispatch backstop failed"); }); + } return { retryNeeded }; } diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index aeae4fd2c84..97204c41acf 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -1,3 +1,4 @@ +import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import type { Db } from "@paperclipai/db"; @@ -134,6 +135,7 @@ export async function applyNativeAttentionStatusDecision(input: { if (effect.cause === "native_workspace_sync_out_unsafe_archive") { await notifyUnsafeWorkspaceExportRecovery(tx, input); } + await notifyQueuedRunDispatch(tx, { companyId: input.companyId }); const [coordinator] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: effect.cause, diff --git a/server/src/services/native-runtime/native-observed-workspace-preservation.ts b/server/src/services/native-runtime/native-observed-workspace-preservation.ts index fe82a3781c7..9fff5b347da 100644 --- a/server/src/services/native-runtime/native-observed-workspace-preservation.ts +++ b/server/src/services/native-runtime/native-observed-workspace-preservation.ts @@ -1,3 +1,4 @@ +import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js"; import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm"; import { companies, environmentLeases, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { remoteWorkspaceAuthoritySchema, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; @@ -206,6 +207,7 @@ export async function settleStoppedObservedWorkspace(db: Db, companyId: string, || !hasRemoteTerminationReceipt(lease) || !hasConfirmedSandboxStopAndRetain(lease)) return false; await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionFinalization); + await notifyQueuedRunDispatch(tx, { companyId }); const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null, recoveryState: "blocked", failureCode: "native_continuation_requires_reconciliation", updatedAt: new Date() }).where(and( diff --git a/server/src/services/native-runtime/native-startup-cleanup.ts b/server/src/services/native-runtime/native-startup-cleanup.ts index e6b80230c2d..be1980de4cf 100644 --- a/server/src/services/native-runtime/native-startup-cleanup.ts +++ b/server/src/services/native-runtime/native-startup-cleanup.ts @@ -1,3 +1,4 @@ +import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js"; import { and, eq, isNull, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, sandboxCreationOperations, type Db } from "@paperclipai/db"; import { conflict } from "../../errors.js"; @@ -202,6 +203,7 @@ export async function acknowledgeNativeStartupCleanup(db: Db, input: { companyId await assertStartupCreationsStopped(tx as unknown as Db, marker); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle); await notifyDeliveryWork(tx, DELIVERY_QUEUES.environmentCleanup); + await notifyQueuedRunDispatch(tx, { companyId: run.companyId }); await tx.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: null, failureCode: "native_retry_cancelled", updatedAt: new Date() }).where(and(eq(nativeRunFinalizations.runId, input.runId), isNull(nativeRunFinalizations.resultId))); }); diff --git a/server/src/services/queued-run-dispatch-liveness.test.ts b/server/src/services/queued-run-dispatch-liveness.test.ts new file mode 100644 index 00000000000..5a4dfe02788 --- /dev/null +++ b/server/src/services/queued-run-dispatch-liveness.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "vitest"; +import { + createQueuedRunRedriveSchedule, + QUEUED_RUN_REDRIVE_AFTER_MS, + QUEUED_RUN_REDRIVE_DELAY_MS, + QUEUED_RUN_REDRIVE_MAX_INTERVAL_MS, + type UndispatchedQueuedRun, +} from "./queued-run-dispatch-liveness.js"; + +const run = (runId: string, companyId: string, createdAt: number): UndispatchedQueuedRun => + ({ runId, companyId, agentId: `${runId}-agent`, issueId: null, createdAt: new Date(createdAt) }); + +describe("queued run re-drive schedule", () => { + it("waits out the bound, then re-drives with a doubling interval up to the cap", () => { + let now = 1_000_000; + const schedule = createQueuedRunRedriveSchedule(() => now); + const queued = [run("a", "company-a", now)]; + expect(schedule.refresh("all", queued)).toEqual({ companyIds: [], nextAt: now + QUEUED_RUN_REDRIVE_AFTER_MS }); + + const intervals: number[] = []; + now += QUEUED_RUN_REDRIVE_AFTER_MS; + for (let attempt = 0; attempt < 8; attempt++) { + expect(schedule.refresh("all", queued)).toEqual({ companyIds: ["company-a"], nextAt: now + QUEUED_RUN_REDRIVE_DELAY_MS }); + const waiting = schedule.refresh("all", queued); + expect(waiting.companyIds).toEqual([]); + intervals.push(waiting.nextAt! - now); + now = waiting.nextAt!; + } + expect(intervals.slice(0, 4)).toEqual([1, 2, 4, 8].map(factor => factor * QUEUED_RUN_REDRIVE_AFTER_MS)); + expect(Math.max(...intervals)).toBe(QUEUED_RUN_REDRIVE_MAX_INTERVAL_MS); + }); + + it("goes quiet once nothing is left undispatched and starts over for a run that returns", () => { + let now = 5_000_000; + const schedule = createQueuedRunRedriveSchedule(() => now); + const stale = run("stale", "company-b", now - 10 * QUEUED_RUN_REDRIVE_AFTER_MS); + expect(schedule.refresh("all", [stale])).toEqual({ companyIds: ["company-b"], nextAt: now + QUEUED_RUN_REDRIVE_DELAY_MS }); + expect(schedule.refresh("all", [])).toEqual({ companyIds: [], nextAt: null }); + now += 1; + expect(schedule.refresh("all", [stale])).toEqual({ companyIds: ["company-b"], nextAt: now + QUEUED_RUN_REDRIVE_DELAY_MS }); + }); + + it("refreshes only the companies a pass read and keeps every other company's runs due", () => { + const now = 9_000_000; + const schedule = createQueuedRunRedriveSchedule(() => now); + const held = run("held", "company-held", now - 30_000); + const stranded = run("stranded", "company-stranded", now - 30_000); + expect(schedule.refresh("all", [held, stranded]).nextAt).toBe(now + 30_000); + expect(schedule.refresh(new Set(["company-held"]), [])).toEqual({ companyIds: [], nextAt: now + 30_000 }); + expect(schedule.refresh(new Set(["company-stranded"]), [])).toEqual({ companyIds: [], nextAt: null }); + }); +}); diff --git a/server/src/services/queued-run-dispatch-liveness.ts b/server/src/services/queued-run-dispatch-liveness.ts new file mode 100644 index 00000000000..8e15d888ced --- /dev/null +++ b/server/src/services/queued-run-dispatch-liveness.ts @@ -0,0 +1,119 @@ +import { and, asc, eq, gte, notInArray, sql } from "drizzle-orm"; +import { agents, companies, heartbeatRuns, type Db } from "@paperclipai/db"; +import { instanceSettingsService, isTruthyRuntimeEnvValue, resolveWorktreeRunExecutionActivation } from "./instance-settings.js"; + +export const QUEUED_RUN_REDRIVE_AFTER_MS = 60_000; +export const QUEUED_RUN_REDRIVE_MAX_INTERVAL_MS = 15 * 60_000; +export const QUEUED_RUN_REDRIVE_DELAY_MS = 1_000; +export const QUEUED_RUN_NOT_DISPATCHED = "queued_run_not_dispatched"; + +const NOT_READY_AGENT_STATUSES = ["paused", "terminated", "pending_approval"]; +const UNDISPATCHED_PAGE_SIZE = 500; + +export type UndispatchedQueuedRun = { + runId: string; + companyId: string; + agentId: string; + issueId: string | null; + createdAt: Date; +}; + +type PageCursor = { at: string; runId: string }; + +async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefined, createdAfter: Date | null | undefined, + after: PageCursor | null, limit: number): + Promise> { + return db.select({ + runId: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + issueId: sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`, + createdAt: heartbeatRuns.createdAt, + cursorAt: sql`${heartbeatRuns.createdAt}::text`, + }).from(heartbeatRuns) + .innerJoin(companies, eq(companies.id, heartbeatRuns.companyId)) + .innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, heartbeatRuns.companyId))) + .where(and( + eq(heartbeatRuns.status, "queued"), + companyId ? eq(heartbeatRuns.companyId, companyId) : undefined, + createdAfter ? gte(heartbeatRuns.createdAt, createdAfter) : undefined, + eq(companies.status, "active"), + eq(agents.lifecycleState, "ready"), + notInArray(agents.status, NOT_READY_AGENT_STATUSES), + after ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${after.at}::timestamptz, ${after.runId}::uuid)` : undefined, + sql`not exists (select 1 from ${heartbeatRuns} live + where live.company_id = ${heartbeatRuns.companyId} and live.agent_id = ${heartbeatRuns.agentId} + and live.status = 'running')`, + )) + .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) + .limit(limit); +} + +/** Every undispatched queued run, read page by page with no cap; one + * company's runs when `companyId` is given. `createdAfter` is the worktree + * execution cutoff: dispatch never starts a run created before it. */ +export async function listUndispatchedQueuedRuns(db: Db, + options: { companyId?: string; createdAfter?: Date | null; pageSize?: number } = {}): + Promise { + const pageSize = options.pageSize ?? UNDISPATCHED_PAGE_SIZE; + const runs: UndispatchedQueuedRun[] = []; + let after: PageCursor | null = null; + for (;;) { + const page = await listUndispatchedQueuedRunPage(db, options.companyId, options.createdAfter, after, pageSize); + runs.push(...page.map(({ cursorAt: _cursorAt, ...run }) => run)); + if (page.length < pageSize) return runs; + const last = page[page.length - 1]!; + after = { at: last.cursorAt, runId: last.runId }; + } +} + +export type QueuedRunRedrivePlan = { companyIds: string[]; nextAt: number | null }; + +/** Re-drive deadlines for undispatched queued runs. `refresh` replaces the + * tracked runs of the companies a dispatch pass just read (or of every company + * after an instance-wide pass) and keeps the others, so a company-scoped pass + * never reads another company's queue. */ +export function createQueuedRunRedriveSchedule(now: () => number = Date.now) { + const tracked = new Map(); + return { + refresh(scope: "all" | ReadonlySet, runs: readonly UndispatchedQueuedRun[]): QueuedRunRedrivePlan { + const current = now(); + const live = new Set(runs.map(run => run.runId)); + for (const [runId, entry] of tracked) { + if ((scope === "all" || scope.has(entry.run.companyId)) && !live.has(runId)) tracked.delete(runId); + } + for (const run of runs) { + const prior = tracked.get(run.runId); + tracked.set(run.runId, prior ? { ...prior, run } + : { run, count: 0, dueAt: run.createdAt.getTime() + QUEUED_RUN_REDRIVE_AFTER_MS }); + } + const companyIds = new Set(); + let nextAt: number | null = null; + for (const [runId, entry] of tracked) { + let dueAt = entry.dueAt; + if (entry.dueAt <= current) { + companyIds.add(entry.run.companyId); + tracked.set(runId, { ...entry, count: entry.count + 1, + dueAt: current + Math.min(QUEUED_RUN_REDRIVE_AFTER_MS * 2 ** entry.count, QUEUED_RUN_REDRIVE_MAX_INTERVAL_MS) }); + dueAt = current + QUEUED_RUN_REDRIVE_DELAY_MS; + } + nextAt = nextAt === null ? dueAt : Math.min(nextAt, dueAt); + } + return { companyIds: [...companyIds], nextAt }; + }, + }; +} + +/** Whether this runtime dispatches queued runs at all, and from which + * creation time, read the way dispatch reads them: a worktree runtime starts + * nothing until worktree execution is armed, and then only runs created after + * its cutoff. */ +export async function readQueuedRunExecutionWindow(db: Db, runtimeEnv: Record = process.env): + Promise<{ dispatches: boolean; createdAfter: Date | null }> { + if (!isTruthyRuntimeEnvValue(runtimeEnv.PAPERCLIP_IN_WORKTREE)) return { dispatches: true, createdAfter: null }; + const activation = resolveWorktreeRunExecutionActivation(await instanceSettingsService(db).getExperimental(), + runtimeEnv.PAPERCLIP_INSTANCE_ID?.trim() || null); + const cutoff = activation.armed ? new Date(activation.cutoff) : null; + return cutoff && !Number.isNaN(cutoff.getTime()) ? { dispatches: true, createdAfter: cutoff } + : { dispatches: false, createdAfter: null }; +} diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 43778feafab..6b03087e3ac 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -11,6 +11,7 @@ import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; import { logger } from "../../middleware/logger.js"; import { getServerProcessStartedAt } from "../../server-info.js"; import { persistActivity } from "../activity-log.js"; +import { listUndispatchedQueuedRuns, QUEUED_RUN_NOT_DISPATCHED, readQueuedRunExecutionWindow } from "../queued-run-dispatch-liveness.js"; import { currentProcessProbe, proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js"; import { LEGACY_OWNER_REASONS, recordVersionAttestations } from "./owner-gone-attestation.js"; import { @@ -134,6 +135,8 @@ const TERMINAL_ISSUE_STATUSES = ["done", "cancelled"]; const ACTIVE_RUN_STATUSES = ["queued", "running", "scheduled_retry"]; /** An in-progress issue must show a next action within this time. */ const NEXT_ACTION_GRACE_MS = 5 * 60_000; +/** A queued run whose ready agent runs nothing should start within this time. */ +const QUEUED_RUN_DISPATCH_GRACE_MS = 5 * 60_000; /** Records read per kind (and per I3/I7 check) per pass; beyond this the * kind is reported truncated (a named coverage gap). */ const MAX_RECORDS_PER_KIND = 2_000; @@ -616,6 +619,27 @@ async function checkVisibleNextAction(db: Db, now: Date, limit: number): Promise return { violations, nextDueAt }; } +/** A queued run counts as an issue's next action only while dispatch can still + * start it. One that stays queued past the grace window while its ready agent + * runs nothing has lost its dispatch signal (invariant 7). */ +async function checkQueuedRunsDispatched(db: Db, now: Date, limit: number): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> { + const window = await readQueuedRunExecutionWindow(db); + if (!window.dispatches) return { violations: [], nextDueAt: null }; + const runs = await listUndispatchedQueuedRuns(db, { pageSize: limit, createdAfter: window.createdAfter }); + const cutoff = now.getTime() - QUEUED_RUN_DISPATCH_GRACE_MS; + const violations: InvariantViolation[] = []; + let nextDueAt: Date | null = null; + for (const run of runs) { + if (run.createdAt.getTime() < cutoff) { + violations.push({ rule: "I7_visible_next_action", kind: "queued_run", recordId: run.runId, companyId: run.companyId, + runId: run.runId, issueId: run.issueId, reason: QUEUED_RUN_NOT_DISPATCHED }); + } else { + nextDueAt ??= new Date(run.createdAt.getTime() + QUEUED_RUN_DISPATCH_GRACE_MS); + } + } + return { violations, nextDueAt }; +} + export async function runOwnershipReconciliationPass(db: Db, options: { state: OwnershipReportState; kinds?: readonly OwnableKind[]; @@ -730,6 +754,11 @@ export async function runOwnershipReconciliationPass(db: Db, options: { if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt; return result.violations; }, + async () => { + const result = await checkQueuedRunsDispatched(db, now, limit); + if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt; + return result.violations; + }, ]) { try { violations.push(...await check()); diff --git a/server/src/services/workspace-recovery-stop-proof.ts b/server/src/services/workspace-recovery-stop-proof.ts index 9f8ae3f206c..52c35a0287a 100644 --- a/server/src/services/workspace-recovery-stop-proof.ts +++ b/server/src/services/workspace-recovery-stop-proof.ts @@ -1,3 +1,4 @@ +import { notifyQueuedRunDispatch } from "./run-dispatch-notifications.js"; import { and, eq, inArray } from "drizzle-orm"; import { heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { logger } from "../middleware/logger.js"; @@ -47,6 +48,7 @@ export function sourceRunProvablyOver(run: typeof heartbeatRuns.$inferSelect, co export async function releaseLapsedCoordinatorOwner(tx: Pick, input: { companyId: string; runId: string }, coordinator: Coordinator | undefined, at: Date): Promise { if (!coordinator?.leaseOwner || !lapsedFinishedCoordinatorOwner(coordinator, at)) return false; + await notifyQueuedRunDispatch(tx, { companyId: input.companyId }); const cleared = await tx.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null, updatedAt: at }) .where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId), eq(nativeRunFinalizations.leaseOwner, coordinator.leaseOwner), diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 06e1ac55d2d..a76144c86f0 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -499,6 +499,7 @@ deletes a sandbox. An archived or stopped sandbox is recorded by its state alone | `w-20261111T202055-s9502` | CR1 75 s and CR2, OpenCode and Claude | No refusal. The Runner survived every outage. A killed provider failed the turn with a clear code: Claude `ACP agent reported a terminal connection failure`, OpenCode `opencode_server_exited`. | | `w-20261111T204144-s9503` | CR1 360 s twice, OpenCode and Claude | No refusal. The Runner and the provider survived both outages. | | `w-20261111T210055-s9504` | CR3 45 s, OpenCode and Claude | Not the staging code, a separate failure. `runnerd process supervision failed: RPC call "environmentExecute" timed out after 40000ms` failed the run as `notification_transport_failed`. The close path suspended a healthy Runner (`lifecycle: suspended`, no stderr). Retry worked. | +| `w-20261111T223714-s9512` | CR1 360 s, Codex W2, `--daytona-warm --ephemeral-server` (with #20909) | `R1:no_rest_within_budget`, U1 `no_owner_no_action` with the follow-up run queued 485 s. The next controller finalized the run that crossed the outage. Its continuation queued while that run's warm lease was still `active`, so the claim was declined. No dispatch pass ran after the lease was retained, and the run stayed `queued` for 19 minutes with no Retry. Lease release and native finalization writes now send a queued-run dispatch signal, and a declined claim logs its reason (`claimQueuedRun: queued run declined at the issue execution claim`). As a backstop, the run-dispatch worker re-drives an undispatched queued run 60 s after it was created, then with a doubling interval up to 15 minutes, and the ownership reconciler reports one still queued after 5 minutes as I7 `queued_run_not_dispatched` and re-drives it once. | ### Volume-free servers (`--ephemeral-server`)