mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
fix(heartbeat): re-drive a queued run whose dispatch signal was lost after a restart (#20986)
## 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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
bf18b348c5
commit
5a7378af56
23 files changed
+604
-23
No files matched your search
@@ -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",
|
||||
|
||||
|
||||
|
||||
@@ -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';
|
||||
+94
-8
@@ -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",
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
@@ -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();
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string | undefined> = [];
|
||||
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<string | undefined> = [];
|
||||
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);
|
||||
});
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>,
|
||||
@@ -9,6 +10,7 @@ export function registerRunDispatchWork(
|
||||
promoteDueScheduledRetries(): Promise<unknown>;
|
||||
nextScheduledRetryAt(): Promise<number | null>;
|
||||
resumeQueuedRunsAfterDrain(scope?: RunDispatchScope): Promise<unknown>;
|
||||
listUndispatchedQueuedRuns?(scope?: RunDispatchScope): Promise<UndispatchedQueuedRun[]>;
|
||||
schedulingSuppressed(): Promise<boolean>;
|
||||
},
|
||||
canRun: () => boolean,
|
||||
@@ -26,6 +28,8 @@ export function registerRunDispatchWork(
|
||||
next: () => suppressed ? Promise.resolve(null) : owner.nextScheduledRetryAt(),
|
||||
});
|
||||
const scopes = new Set<string>(["*"]);
|
||||
const redrive = createQueuedRunRedriveSchedule();
|
||||
let dispatched: "all" | Set<string> = 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,
|
||||
|
||||
@@ -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 };
|
||||
|
||||
@@ -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 }));
|
||||
}
|
||||
@@ -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 };
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)));
|
||||
});
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
});
|
||||
@@ -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<Array<UndispatchedQueuedRun & { cursorAt: string }>> {
|
||||
return db.select({
|
||||
runId: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
issueId: sql<string | null>`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
cursorAt: sql<string>`${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<UndispatchedQueuedRun[]> {
|
||||
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<string, { run: UndispatchedQueuedRun; count: number; dueAt: number }>();
|
||||
return {
|
||||
refresh(scope: "all" | ReadonlySet<string>, 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<string>();
|
||||
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<string, string | undefined> = 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 };
|
||||
}
|
||||
@@ -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());
|
||||
|
||||
@@ -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<Db, "update">, input: { companyId: string; runId: string },
|
||||
coordinator: Coordinator | undefined, at: Date): Promise<boolean> {
|
||||
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),
|
||||
|
||||
@@ -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`)
|
||||
|
||||
|
||||
Reference in new issue
Block a user