From 46f2c0bb9a999b414497ba901b64a7c750c7fbc0 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:25:24 -0500 Subject: [PATCH] fix(server): bound and scope routine projection recovery Co-Authored-By: Paperclip --- doc/plans/2026-10-30-routine-status-sleep.md | 13 +++-- .../__tests__/execution-control-work.test.ts | 4 +- ...tbeat-workspace-branch-containment.test.ts | 20 +++---- .../routine-execution-status-work.test.ts | 52 +++++++++++++++++++ .../routine-execution-status-worker.test.ts | 37 ++++++++++--- .../routine-status-reconciliation.test.ts | 17 +++++- .../server-startup-feedback-export.test.ts | 27 ++++++++++ .../execution-control-notifications.ts | 9 ++-- .../execution-control-reconciliation.ts | 2 +- .../services/execution-recovery-resolution.ts | 6 ++- .../services/explicit-native-continuation.ts | 2 +- server/src/services/heartbeat.ts | 2 +- server/src/services/heartbeat/queue.ts | 2 + .../src/services/heartbeat/run-lifecycle.ts | 4 +- .../services/heartbeat/runtime-execution.ts | 4 +- server/src/services/issue-recovery-actions.ts | 10 ++-- .../issue-review-state-notifications.ts | 2 +- server/src/services/issues.ts | 2 +- .../src/services/legacy-execution-recovery.ts | 2 +- .../native-failed-workspace-recovery.ts | 2 +- .../native-runtime/native-restart-recovery.ts | 2 +- .../native-runtime/native-run-finalizer.ts | 4 +- .../native-runtime/native-run-projection.ts | 2 +- .../native-runtime/native-safe-replacement.ts | 4 +- .../native-runtime/native-session-executor.ts | 6 +-- .../native-runtime/native-superseded-retry.ts | 2 +- .../native-workspace-reconstruction.test.ts | 8 ++- .../native-workspace-reconstruction.ts | 2 + .../routine-execution-status-notifications.ts | 7 ++- .../services/routine-execution-status-work.ts | 17 ++++-- server/src/services/routine-run-status.ts | 29 +++++------ server/src/services/routines.ts | 5 +- server/src/services/system-recovery.ts | 2 +- server/src/services/task-retry.ts | 2 +- 34 files changed, 227 insertions(+), 84 deletions(-) diff --git a/doc/plans/2026-10-30-routine-status-sleep.md b/doc/plans/2026-10-30-routine-status-sleep.md index ea6f9364999..271d0b544a7 100644 --- a/doc/plans/2026-10-30-routine-status-sleep.md +++ b/doc/plans/2026-10-30-routine-status-sleep.md @@ -7,14 +7,16 @@ scanned every five minutes, even with no routines or only completed history. ## Ownership and trigger `routine-run-status.ts` owns the projection, custody checks and repair transaction. -The worker loads actual mismatches at startup. It accepts issue hints from issue +The worker loads actual mismatches in keyset pages of 50 at startup. It accepts issue hints from issue creation, status/origin changes, checkout/release, routine issue association, execution control writes and review state writes. It uses the existing transaction-aware work notification protocol. An outer transaction owns the hint; a savepoint release does not wake the repair worker before commit. -A standard execution must name the same routine, invocation, company and linked -task. Plugin-managed executions may use their declared custom origin. Repair +An execution must retain reciprocal task/invocation links in the same company +and belong to a routine in that company. This is the existing routine status +owner contract. Origin labels can change when a plugin declaration changes or +is removed; retained reciprocal links preserve historical execution custody. Repair locks the routine and task and rechecks the linked invocation before writing. It preserves skipped/coalesced history and unrelated dispatch failures. A task that resumes clears an execution-task transient failure through the same owner @@ -33,8 +35,9 @@ history leave no recurring worker SQL. The existing Cloud and heartbeat schedule enablement gates remain in place. Ordinary routine scheduling is unchanged. Supported app writers register `notifyRoutineExecutionStatus(tx, issueId)` before -their mutation. Common execution writers without an issue hint request debt -reconstruction. Backups and independent SQL import/repair tools must restore +their mutation. Common execution writers without an issue hint request debt reconstruction only +for their recorded company. Projection debt settles even when that company is +paused; company activity does not create or cancel this obligation. Backups and independent SQL import/repair tools must restore before startup or call that same transaction contract; process-local hints do not observe independent SQL or another process. No permanent listener or periodic configured-routine scan is added. diff --git a/server/src/__tests__/execution-control-work.test.ts b/server/src/__tests__/execution-control-work.test.ts index 0046d8574aa..14cd9456eb5 100644 --- a/server/src/__tests__/execution-control-work.test.ts +++ b/server/src/__tests__/execution-control-work.test.ts @@ -383,11 +383,11 @@ describe.skipIf(!support.supported)("execution-control durable work", () => { const due = new Date(Date.now() + 120_000), earlier = new Date(Date.now() + 60_000); for (const at of [due, earlier]) { await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns) - .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id))); + .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)), { companyId }); await settle(); expect(c.nextWakeAt()).toBe(at.getTime()); } await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns) - .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id))); + .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)), { companyId }); await quiet(c); }); it("delivers a reconciled continuation when marking and resolving commit together", async () => { diff --git a/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts b/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts index 52d434a3078..484cf55e8ff 100644 --- a/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts +++ b/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts @@ -1,3 +1,5 @@ +import { subscribeDatabaseWork } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { execFile } from "node:child_process"; import { createHash, randomUUID } from "node:crypto"; @@ -888,7 +890,7 @@ describeEmbeddedPostgres("heartbeat workspace branch containment", () => { const tempRoots: string[] = []; beforeAll(async () => { - if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl); + if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl, { maxConnections: 6 }); else { tempDb = await startEmbeddedPostgresTestDatabase("paperclip-branch-containment-"); db = createDb(tempDb.connectionString); @@ -1173,15 +1175,15 @@ describeEmbeddedPostgres("heartbeat workspace branch containment", () => { }, }); + const routineEvents: string[] = []; + const unsubscribeRoutine = subscribeDatabaseWork(db, DELIVERY_QUEUES.routineExecutionStatus, event => routineEvents.push(event)); const heartbeat = heartbeatService(db); - const run = await heartbeat.wakeup(agentId, { - source: "assignment", - triggerDetail: "system", - reason: "issue_assigned", - payload: { issueId }, - contextSnapshot: { issueId, wakeReason: "issue_assigned" }, - }); - + let run; + try { + run = await heartbeat.wakeup(agentId, { source: "assignment", triggerDetail: "system", reason: "issue_assigned", + payload: { issueId }, contextSnapshot: { issueId, wakeReason: "issue_assigned" } }); + } finally { unsubscribeRoutine(); } + expect(routineEvents).toContain("intent"); expect(routineEvents).toContain("settled"); expect(run).toBeNull(); expect(adapterExecute).not.toHaveBeenCalled(); diff --git a/server/src/__tests__/routine-execution-status-work.test.ts b/server/src/__tests__/routine-execution-status-work.test.ts index 7415b63bfd5..53d7e5c0225 100644 --- a/server/src/__tests__/routine-execution-status-work.test.ts +++ b/server/src/__tests__/routine-execution-status-work.test.ts @@ -1,3 +1,5 @@ +import { writeExecutionControl } from "../services/execution-control-notifications.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -122,6 +124,56 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await settled(); await quiet(); expect(await saved(run.id)).toEqual(before); expect((await saved(otherRun.id)).status).toBe("issue_created"); expect(errors).toEqual([]); }); + it.each(["changed", "removed"])("keeps historical custom-origin custody after its plugin declaration is %s", async mode => { + const f = await fixture(), pluginKey = `routine-history-${randomUUID()}`, originId = `operation:${randomUUID()}`; + const [plugin] = await db.insert(plugins).values({ pluginKey, packageName: pluginKey, version: "0.1.0", + manifestJson: { id: pluginKey, apiVersion: 1, version: "0.1.0", displayName: "Routine test", description: "Routine test", + author: "Paperclip", categories: ["automation"], capabilities: ["routines.managed"], entrypoints: { worker: "./worker.js" } } }).returning(); + const [resource] = await db.insert(pluginManagedResources).values({ companyId: f.companyId, pluginId: plugin.id, pluginKey, + resourceKind: "routine", resourceKey: "history", resourceId: f.routine.id, defaultsJson: { issueTemplate: { originId } } }).returning(); + const run = await dispatch(f); + expect((await db.select().from(issues).where(eq(issues.id, run.linkedIssueId!)))[0].originId).toBe(originId); + if (mode === "changed") await db.update(pluginManagedResources).set({ defaultsJson: { issueTemplate: { originId: "new-origin" } } }).where(eq(pluginManagedResources.id, resource.id)); + else await db.delete(pluginManagedResources).where(eq(pluginManagedResources.id, resource.id)); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, run.linkedIssueId!)); + await start(); await projected(run.id, "completed"); await quiet(); expect(errors).toEqual([]); + }); + it("reconstructs more than one page and does not lose a committed issue hint behind the startup cursor", async () => { + const f = await fixture(), ids = Array.from({ length: 53 }, () => randomUUID()).sort(); + const issueIds = ids.map(() => randomUUID()); + await db.insert(issues).values(ids.map((id, index) => ({ id: issueIds[index], companyId: f.companyId, title: "Legacy execution", + status: index === 0 ? "todo" : "done", originKind: "routine_execution", originRunId: id, originId: f.routine.id }))); + await db.insert(routineRuns).values(ids.map((id, index) => ({ id, companyId: f.companyId, routineId: f.routine.id, + source: "schedule", linkedIssueId: issueIds[index], status: "issue_created" }))); + const service = routineExecutionStatusService(db), load = service.load; + let entered!: () => void, release!: () => void; + const enteredGate = new Promise(resolve => { entered = resolve; }); + const releaseGate = new Promise(resolve => { release = resolve; }); + const loaded = vi.spyOn(service, "load").mockImplementationOnce(async (scope, afterId) => { + const page = await load(scope, afterId); expect(page).toHaveLength(50); entered(); await releaseGate; return page; + }); + const starting = start(service); await enteredGate; + await issueService(db).update(issueIds[0]!, { status: "done" }); release(); await starting; + await vi.advanceTimersByTimeAsync(3000); + await vi.waitFor(async () => expect((await db.select().from(routineRuns)).every(row => row.status === "completed")).toBe(true)); + expect(loaded.mock.calls.some(([scope, afterId]) => scope === undefined && afterId !== undefined)).toBe(true); + expect(loaded.mock.calls.some(([scope]) => scope?.issueId === issueIds[0])).toBe(true); + await quiet(); expect(errors).toEqual([]); + }); + it("scopes shared execution writes to their company and settles paused-company projection debt", async () => { + const f = await fixture(), other = await fixture(), run = await dispatch(f), otherRun = await dispatch(other); + const service = routineExecutionStatusService(db), load = vi.spyOn(service, "load"); + await start(service); await quiet(); load.mockClear(); + await db.update(companies).set({ status: "paused" }).where(eq(companies.id, f.companyId)); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, otherRun.linkedIssueId!)); + await writeExecutionControl(db, DELIVERY_QUEUES.executionStatus, + tx => tx.update(issues).set({ status: "done" }).where(eq(issues.id, run.linkedIssueId!)), { companyId: f.companyId }); + await projected(run.id, "completed"); + expect(load.mock.calls.every(([scope]) => scope?.companyId === f.companyId)).toBe(true); + expect((await saved(otherRun.id)).status).toBe("issue_created"); await quiet(); + await db.transaction(async tx => { await notifyRoutineExecutionStatus(tx, otherRun.linkedIssueId!); }); + await projected(otherRun.id, "completed"); await quiet(); expect(errors).toEqual([]); + }); it("waits for the real outer transaction and forgets a rolled-back nested status write", async () => { const f = await fixture(), run = await dispatch(f); await start(); await quiet(); let release!: () => void, entered!: () => void; diff --git a/server/src/__tests__/routine-execution-status-worker.test.ts b/server/src/__tests__/routine-execution-status-worker.test.ts index b88babbbd98..b62245ce3e2 100644 --- a/server/src/__tests__/routine-execution-status-worker.test.ts +++ b/server/src/__tests__/routine-execution-status-worker.test.ts @@ -1,3 +1,4 @@ +import type { RoutineExecutionStatusScope } from "../services/routine-execution-status-notifications.js"; import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { afterEach, beforeEach, expect, it, vi } from "vitest"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -18,19 +19,19 @@ afterEach(async () => { await coordinator.stop(); vi.useRealTimers(); }); it("loads actual debt once and leaves no recurring owner work", async () => { const service = { load: vi.fn(async () => []), settle: vi.fn(async () => true) }; await registerRoutineExecutionStatusWork(coordinator, service, () => canRun).ready; - expect(service.load).toHaveBeenCalledExactlyOnceWith(undefined); + expect(service.load).toHaveBeenCalledExactlyOnceWith(undefined, undefined); expect(coordinator.nextWakeAt()).toBeNull(); await vi.advanceTimersByTimeAsync(20 * 60_000); expect(service.load).toHaveBeenCalledOnce(); expect(service.settle).not.toHaveBeenCalled(); expect(errors).toEqual([]); }); it("reconstructs only an invalidated issue and coalesces duplicate requests", async () => { - const service = { load: vi.fn(async (_issueId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) }; + const service = { load: vi.fn(async (_scope?: RoutineExecutionStatusScope, _afterId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) }; const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready; service.load.mockResolvedValueOnce([entry]); - worker.invalidate("issue"); worker.invalidate("issue"); + worker.invalidate(JSON.stringify({ issueId: "issue" })); worker.invalidate(JSON.stringify({ issueId: "issue" })); await vi.advanceTimersByTimeAsync(1000); - expect(service.load).toHaveBeenLastCalledWith("issue"); expect(service.load).toHaveBeenCalledTimes(2); + expect(service.load).toHaveBeenLastCalledWith({ issueId: "issue" }, undefined); expect(service.load).toHaveBeenCalledTimes(2); expect(service.settle).toHaveBeenCalledExactlyOnceWith(expect.objectContaining(entry)); expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); }); @@ -50,7 +51,7 @@ it("retains startup and first work during drain without passive owner calls", as canRun = false; const service = { load: vi.fn(async () => [entry]), settle: vi.fn(async () => true) }; const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready; - worker.invalidate("issue"); + worker.invalidate(JSON.stringify({ issueId: "issue" })); await vi.advanceTimersByTimeAsync(20 * 60_000); expect(service.load).not.toHaveBeenCalled(); expect(service.settle).not.toHaveBeenCalled(); canRun = true; await vi.advanceTimersByTimeAsync(30_000); @@ -69,11 +70,11 @@ it("keeps a failed owner across drain and retries after resume", async () => { it("retains a notification that arrives while the startup reconstruction is in flight", async () => { let release!: () => void; const gate = new Promise(resolve => { release = resolve; }); - const service = { load: vi.fn(async (_issueId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) }; + const service = { load: vi.fn(async (_scope?: RoutineExecutionStatusScope, _afterId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) }; service.load.mockImplementationOnce(async () => { await gate; return []; }); const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await vi.waitFor(() => expect(service.load).toHaveBeenCalledOnce()); - worker.invalidate("issue"); service.load.mockResolvedValueOnce([entry]); release(); await worker.ready; + worker.invalidate(JSON.stringify({ issueId: "issue" })); service.load.mockResolvedValueOnce([entry]); release(); await worker.ready; await vi.advanceTimersByTimeAsync(1000); expect(service.settle).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); }); @@ -81,6 +82,26 @@ it("retains a notification that arrives while the startup reconstruction is in f it("cancels retries and subscriptions at shutdown", async () => { const service = { load: vi.fn(async () => [entry]), settle: vi.fn(async () => true).mockRejectedValue(new Error("offline")) }; const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready; - await worker.stop(); worker.invalidate("issue"); await vi.advanceTimersByTimeAsync(20 * 60_000); + await worker.stop(); worker.invalidate(JSON.stringify({ issueId: "issue" })); await vi.advanceTimersByTimeAsync(20 * 60_000); expect(service.settle).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull(); }); + + +it("reconstructs a backlog in bounded keyset pages while retaining a scoped hint behind the cursor", async () => { + const firstPage = Array.from({ length: 50 }, (_, i) => ({ ...entry, id: `run-${String(i).padStart(3, "0")}` })); + const late = { ...entry, id: "run-late", issueId: "late-issue" }; + const service = { + load: vi.fn(async (scope?: RoutineExecutionStatusScope, afterId?: string) => scope?.issueId ? [late] + : afterId ? [{ ...entry, id: "run-050" }] : firstPage), + settle: vi.fn(async () => true), + }; + const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready; + expect(service.load).toHaveBeenCalledExactlyOnceWith(undefined, undefined); + worker.invalidate(JSON.stringify({ issueId: "late-issue" })); + await vi.advanceTimersByTimeAsync(3000); + expect(service.load).toHaveBeenCalledWith(undefined, "run-049"); + expect(service.load).toHaveBeenCalledWith({ issueId: "late-issue" }, undefined); + expect(service.settle).toHaveBeenCalledTimes(52); + expect(service.settle).toHaveBeenCalledWith(expect.objectContaining(late)); + expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); +}); diff --git a/server/src/__tests__/routine-status-reconciliation.test.ts b/server/src/__tests__/routine-status-reconciliation.test.ts index 964e43d6035..343c6fb38b6 100644 --- a/server/src/__tests__/routine-status-reconciliation.test.ts +++ b/server/src/__tests__/routine-status-reconciliation.test.ts @@ -38,11 +38,10 @@ if (!support.supported) console.warn(`Skipping routine status database tests: ${ expect((await db.select().from(routineRuns))[0]).toMatchObject({ status: "issue_created", completedAt: null }); } }); - it.each(["active_issue", "wrong_origin", "wrong_routine", "wrong_run", "wrong_company", "wrong_routine_company", "different_status"])("does not settle an unsupported handoff: %s", async mode => { + it.each(["active_issue", "wrong_origin", "wrong_run", "wrong_company", "wrong_routine_company", "different_status"])("does not settle an unsupported handoff: %s", async mode => { const f = await seed(); if (mode === "active_issue") await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id)); if (mode === "wrong_origin") await db.update(issues).set({ originKind: "manual" }).where(eq(issues.id, f.issue.id)); - if (mode === "wrong_routine") await db.update(issues).set({ originId: randomUUID() }).where(eq(issues.id, f.issue.id)); if (mode === "wrong_run") await db.update(issues).set({ originRunId: randomUUID() }).where(eq(issues.id, f.issue.id)); if (mode === "wrong_company") { const other = await seed("done"); await db.update(routineRuns).set({ companyId: other.companyId }).where(eq(routineRuns.id, f.run.id)); } if (mode === "wrong_routine_company") { const other = await seed("done"); @@ -55,6 +54,20 @@ if (!support.supported) console.warn(`Skipping routine status database tests: ${ expect((await db.select().from(routineRuns)).find(row => row.id === f.run.id)).toEqual(before.find(row => row.id === f.run.id)); if (mode === "wrong_routine_company") await expect(routineService(db).syncRunStatusForIssue(f.issue.id)).resolves.toBeNull(); }); + it("uses reciprocal durable custody when a retained origin no longer names the current routine declaration", async () => { + const f = await seed("done"); + await db.update(issues).set({ originId: "previous-declaration-origin" }).where(eq(issues.id, f.issue.id)); + expect(await routineService(db).reconcileExecutionRunStatuses()).toEqual({ scanned: 1, settled: 1, deferred: 0 }); + expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.run.id)))[0].status).toBe("completed"); + }); + it.each(["unknown-status", "other-code"])("preserves a failed invocation with an unrelated transient marker: %s", async mode => { + const f = await seed("in_progress"); + await db.update(routineRuns).set({ status: "failed", failureReason: null, + triggerPayload: { transientFailure: { code: mode === "other-code" ? "unrelated_failure" : "execution_issue_status", + status: mode === "unknown-status" ? "unrecognized" : "blocked" } } }).where(eq(routineRuns.id, f.run.id)); + expect(await routineService(db).reconcileExecutionRunStatuses()).toEqual({ scanned: 0, settled: 0, deferred: 0 }); + expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.run.id)))[0].status).toBe("failed"); + }); it.each(["resume", "retarget", "routine_transfer"])("rechecks %s after candidate selection and before repair", async mode => { const f = await seed(); const transaction = db.transaction.bind(db); vi.spyOn(db, "transaction").mockImplementationOnce(async callback => { diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bbb020c865d..caf5565762a 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,3 +1,4 @@ +import { routineExecutionStatusService } from "../services/routine-run-status.js"; import { nativeHarnessBackupExpirationService } from "../services/native-runtime/native-harness-backup-expiration.js"; import { skillVersionRetentionService } from "../services/skill-version-retention.js"; import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js"; @@ -392,6 +393,11 @@ vi.mock("../services/approval-retention.js", async importOriginal => ({ approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), })); +vi.mock("../services/routine-run-status.js", async importOriginal => ({ + ...await importOriginal(), + routineExecutionStatusService: vi.fn(() => ({ load: vi.fn(async () => []), settle: vi.fn(async () => false) })), +})); + vi.mock("../services/index.js", () => ({ backfillLegacyToolOAuthTokens: vi.fn(async () => ({ scannedConnections: 0, @@ -1366,6 +1372,27 @@ describe("startServer feedback export wiring", () => { } }); + it.each([{ cloud: false, enabled: true }, { cloud: true, enabled: false }, { cloud: true, enabled: true }])( + "registers routine status only for enabled Cloud scheduling ($cloud/$enabled)", async ({ cloud, enabled }) => { + if (cloud) vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "startup-test-token"); + else vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", undefined); + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + const intervalCallbacks: Array<() => void> = []; + const intervals = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void) => { + intervalCallbacks.push(callback); return 1 as unknown as ReturnType; + }) as typeof setInterval); + try { + await startServer(); await settleStartupHousekeeping(); + expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.routineExecutionStatus)).toHaveLength(cloud && enabled ? 1 : 0); + const service = vi.mocked(routineExecutionStatusService).mock.results.at(-1)?.value; + if (cloud && enabled) expect(service?.load).toHaveBeenCalledExactlyOnceWith(undefined, undefined); + for (let index = 0; index < 4; index++) for (const callback of intervalCallbacks) callback(); + await settleStartupHousekeeping(); + if (cloud && enabled) expect(service?.load).toHaveBeenCalledOnce(); + } finally { intervals.mockRestore(); vi.unstubAllEnvs(); } + }, + ); + it("keeps routine ticks and setup cleanup active when heartbeat scheduling is suppressed", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true, diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 33a99774a48..fcbb892578c 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -2,8 +2,10 @@ import { notifyRoutineExecutionStatus } from "./routine-execution-status-notific import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js"; -export async function notifyExecutionRunWork(transaction: object, issueId?: string) { - await notifyRoutineExecutionStatus(transaction, issueId); +export type ExecutionWorkScope = { companyId: string; agentId?: string; runId?: string; issueId?: string }; + +export async function notifyExecutionRunWork(transaction: object, scope?: ExecutionWorkScope) { + if (scope) await notifyRoutineExecutionStatus(transaction, scope.issueId, scope.companyId); for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { await notifyDeliveryWork(transaction, queue); @@ -14,9 +16,10 @@ export function writeExecutionControl( db: Db | Parameters[0]>[0], queue: DeliveryQueue | readonly DeliveryQueue[], write: (tx: Db) => PromiseLike, + scope: ExecutionWorkScope | undefined, ): Promise { return db.transaction(async transaction => { - await notifyRoutineExecutionStatus(transaction); + if (scope) await notifyRoutineExecutionStatus(transaction, scope.issueId, scope.companyId); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors); for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); return write(transaction as unknown as Db); diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 2f2a435982d..621fd210cfc 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -196,7 +196,7 @@ export async function reconcileAbandonedExecutionControl( await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index f3f5a3e4310..a6f80f26ba0 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -1,3 +1,4 @@ +import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; @@ -546,7 +547,7 @@ export async function settleUnrecoverableExecutions( companyId: run.companyId, issueId: task.id, runId: run.id, }).limit(1).for("update"); if (!hold) return; - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); const note = "Exact session recovery stopped. The recorded workspace recovery holds this execution without replaying its actions."; await tx.update(issueRecoveryActions).set({ status: "cancelled", outcome: "cancelled", resolvedAt: now, updatedAt: now, wakePolicy: null, monitorPolicy: null, @@ -596,6 +597,7 @@ export async function settleUnrecoverableExecutions( if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, task.id, task.companyId); const [projected] = await tx .update(issues) .set({ @@ -654,7 +656,7 @@ export async function settleUnrecoverableExecutions( replay: "not_authorized", }, }); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); await tx .update(heartbeatRuns) .set({ executionStatusDeliveryId: randomUUID() }) diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 074ffc3bdf5..084a25e457f 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -475,7 +475,7 @@ export async function admitExplicitNativeContinuation(input: { }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.queuedCommentInterruptId), eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"))); if (cancelledStartupIds.size) await db.transaction(async tx => { - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId, agentId }); for (const runId of cancelledStartupIds) { await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index eb655fdef25..a4cec7b81b0 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -3724,7 +3724,7 @@ export function heartbeatService( await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx .update(heartbeatRuns) .set({ executionControlDeadlineAt: null }) - .where(eq(heartbeatRuns.id, run.id))); + .where(eq(heartbeatRuns.id, run.id)), { companyId: run.companyId }); } nativeOwnershipHeld = nativeOwnershipHeld || diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index a41b486b647..d2786778cf7 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyRoutineExecutionStatus } from "../routine-execution-status-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -3919,6 +3920,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id }); + await notifyRoutineExecutionStatus(tx, issue.id, issue.companyId); await tx .update(issues) .set({ diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index f796da03a5e..ac4e800f279 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1044,8 +1044,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl if (previous && isHeartbeatRunTerminalStatus(status)) await notifyRetainedNativeSessionCleanup(tx, { companyId: previous.companyId, runId: previous.id }); if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId); - await notifyExecutionRunWork(tx, previous?.nativeIssueId ?? previous?.issueId - ?? readNonEmptyString(parseObject(previous?.contextSnapshot).issueId) ?? undefined); + await notifyExecutionRunWork(tx, previous ? { companyId: previous.companyId, agentId: previous.agentId, runId: previous.id, + issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined } : undefined); if (previous) await notifyChatRunMilestones(tx, { companyId: previous.companyId, runId: previous.id, issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined }); return write(tx); diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index b5b64abc948..bdbbfa07752 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -966,7 +966,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running"), ), - )); + ), { companyId: run.companyId }); const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded"; await recordWorkspaceFinalize(workspaceFinalizeStatus); if (adapterResult.nativeFinalization) { @@ -1042,7 +1042,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running"), ), - )); + ), { companyId: run.companyId }); if ( issueRef && context.resumeSessionGoalHeartbeat === true && diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 5fb28bcb6d8..48fb001c0b4 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -392,7 +392,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now }) .where(and(eq(issueRecoveryActions.id, existing.id), inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]), - sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning()); + sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning(), { companyId: input.companyId }); if (updated) return toReadModel(updated); } return existing; @@ -467,7 +467,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ), ) .returning(); - }); + }, { companyId: input.companyId }); if (!exhausted) { return retryUpsertSourceScoped(input, retryCount); } @@ -539,7 +539,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ), ) .returning(); - }); + }, { companyId: input.companyId }); if (!updated) { return retryUpsertSourceScoped(input, retryCount); } @@ -554,7 +554,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { .insert(issueRecoveryActions) .values(buildInsertValues(input, ownerType, now)) .returning(); - }); + }, { companyId: input.companyId }); return toReadModel(created!); } catch (error) { if (!isUniqueRecoveryActionConflict(error)) throw error; @@ -608,7 +608,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { }) .where(and(...predicates)) .returning(); - }); + }, { companyId: input.companyId }); return updated ? toReadModel(updated) : null; } diff --git a/server/src/services/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index 15bc2a24287..9ef78e0903d 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -5,7 +5,7 @@ import { notifyNativeReviewResponse } from "./native-review-response-notificatio import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) { - await notifyRoutineExecutionStatus(tx, scope.issueId); + await notifyRoutineExecutionStatus(tx, scope.issueId, scope.companyId); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyExternalConversationSettlement(tx, scope); await notifyNativeReviewResponse(tx, scope); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index eeebf326267..ece99b16752 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -10607,7 +10607,7 @@ export function issueService(db: Db) { } if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); - if (values.originKind === "routine_execution") await notifyRoutineExecutionStatus(tx, values.id); + if (values.originKind === "routine_execution") await notifyRoutineExecutionStatus(tx, values.id, companyId); const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) { diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 427704de63f..c462cdf1fca 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -135,7 +135,7 @@ export async function terminalizeLegacyExecution(input: { }; if (issueId) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: current.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id }); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: current.companyId }); let [updated] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index 5c3268f6afc..a1c6c914b33 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -65,7 +65,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick { const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RECONSTRUCTION_TEST_DATABASE_URL"); if (external) { await applyPendingMigrations(external); - db = createDb(external); + db = createDb(external, { maxConnections: 6 }); return; } temporary = await startEmbeddedPostgresTestDatabase("workspace-reconstruction-"); @@ -144,17 +144,21 @@ describe("reconstruction after confirmed native workspace loss", () => { expect(await queue.nextTimerHeartbeatAt(agent)).toBeNull(); const events: string[] = []; const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event)); + const routineEvents: string[] = []; + const unsubscribeRoutine = subscribeDatabaseWork(db, DELIVERY_QUEUES.routineExecutionStatus, event => routineEvents.push(event)); try { await db.transaction(async tx => { await reconstructNativeWorkspace({ ...f, db: tx as unknown as Db }); expect(events).toContain("intent"); expect(events).not.toContain("settled"); + expect(routineEvents).toContain("intent"); expect(routineEvents).not.toContain("settled"); expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); }); expect(events).toContain("settled"); + expect(routineEvents).toContain("settled"); expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("todo"); expect(await queue.nextTimerHeartbeatAt(agent)).toBe(agent.createdAt.getTime() + 600_000); - } finally { unsubscribe(); } + } finally { unsubscribe(); unsubscribeRoutine(); } }); it("pins a historical cancelled workspace through public heartbeat startup without replaying its turn", async () => { diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.ts b/server/src/services/native-runtime/native-workspace-reconstruction.ts index 5273b30216a..34e8b25d10a 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.ts @@ -1,3 +1,4 @@ +import { notifyRoutineExecutionStatus } from "../routine-execution-status-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { and, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm"; @@ -187,6 +188,7 @@ export async function reconstructNativeWorkspace(input: { }).where(eq(issueRecoveryActions.id, source.action.id)); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, input.issueId, input.companyId); await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null, blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id)); const publication = await persistActivity(db, { companyId: input.companyId, actorType: "user", actorId: input.actorId, diff --git a/server/src/services/routine-execution-status-notifications.ts b/server/src/services/routine-execution-status-notifications.ts index cdd2be826c0..fb9df28dda9 100644 --- a/server/src/services/routine-execution-status-notifications.ts +++ b/server/src/services/routine-execution-status-notifications.ts @@ -1,5 +1,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export function notifyRoutineExecutionStatus(tx: object, issueId?: string) { - return notifyDeliveryWork(tx, DELIVERY_QUEUES.routineExecutionStatus, issueId ?? "*"); +export type RoutineExecutionStatusScope = { issueId?: string; companyId?: string }; + +export function notifyRoutineExecutionStatus(tx: object, issueId?: string, companyId?: string) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.routineExecutionStatus, + issueId || companyId ? JSON.stringify({ issueId, companyId } satisfies RoutineExecutionStatusScope) : "*"); } diff --git a/server/src/services/routine-execution-status-work.ts b/server/src/services/routine-execution-status-work.ts index 81c8e566a7c..0a4437f89f2 100644 --- a/server/src/services/routine-execution-status-work.ts +++ b/server/src/services/routine-execution-status-work.ts @@ -1,18 +1,27 @@ import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; -import type { routineExecutionStatusService } from "./routine-run-status.js"; +import type { RoutineExecutionStatusScope } from "./routine-execution-status-notifications.js"; +import { ROUTINE_EXECUTION_STATUS_PAGE_SIZE, type routineExecutionStatusService } from "./routine-run-status.js"; export function registerRoutineExecutionStatusWork( coordinator: ReturnType, service: ReturnType, canRun: () => boolean, ) { + const cursors = new Map(); return registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.routineExecutionStatus, - decodeScope: hint => hint === "*" ? undefined : hint, - matches: (entry, issueId?: string) => issueId === undefined || entry.issueId === issueId, - load: async (issueId?: string) => (await service.load(issueId)).map(row => ({ ...row, dueAt: Date.now() })), + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RoutineExecutionStatusScope, + matches: (entry, scope?: RoutineExecutionStatusScope) => (!scope?.issueId || entry.issueId === scope.issueId) + && (!scope?.companyId || entry.companyId === scope.companyId), + load: async (scope?: RoutineExecutionStatusScope) => { + const key = JSON.stringify(scope) ?? "*"; + const rows = await service.load(scope, cursors.get(key)); + const complete = rows.length < ROUTINE_EXECUTION_STATUS_PAGE_SIZE; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete }; + }, check: async entry => { await service.settle(entry); return null; }, }, canRun); } diff --git a/server/src/services/routine-run-status.ts b/server/src/services/routine-run-status.ts index 79c1ff9a231..2a182def8f9 100644 --- a/server/src/services/routine-run-status.ts +++ b/server/src/services/routine-run-status.ts @@ -1,7 +1,7 @@ -import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; +import { notifyRoutineExecutionStatus, type RoutineExecutionStatusScope } from "./routine-execution-status-notifications.js"; import { and, eq, inArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { issues, pluginManagedResources, routineRuns, routines } from "@paperclipai/db"; +import { issues, routineRuns, routines } from "@paperclipai/db"; const EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE = "execution_issue_status"; const EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES = ["blocked", "cancelled"] as const; @@ -141,17 +141,11 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string } -export function routineExecutionOriginCondition() { - return or(sql`${issues.originId} = ${routineRuns.routineId}::text`, sql`exists ( - select 1 from ${pluginManagedResources} resource - where resource.company_id = ${routineRuns.companyId} - and resource.resource_kind = 'routine' and resource.resource_id = ${routineRuns.routineId} - and resource.defaults_json -> 'issueTemplate' ->> 'originId' = ${issues.originId} - )`); -} +export const ROUTINE_EXECUTION_STATUS_PAGE_SIZE = 50; export function routineExecutionStatusDebtCondition() { - const transientFailure = sql`(${routineRuns.triggerPayload} -> 'transientFailure' ->> 'code' = ${EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE} + const transientFailure = sql`((${routineRuns.triggerPayload} -> 'transientFailure' ->> 'code' = ${EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE} + and ${routineRuns.triggerPayload} -> 'transientFailure' ->> 'status' in ('blocked', 'cancelled')) or ${routineRuns.failureReason} in ('Execution issue moved to blocked', 'Execution issue moved to cancelled'))`; return or( and(eq(routineRuns.status, "issue_created"), inArray(issues.status, ["done", "blocked", "cancelled"])), @@ -165,15 +159,18 @@ export function routineExecutionStatusDebtCondition() { } export function routineExecutionStatusService(db: Db) { - async function load(issueId?: string) { + async function load(scope?: RoutineExecutionStatusScope, afterId?: string) { return db.select({ id: routineRuns.id, companyId: routineRuns.companyId, issueId: issues.id, routineId: routineRuns.routineId }) .from(routineRuns) .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, routineRuns.companyId))) .innerJoin(issues, and(eq(issues.id, routineRuns.linkedIssueId), eq(issues.companyId, routineRuns.companyId), - eq(issues.originKind, "routine_execution"), sql`${issues.originRunId} = ${routineRuns.id}::text`, routineExecutionOriginCondition())) - .where(and(routineExecutionStatusDebtCondition(), ...(issueId ? [eq(issues.id, issueId)] : []))) - .orderBy(routineRuns.id); + eq(issues.originKind, "routine_execution"), sql`${issues.originRunId} = ${routineRuns.id}::text`)) + .where(and(routineExecutionStatusDebtCondition(), + ...(scope?.issueId ? [eq(issues.id, scope.issueId)] : []), + ...(scope?.companyId ? [eq(routineRuns.companyId, scope.companyId)] : []), + ...(afterId ? [sql`${routineRuns.id} > ${afterId}::uuid`] : []))) + .orderBy(routineRuns.id).limit(ROUTINE_EXECUTION_STATUS_PAGE_SIZE); } async function settle(candidate: Awaited>[number]) { return db.transaction(async tx => { @@ -190,7 +187,7 @@ export function routineExecutionStatusService(db: Db) { .innerJoin(issues, eq(issues.id, routineRuns.linkedIssueId)) .where(and(eq(routineRuns.id, candidate.id), eq(routineRuns.companyId, candidate.companyId), eq(routineRuns.routineId, candidate.routineId), eq(routineRuns.linkedIssueId, issue.id), - routineExecutionOriginCondition(), routineExecutionStatusDebtCondition())).for("update", { of: routineRuns }); + routineExecutionStatusDebtCondition())).for("update", { of: routineRuns }); if (!run) return false; return (await syncRoutineRunStatusForIssue(tx as unknown as Db, issue.id)) !== null; }); diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index c391207950b..aa5771a9b20 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -2002,6 +2002,8 @@ export function routineService( } + let statusReconcileCursor: string | undefined; + async function syncRunStatusForIssue(issueId: string, executor: Db = db) { return syncRoutineRunStatusForIssue(executor, issueId); } @@ -3246,7 +3248,8 @@ export function routineService( reconcileExecutionRunStatuses: async (): Promise<{ scanned: number; settled: number; deferred: number }> => { const service = routineExecutionStatusService(db); - const candidates = await service.load(); + const candidates = await service.load(undefined, statusReconcileCursor); + statusReconcileCursor = candidates.length === 50 ? candidates.at(-1)!.id : undefined; let settled = 0, deferred = 0; for (const candidate of candidates) { try { diff --git a/server/src/services/system-recovery.ts b/server/src/services/system-recovery.ts index 8dba366e1af..a554a765c4c 100644 --- a/server/src/services/system-recovery.ts +++ b/server/src/services/system-recovery.ts @@ -71,7 +71,7 @@ export async function updateSystemRecovery(db: Db, source: WorkspaceRecoverySour sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,revision}' = ${String(previous.revision)}`)) .returning({ id: issueRecoveryActions.id }); return Boolean(updated); - }); + }, { companyId: source.companyId }); } export async function loadSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity) { diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index e9f0bd4edbf..17075c34eb9 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -334,7 +334,7 @@ export async function requestTaskRetry(input: { details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy) } }); return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id, chatActionId: intent.chatActionId }; - }); + }, { companyId: input.companyId }); } /** Chat owns admission and authorization. The task receipt follows that