diff --git a/doc/plans/2026-10-30-routine-status-sleep.md b/doc/plans/2026-10-30-routine-status-sleep.md new file mode 100644 index 00000000000..0afe809bb8a --- /dev/null +++ b/doc/plans/2026-10-30-routine-status-sleep.md @@ -0,0 +1,61 @@ +# Routine execution status settlement + +The Cloud routine status worker repairs a routine invocation when its linked +execution task and saved routine status disagree. The previous heartbeat callback +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 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, review state writes and subtree status restoration. +Bulk restoration registers each issue hint in the same transaction before its +grouped write. Restored history does not depend on an optional agent wake. 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. + +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. +Direct route hooks also lock the execution task and linked invocation in their +own transaction, nested under caller transactions when present. An older hook +cannot write after a newer task status has committed. It preserves skipped/coalesced history and unrelated dispatch failures. A task +that resumes clears an execution-task transient failure through the same owner +used by native completion and issue routes. + +`activeIssue` remains derived from open tasks and live heartbeat runs. A terminal +heartbeat with an open task does not create a completion obligation. The worker +does not retry merely because a run is active or a routine exists. + +## Lifecycle and writer contract + +`registerScopedDeadlineWork` owns coalescing, fair batches, failure backoff, drain +and shutdown. A failed owner retries with bounded backoff. Successful settlement +removes that owner. Settled configuration, healthy active tasks and completed +history leave no recurring worker SQL. The existing Cloud and heartbeat scheduler +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 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. + +The in-process retry deadline does not establish external Cloud wake coverage. +This change preserves the existing host sleep safety checks and does not claim +measured host or database suspension. + +## Acceptance + +Tests exercise ordinary issue writes and a real heartbeat lifecycle transition +against PostgreSQL with the production coordinator. Provider wakeup is simulated. +They inspect durable routine rows and the normal routine detail read separately. +Coverage includes empty/configured/history quiet periods, first producer, +terminal outcomes and reopen, legacy mismatch reconstruction, skipped/coalesced +custody, outer commit/rollback, retry/restart, drain/resume and standby/disable. +Focused worker tests exercise in-flight invalidations, retry isolation and stop. 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__/issue-tree-control-service.test.ts b/server/src/__tests__/issue-tree-control-service.test.ts index 923f399d3bd..e98efb18309 100644 --- a/server/src/__tests__/issue-tree-control-service.test.ts +++ b/server/src/__tests__/issue-tree-control-service.test.ts @@ -41,7 +41,7 @@ describeEmbeddedPostgres("issueTreeControlService", () => { beforeAll(async () => { if (externalUrl) { await applyPendingMigrations(externalUrl); - db = createDb(externalUrl); + db = createDb(externalUrl, { maxConnections: 6 }); return; } tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-tree-control-"); diff --git a/server/src/__tests__/routine-execution-status-work.test.ts b/server/src/__tests__/routine-execution-status-work.test.ts new file mode 100644 index 00000000000..0a6ff9339cb --- /dev/null +++ b/server/src/__tests__/routine-execution-status-work.test.ts @@ -0,0 +1,314 @@ +import { issueTreeControlService } from "../services/issue-tree-control.js"; +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"; +import { agents, companies, createDb, heartbeatRuns, issues, pluginManagedResources, plugins, routineRuns, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyRoutineExecutionStatus } from "../services/routine-execution-status-notifications.js"; +import { registerRoutineExecutionStatusWork } from "../services/routine-execution-status-work.js"; +import { routineExecutionStatusService } from "../services/routine-run-status.js"; +import { routineService } from "../services/routines.js"; +import { issueService } from "../services/issues.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js"; +import { createHeartbeatRunState } from "../services/heartbeat/run-state.js"; +import { createHeartbeatRunPreparation } from "../services/heartbeat/run-preparation.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ROUTINE_STATUS_WORK_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("routine execution status work with PostgreSQL", () => { + let db: Db, temporary: Awaited> | undefined; + let coordinator: ReturnType | undefined; + let enabled = true, draining = false, standby = false; + const errors: unknown[] = []; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("routine-status-work-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 }); + }, 60_000); + beforeEach(() => { + enabled = true; draining = false; standby = false; errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`SET client_min_messages = warning; TRUNCATE companies CASCADE`); + }); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + const canRun = () => enabled && !draining && !standby; + async function start(service = routineExecutionStatusService(db)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile: () => !standby, + onError: error => errors.push(error) }); + const worker = registerRoutineExecutionStatusWork(coordinator, service, canRun); + await worker.ready; + return worker; + } + async function settled() { await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); } + async function quiet() { + await settled(); + const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); execute.mockRestore(); transaction.mockRestore(); } + } + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Routine status work", issuePrefix: `R${companyId.slice(0, 6)}`, + defaultResponsibleUserId: randomUUID(), requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Routine worker", status: "active", lifecycleState: "ready", adapterType: "codex_local" }); + const service = routineService(db, { heartbeat: { wakeup: async () => null }, runtimeEnv: {} }); + const routine = await service.create(companyId, { title: "Status projection", assigneeAgentId: agentId, + priority: "medium", status: "active", concurrencyPolicy: "always_enqueue", catchUpPolicy: "skip_missed" }, {}); + return { companyId, agentId, routine, service }; + } + async function dispatch(f: Awaited>) { + return f.service.runRoutine(f.routine.id, { source: "manual", idempotencyKey: randomUUID() }); + } + async function saved(id: string) { return (await db.select().from(routineRuns).where(eq(routineRuns.id, id)))[0]; } + async function projected(id: string, status: string) { + await vi.waitFor(async () => expect((await saved(id)).status).toBe(status)); + await settled(); + } + + it("settles empty startup, configured idle, healthy active and retained terminal history without recurring SQL", async () => { + await start(); await quiet(); + const f = await fixture(); await quiet(); + const run = await dispatch(f); await quiet(); + await issueService(db).update(run.linkedIssueId!, { status: "done" }); + await projected(run.id, "completed"); + const before = await saved(run.id); await quiet(); + expect(await saved(run.id)).toEqual(before); expect(errors).toEqual([]); + }); + it.each(["done", "blocked", "cancelled"])("reactivates from the first ordinary %s transition and preserves input", async status => { + await start(); await quiet(); + const f = await fixture(), run = await dispatch(f); + await issueService(db).update(run.linkedIssueId!, { status }); + await projected(run.id, status === "done" ? "completed" : "failed"); + expect((await saved(run.id)).completedAt).toBeInstanceOf(Date); + await quiet(); expect(errors).toEqual([]); + if (status !== "done") { + await issueService(db).update(run.linkedIssueId!, { status: "in_progress" }); + await projected(run.id, "issue_created"); + expect(await saved(run.id)).toMatchObject({ completedAt: null, failureReason: null }); + await quiet(); + } + }); + it("reconstructs legacy terminal debt and reopened transient failures but leaves skipped and coalesced history unchanged", async () => { + const f = await fixture(), first = await dispatch(f), second = await dispatch(f); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, first.linkedIssueId!)); + await db.update(routineRuns).set({ status: "failed", failureReason: "Execution issue moved to blocked", completedAt: new Date() }).where(eq(routineRuns.id, second.id)); + const terminal = await db.insert(routineRuns).values(["skipped", "coalesced"].map(status => ({ companyId: f.companyId, + routineId: f.routine.id, source: "manual", linkedIssueId: first.linkedIssueId, status, completedAt: new Date() }))).returning(); + await start(); await projected(first.id, "completed"); await projected(second.id, "issue_created"); + for (const row of terminal) expect(await saved(row.id)).toEqual(row); + await quiet(); expect(errors).toEqual([]); + }); + it("settles the declared custom origin of a real plugin-managed dispatch and preserves other companies", async () => { + const f = await fixture(), other = await fixture(); + const pluginKey = `routine-status-${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(); + await db.insert(pluginManagedResources).values({ companyId: f.companyId, pluginId: plugin.id, pluginKey, + resourceKind: "routine", resourceKey: "status", resourceId: f.routine.id, defaultsJson: { issueTemplate: { originId } } }); + const run = await dispatch(f), otherRun = await dispatch(other); + expect((await db.select().from(issues).where(eq(issues.id, run.linkedIssueId!)))[0].originId).toBe(originId); + await start(); await quiet(); + await issueService(db).update(run.linkedIssueId!, { status: "done" }); await projected(run.id, "completed"); + const before = await saved(run.id); + await db.transaction(async tx => { await notifyRoutineExecutionStatus(tx, run.linkedIssueId!); await notifyRoutineExecutionStatus(tx, run.linkedIssueId!); }); + 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("projects subtree cancel and restore without waking agents, then becomes quiet", async () => { + const f = await fixture(), root = await dispatch(f), child = await dispatch(f); + await issueService(db).update(child.linkedIssueId!, { parentId: root.linkedIssueId }); + const tree = issueTreeControlService(db), actor = { actorType: "system" as const, actorId: "routine-status-test" }; + await start(); await quiet(); + const cancel = await tree.createHold(f.companyId, root.linkedIssueId!, { mode: "cancel", actor }); + expect((await tree.cancelIssueStatusesForHold(f.companyId, root.linkedIssueId!, cancel.hold.id)).updatedIssueIds.sort()) + .toEqual([root.linkedIssueId, child.linkedIssueId].sort()); + await projected(root.id, "failed"); await projected(child.id, "failed"); await quiet(); + const restore = await tree.createHold(f.companyId, root.linkedIssueId!, { mode: "restore", actor }); + const restored = await tree.restoreIssueStatusesForHold(f.companyId, root.linkedIssueId!, restore.hold.id, { actor }); + expect(restored.updatedIssues).toEqual(expect.arrayContaining([ + expect.objectContaining({ id: root.linkedIssueId, status: "todo" }), + expect.objectContaining({ id: child.linkedIssueId, status: "todo" }), + ])); + await projected(root.id, "issue_created"); await projected(child.id, "issue_created"); + expect(await saved(root.id)).toMatchObject({ completedAt: null, failureReason: null }); + expect(await saved(child.id)).toMatchObject({ completedAt: null, failureReason: null }); + expect(await db.select().from(heartbeatRuns)).toEqual([]); + await quiet(); expect(errors).toEqual([]); + }); + it("serializes a direct status hook with later task changes before projecting the routine", async () => { + const f = await fixture(), run = await dispatch(f); await start(); await quiet(); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, run.linkedIssueId!)); + let entered!: () => void, release!: () => void; + const enteredGate = new Promise(resolve => { entered = resolve; }); + const releaseGate = new Promise(resolve => { release = resolve; }); + const holding = db.transaction(async tx => { + await tx.select().from(routineRuns).where(eq(routineRuns.id, run.id)).for("update"); + entered(); await releaseGate; + }); + let direct: Promise | undefined, changing: Promise | undefined; + let directPid: number | undefined; + try { + await enteredGate; direct = f.service.syncRunStatusForIssue(run.linkedIssueId!); + await vi.waitFor(async () => { + const rows = await db.execute(sql`select pid from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + and (query like 'update "routine_runs"%' or query like 'select %"routine_runs"%')`); + expect(rows).toHaveLength(1); directPid = Number(rows[0]!.pid); + }, { timeout: 5000 }); + await expect(db.transaction(tx => tx.select({ id: issues.id }).from(issues) + .where(eq(issues.id, run.linkedIssueId!)).for("update", { noWait: true }))) + .rejects.toMatchObject({ cause: expect.objectContaining({ code: "55P03" }) }); + changing = issueService(db).update(run.linkedIssueId!, { status: "blocked" }); + await vi.waitFor(async () => { + const rows = await db.execute(sql`select count(*)::int as count from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + and ${directPid}::int = any(pg_blocking_pids(pid))`); + expect(Number(rows[0]?.count)).toBeGreaterThan(0); + }, { timeout: 5000 }); + release(); await holding; await direct; await changing; + await projected(run.id, "failed"); await quiet(); expect(errors).toEqual([]); + } finally { release(); await holding; await direct; await changing; } + }); + 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; + const gate = new Promise(resolve => { release = resolve; }); + const written = new Promise(resolve => { entered = resolve; }); + const writing = db.transaction(async tx => { + await issueService(db).update(run.linkedIssueId!, { status: "done" }, tx); + entered(); await gate; + }); + try { await written; expect((await saved(run.id)).status).toBe("issue_created"); } + finally { release(); await writing; } + await projected(run.id, "completed"); await quiet(); + const other = await dispatch(f); + await expect(db.transaction(async tx => { + await issueService(db).update(other.linkedIssueId!, { status: "cancelled" }, tx); + throw new Error("outer rollback"); + })).rejects.toThrow("outer rollback"); + await settled(); expect((await saved(other.id)).status).toBe("issue_created"); await quiet(); expect(errors).toEqual([]); + }); + it("keeps the first write during drain and reconstructs on resume without passive SQL", async () => { + const f = await fixture(), run = await dispatch(f); await start(); await quiet(); + draining = true; + await issueService(db).update(run.linkedIssueId!, { status: "cancelled" }); + await settled(); await quiet(); expect((await saved(run.id)).status).toBe("issue_created"); + draining = false; await vi.advanceTimersByTimeAsync(30_000); await projected(run.id, "failed"); + await quiet(); expect(errors).toEqual([]); + }); + it.each(["standby", "disabled"])("does no owner SQL while %s and reconstructs when enabled", async mode => { + const f = await fixture(), run = await dispatch(f); + await issueService(db).update(run.linkedIssueId!, { status: "done" }); + if (mode === "standby") standby = true; else enabled = false; + await start(); await quiet(); expect((await saved(run.id)).status).toBe("issue_created"); + standby = false; enabled = true; await vi.advanceTimersByTimeAsync(30_000); + await projected(run.id, "completed"); await quiet(); expect(errors).toEqual([]); + }); + it("retries failed exact owners, retains the retry across drain and reconstructs after restart", async () => { + const f = await fixture(), run = await dispatch(f), service = routineExecutionStatusService(db); + const settle = vi.spyOn(service, "settle").mockRejectedValueOnce(new Error("projection unavailable")); + await start(service); + await issueService(db).update(run.linkedIssueId!, { status: "done" }); + await vi.waitFor(() => expect(errors).toHaveLength(1)); + expect((await saved(run.id)).status).toBe("issue_created"); + await vi.advanceTimersByTimeAsync(29_000); expect(settle).toHaveBeenCalledTimes(1); + draining = true; await vi.advanceTimersByTimeAsync(60_000); await quiet(); + draining = false; await vi.advanceTimersByTimeAsync(30_000); await projected(run.id, "completed"); await quiet(); + await coordinator!.stop(); coordinator = undefined; + const other = await dispatch(f); await issueService(db).update(other.linkedIssueId!, { status: "blocked" }); + await start(); await projected(other.id, "failed"); await quiet(); + }); + it("preserves notifications during a sweep, coalesces duplicate hints and never overlaps", async () => { + const f = await fixture(), first = await dispatch(f), second = await dispatch(f), service = routineExecutionStatusService(db); + let entered!: () => void, release!: () => void; + const enteredGate = new Promise(resolve => { entered = resolve; }); + const releaseGate = new Promise(resolve => { release = resolve; }); + const load = service.load; + vi.spyOn(service, "load").mockImplementationOnce(async issueId => { const rows = await load(issueId); entered(); await releaseGate; return rows; }); + const starting = start(service); await enteredGate; + await issueService(db).update(second.linkedIssueId!, { status: "done" }); + await db.transaction(async tx => { await notifyRoutineExecutionStatus(tx, second.linkedIssueId!); await notifyRoutineExecutionStatus(tx, second.linkedIssueId!); }); + release(); await starting; + await projected(second.id, "completed"); expect((await saved(first.id)).status).toBe("issue_created"); + await quiet(); expect(errors).toEqual([]); + }); + it.each(["succeeded", "failed", "cancelled", "interrupted", "timed_out"])("uses real %s run lifecycle changes for activeIssue without an open-issue retry", async status => { + const f = await fixture(), run = await dispatch(f); await start(); + const [execution] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "running", contextSnapshot: { issueId: run.linkedIssueId } }).returning(); + await db.update(issues).set({ executionRunId: execution.id }).where(eq(issues.id, run.linkedIssueId!)); + expect((await f.service.getDetail(f.routine.id))?.activeIssue?.id).toBe(run.linkedIssueId); + const deps = { + ...createHeartbeatRunState(db), ...createHeartbeatRunPreparation(db), + issuesSvc: { addComment: vi.fn(), listReviewAttention: vi.fn() }, + recovery: { escalateStrandedAssignedIssue: vi.fn() }, companySkills: { completeTestRunForIssue: vi.fn() }, + budgets: { getInvocationBlock: vi.fn(async () => null) }, treeControlSvc: { getActivePauseHoldGate: vi.fn(async () => null) }, + enqueueWakeup: vi.fn(), getCurrentUserRedactionOptions: vi.fn(async () => ({})), + getAgentInvokability: vi.fn(async () => ({ invokable: true })), emitTerminalAgentTaskRun: vi.fn(), budgetHooks: {}, + } satisfies HeartbeatLifecycleDependencies; + await createHeartbeatLifecycle(db, deps).setRunStatusIfRunning(execution.id, status, { finishedAt: new Date() }); + expect((await f.service.getDetail(f.routine.id))?.activeIssue).toBeNull(); + expect((await saved(run.id)).status).toBe("issue_created"); await quiet(); expect(errors).toEqual([]); + }); +}); diff --git a/server/src/__tests__/routine-execution-status-worker.test.ts b/server/src/__tests__/routine-execution-status-worker.test.ts new file mode 100644 index 00000000000..b62245ce3e2 --- /dev/null +++ b/server/src/__tests__/routine-execution-status-worker.test.ts @@ -0,0 +1,107 @@ +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"; +import { registerRoutineExecutionStatusWork } from "../services/routine-execution-status-work.js"; + +let coordinator: ReturnType; +let canRun = true; +const errors: unknown[] = []; +const entry = { id: "run", companyId: "company", routineId: "routine", issueId: "issue" }; +beforeEach(() => { + vi.useFakeTimers(); canRun = true; errors.length = 0; + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, canReconcile: () => true, + onError: error => errors.push(error) }); +}); +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, 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 (_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(JSON.stringify({ issueId: "issue" })); worker.invalidate(JSON.stringify({ issueId: "issue" })); + await vi.advanceTimersByTimeAsync(1000); + 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([]); +}); + +it("retries only the failed owner at backoff and disarms after success", async () => { + const service = { load: vi.fn(async () => [entry]), settle: vi.fn(async () => true).mockRejectedValueOnce(new Error("locked")) }; + await registerRoutineExecutionStatusWork(coordinator, service, () => canRun).ready; + expect(errors).toHaveLength(1); + const retryAt = coordinator.nextWakeAt(); expect(retryAt).toBe(Date.now() + 30_000); + await vi.advanceTimersByTimeAsync(29_999); expect(service.settle).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); expect(service.settle).toHaveBeenCalledTimes(2); + expect(service.load).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(20 * 60_000); expect(service.settle).toHaveBeenCalledTimes(2); +}); + +it("retains startup and first work during drain without passive owner calls", async () => { + 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(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); + expect(service.settle).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); +}); + +it("keeps a failed owner across drain and retries after resume", async () => { + const service = { load: vi.fn(async () => [entry]), settle: vi.fn(async () => true).mockRejectedValueOnce(new Error("write failed")) }; + await registerRoutineExecutionStatusWork(coordinator, service, () => canRun).ready; + canRun = false; await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(service.settle).toHaveBeenCalledOnce(); expect(service.load).toHaveBeenCalledOnce(); + canRun = true; await vi.advanceTimersByTimeAsync(30_000); + expect(service.settle).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull(); +}); + +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 (_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(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([]); +}); + +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(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 eba0718ce25..343c6fb38b6 100644 --- a/server/src/__tests__/routine-status-reconciliation.test.ts +++ b/server/src/__tests__/routine-status-reconciliation.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; @@ -5,11 +6,12 @@ import { companies, createDb, issues, routineRuns, routines } from "@paperclipai import { routineService } from "../services/routines.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ROUTINE_STATUS_TEST_DATABASE_URL"); +const support = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); if (!support.supported) console.warn(`Skipping routine status database tests: ${support.reason}`); (support.supported ? describe : describe.skip)("routine status reconciliation", () => { let db: ReturnType, database: Awaited>; - beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("routine-status-"); db = createDb(database.connectionString); }, 60_000); + beforeAll(async () => { if (!external) database = await startEmbeddedPostgresTestDatabase("routine-status-"); db = createDb(external ?? database.connectionString, { maxConnections: 6 }); }, 60_000); afterEach(async () => { vi.restoreAllMocks(); await db.execute(sql`SET client_min_messages = warning; TRUNCATE companies CASCADE`); }); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); async function seed(status = "blocked") { @@ -36,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"); @@ -53,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 9ff16029a29..172d0b68734 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,5 +1,6 @@ import { statusCardEvaluationService, registerStatusCardEvaluationWork } from "../services/status-card-evaluation.js"; import { attentionRetentionService } from "../services/attention-retention.js"; +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"; @@ -400,6 +401,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, @@ -1422,6 +1428,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/index.ts b/server/src/index.ts index 7e9a4fed7d7..d71125d7b80 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,4 +1,6 @@ import { attentionRetentionService, registerAttentionRetentionWork } from "./services/attention-retention.js"; +import { registerRoutineExecutionStatusWork } from "./services/routine-execution-status-work.js"; +import { routineExecutionStatusService } from "./services/routine-run-status.js"; import { registerProviderTraceRetentionWork } from "./services/provider-trace-retention-work.js"; import { registerRunEvidenceMigrationWork } from "./services/run-evidence-migration-work.js"; import { registerSkillVersionRetentionWork } from "./services/skill-version-retention-work.js"; @@ -1564,15 +1566,6 @@ async function startServerWithDatabaseTeardown( heartbeat.drainActiveRunExecutions(); prepareHotRestartShutdown = heartbeat.prepareHotRestartShutdown; const routines = routineService(db as any, { pluginWorkerManager }); - let nextRoutineStatusReconcile = 0; - const scheduleRoutineStatusReconcile = () => { - if (!isCloudManagedInstance() || isWarmStandby() || isIdleTaskDrainActive() || heartbeatSchedulerStopped || - Date.now() < nextRoutineStatusReconcile) return; - nextRoutineStatusReconcile = Date.now() + 5 * 60_000; - trackHeartbeatSchedulerWork(routines.reconcileExecutionRunStatuses().then(result => { - if (result.settled > 0) logger.info(result, "reconciled routine execution issue status"); - }).catch(() => logger.warn("routine execution status reconciliation deferred"))); - }; const statusCardEvaluation = statusCardEvaluationService(db as any, heartbeat); const issues = issueService(db as any); const mergedPullRequestConfirmations = issueThreadInteractionService(db as any, { @@ -1979,7 +1972,6 @@ async function startServerWithDatabaseTeardown( // captures the authoritative set of running heartbeat rows. trackHeartbeatSchedulerWork((async () => { if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - scheduleRoutineStatusReconcile(); const sweptRuntimeStatuses = heartbeat.sweepExpiredRuntimeStatuses(); if (sweptRuntimeStatuses > 0) { logger.info( @@ -2056,6 +2048,11 @@ async function startServerWithDatabaseTeardown( })); }; const startHeartbeatScheduler = () => { + if (isCloudManagedInstance()) { + const status = registerRoutineExecutionStatusWork(deliveryWork, routineExecutionStatusService(db as any), + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped); + trackHeartbeatSchedulerWork(status.ready); + } const scheduled = registerScheduledWork(deliveryWork, { retryMs: config.heartbeatSchedulerIntervalMs, tickAgentTimers: async signal => { diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 18e66ce5064..a42bf9e484b 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -118,6 +118,7 @@ export const DELIVERY_QUEUES = { pluginLifecycle: "plugin-resource-lifecycle", toolAction: "tool-action-receipts", terminalWorkspaces: "terminal-workspaces", + routineExecutionStatus: "routine-execution-status", routineSchedules: "routine-schedules", issueMonitors: "issue-monitors", agentTimers: "agent-timers", diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 8fdc1fad1d6..d8193013f21 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -1,6 +1,7 @@ import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; +import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js"; @@ -12,6 +13,7 @@ export async function notifyExecutionRunWork(transaction: object, scope?: Execut await notifyRunDispatch(transaction, scope); await notifyStatusCardIssueChange(transaction, scope.companyId); } + 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); @@ -28,6 +30,7 @@ export function writeExecutionControl( await notifyRunDispatch(transaction, scope); if (scope) await notifyStatusCardIssueChange(transaction, scope.companyId); if (scope) await notifyAttentionRetentionWork(transaction, scope); + 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-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index d56e7d926f1..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"; @@ -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({ diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 16207b0fb28..ca9fb995fc2 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,5 +1,6 @@ import { notifyRunDispatch } from "../run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; +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"; @@ -3924,6 +3925,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 9d3b30367d2..8224f75208e 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1045,7 +1045,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, companyId ? { companyId } : undefined); + await notifyExecutionRunWork(tx, previous ? { companyId: previous.companyId, 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/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index 997a37ecbf7..f87d38f7961 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -1,6 +1,7 @@ import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; +import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; @@ -10,6 +11,7 @@ export async function notifyIssueReviewState(tx: object, scope: { companyId: str await notifyRunDispatch(tx, scope); await notifyStatusCardIssueChange(tx, scope.companyId); await notifyAttentionRetentionWork(tx, { companyId: scope.companyId }); + 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/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index a8009559827..e784be722a5 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,4 +1,5 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; +import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; @@ -1007,6 +1008,7 @@ export function issueTreeControlService(db: Db) { for (const [status, issueIdsForStatus] of issueIdsByStatus) { if (issueIdsForStatus.length === 0) continue; for (const issueId of issueIdsForStatus) { + await notifyRoutineExecutionStatus(tx, issueId, companyId); await notifyNativeStatusReassessment(tx, { companyId, issueId }); await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); await notifyChatResponseAvailability(tx, { companyId, issueId }); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 312b089c7f9..e5ddd9b5168 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,6 +1,7 @@ import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; +import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -10620,6 +10621,7 @@ export function issueService(db: Db) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyStatusCardIssueChange(tx, companyId); await notifyAttentionRetentionWork(tx, { companyId }); + 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) { @@ -11513,6 +11515,7 @@ export function issueService(db: Db) { || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { await notifyRunDispatch(tx, { companyId: receiptExisting.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); } if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt && (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined))) { @@ -11561,6 +11564,9 @@ export function issueService(db: Db) { await notifyNativeWorkspaceFinalization(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeStatusReassessment(tx, { companyId: receiptExisting.companyId, issueId: id }); } + if (patch.originKind !== undefined || patch.originRunId !== undefined || patch.originId !== undefined) { + await notifyRoutineExecutionStatus(tx, id); + } const updated = await tx .update(issues) .set(patch) @@ -11983,6 +11989,7 @@ export function issueService(db: Db) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); removedIssue = await tx .delete(issues) .where(eq(issues.id, id)) @@ -12114,6 +12121,7 @@ export function issueService(db: Db) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); return tx .update(issues) .set({ @@ -12249,6 +12257,7 @@ export function issueService(db: Db) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); return tx .update(issues) .set(adoptionSet) @@ -12506,6 +12515,7 @@ export function issueService(db: Db) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); const updated = await tx .update(issues) .set({ @@ -12569,6 +12579,7 @@ export function issueService(db: Db) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); const updated = await tx .update(issues) .set(patch) diff --git a/server/src/services/native-runtime/native-routine-status.test.ts b/server/src/services/native-runtime/native-routine-status.test.ts index 668de83a4d5..415622ea2c3 100644 --- a/server/src/services/native-runtime/native-routine-status.test.ts +++ b/server/src/services/native-runtime/native-routine-status.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; @@ -8,13 +9,14 @@ import { routineService } from "../routines.js"; import { commitNativeStatusDecision } from "./status-decision-committer.js"; import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeStatusDecision } from "./status-arbiter.js"; -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_ROUTINE_STATUS_TEST_DATABASE_URL"); +const support = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); if (!support.supported) console.warn(`Skipping native routine status tests: ${support.reason}`); (support.supported ? describe : describe.skip)("native routine history status projection", () => { let db: ReturnType, database: Awaited>; const companyId = randomUUID(), agentId = randomUUID(); beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("native-routine-status-"); db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("native-routine-status-"); db = createDb(external ?? database.connectionString, { maxConnections: 6 }); await db.insert(companies).values({ id: companyId, name: "Routine status fixture", issuePrefix: "NRS" }); await db.insert(agents).values({ id: agentId, companyId, name: "Routine worker", adapterType: "paperclip_runner", status: "running" }); }, 60_000); diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.test.ts b/server/src/services/native-runtime/native-workspace-reconstruction.test.ts index fcff7eb58ec..8afb9820bee 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.test.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.test.ts @@ -38,7 +38,7 @@ describe("reconstruction after confirmed native workspace loss", () => { 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 2c327994fe4..2daf462972f 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.ts @@ -1,5 +1,6 @@ import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; import { notifyAttentionRetentionWork } from "../retention-work-notifications.js"; +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"; @@ -191,6 +192,7 @@ export async function reconstructNativeWorkspace(input: { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyStatusCardIssueChange(tx, input.companyId); + 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 new file mode 100644 index 00000000000..fb9df28dda9 --- /dev/null +++ b/server/src/services/routine-execution-status-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +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 new file mode 100644 index 00000000000..0a4437f89f2 --- /dev/null +++ b/server/src/services/routine-execution-status-work.ts @@ -0,0 +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 { 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 : 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 77341f9cba4..46a2bae4cf5 100644 --- a/server/src/services/routine-run-status.ts +++ b/server/src/services/routine-run-status.ts @@ -1,4 +1,5 @@ -import { and, eq } from "drizzle-orm"; +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, routineRuns, routines } from "@paperclipai/db"; @@ -37,6 +38,7 @@ function legacyExecutionIssueTransientFailureStatus( } export async function finalizeRoutineRun(executor: Db, runId: string, patch: Partial) { + if (patch.linkedIssueId) await notifyRoutineExecutionStatus(executor, patch.linkedIssueId); return executor .update(routineRuns) .set({ @@ -49,6 +51,10 @@ export async function finalizeRoutineRun(executor: Db, runId: string, patch: Par } export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string) { + return executor.transaction(tx => syncLockedRoutineRunStatusForIssue(tx as unknown as Db, issueId)); +} + +async function syncLockedRoutineRunStatusForIssue(executor: Db, issueId: string) { const issue = await executor .select({ id: issues.id, @@ -58,7 +64,8 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string originRunId: issues.originRunId, }) .from(issues) - .where(eq(issues.id, issueId)) + .where(and(eq(issues.id, issueId), eq(issues.originKind, "routine_execution"))) + .for("update") .then((rows) => rows[0] ?? null); if (!issue || issue.originKind !== "routine_execution" || !issue.originRunId) return null; const run = await executor @@ -67,14 +74,17 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string status: routineRuns.status, failureReason: routineRuns.failureReason, triggerPayload: routineRuns.triggerPayload, + completedAt: routineRuns.completedAt, }) .from(routineRuns) .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, issue.companyId))) .where(and(eq(routineRuns.id, issue.originRunId), eq(routineRuns.companyId, issue.companyId), eq(routineRuns.linkedIssueId, issue.id))) + .for("update", { of: routineRuns }) .then((rows) => rows[0] ?? null); - if (!run) return null; + if (!run || ["received", "skipped", "coalesced"].includes(run.status)) return null; if (issue.status === "done") { + if (run.status === "completed" && run.failureReason === null && run.completedAt) return null; const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); const transientFailureClearedAt = executionIssueTransientFailureClearedAtFromPayload(run.triggerPayload); @@ -99,6 +109,7 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string } if (issue.status === "blocked" || issue.status === "cancelled") { const failureReason = executionIssueTransientFailureReason(issue.status); + if (run.status === "failed" && run.failureReason === failureReason && run.completedAt) return null; return finalizeRoutineRun(executor, issue.originRunId, { status: "failed", failureReason, @@ -135,3 +146,57 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string return null; } + +export const ROUTINE_EXECUTION_STATUS_PAGE_SIZE = 50; + +export function routineExecutionStatusDebtCondition() { + 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"])), + and(eq(routineRuns.status, "completed"), or(inArray(issues.status, ["blocked", "cancelled"]), + and(eq(issues.status, "done"), sql`(${routineRuns.completedAt} is null or ${routineRuns.failureReason} is not null)`))), + and(eq(routineRuns.status, "failed"), transientFailure, + sql`(${issues.status} not in ('blocked', 'cancelled') + or ${routineRuns.completedAt} is null + or ${routineRuns.failureReason} is distinct from 'Execution issue moved to ' || ${issues.status})`), + ); +} + +export function routineExecutionStatusService(db: Db) { + 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`)) + .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 => { + await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`); + await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + const [routine] = await tx.select({ id: routines.id }).from(routines).where(and( + eq(routines.id, candidate.routineId), eq(routines.companyId, candidate.companyId))).for("share"); + if (!routine) return false; + const [issue] = await tx.select().from(issues).where(and(eq(issues.id, candidate.issueId), + eq(issues.companyId, candidate.companyId), eq(issues.originKind, "routine_execution"), + eq(issues.originRunId, candidate.id))).for("update"); + if (!issue) return false; + const [run] = await tx.select({ id: routineRuns.id }).from(routineRuns) + .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), + routineExecutionStatusDebtCondition())).for("update", { of: routineRuns }); + if (!run) return false; + return (await syncRoutineRunStatusForIssue(tx as unknown as Db, issue.id)) !== null; + }); + } + return { load, settle }; +} diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index a831a3e489a..30fefbbabbf 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -74,7 +74,7 @@ import { secretService } from "./secrets.js"; import { getSecretProvider } from "../secrets/provider-registry.js"; import { parseCron, validateCron } from "./cron.js"; import { heartbeatService } from "./heartbeat.js"; -import { finalizeRoutineRun, syncRoutineRunStatusForIssue } from "./routine-run-status.js"; +import { finalizeRoutineRun, routineExecutionStatusService, syncRoutineRunStatusForIssue } from "./routine-run-status.js"; import { instanceSettingsService, isTruthyRuntimeEnvValue, @@ -2003,7 +2003,8 @@ export function routineService( return run; } - let statusReconcileCursor: string | null = null; + + let statusReconcileCursor: string | undefined; async function syncRunStatusForIssue(issueId: string, executor: Db = db) { return syncRoutineRunStatusForIssue(executor, issueId); @@ -3247,41 +3248,14 @@ export function routineService( return { triggered }; }, - // Repair a missed issue-status hook through the same status transition as - // normal issue writes. Never discard queued input or infer run completion - // from age. Lock both owners and re-read the issue before applying repair. reconcileExecutionRunStatuses: async (): Promise<{ scanned: number; settled: number; deferred: number }> => { - const candidates = await 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`, sql`${issues.originId} = ${routineRuns.routineId}::text`)) - .where(and(eq(routineRuns.status, "issue_created"), inArray(issues.status, ["done", "cancelled", "blocked"]), - ...(statusReconcileCursor ? [sql`${routineRuns.id}::text > ${statusReconcileCursor}`] : []))) - .orderBy(asc(routineRuns.id)).limit(50); - statusReconcileCursor = candidates.length === 50 ? candidates.at(-1)!.id : null; - let settled = 0; - let deferred = 0; + const service = routineExecutionStatusService(db); + 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 { - const repaired = await db.transaction(async tx => { - await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`); - await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); - const [routine] = await tx.select({ id: routines.id }).from(routines).where(and( - eq(routines.id, candidate.routineId), eq(routines.companyId, candidate.companyId))).for("share"); - if (!routine) return false; - const [issue] = await tx.select().from(issues).where(and(eq(issues.id, candidate.issueId), - eq(issues.companyId, candidate.companyId), eq(issues.originKind, "routine_execution"), - eq(issues.originRunId, candidate.id), eq(issues.originId, candidate.routineId))).for("update"); - if (!issue || !["done", "cancelled", "blocked"].includes(issue.status)) return false; - const [run] = await tx.select().from(routineRuns).where(and(eq(routineRuns.id, candidate.id), - eq(routineRuns.companyId, candidate.companyId), eq(routineRuns.routineId, candidate.routineId), - eq(routineRuns.linkedIssueId, issue.id), eq(routineRuns.status, "issue_created"))).for("update"); - if (!run) return false; - return (await syncRunStatusForIssue(issue.id, tx as unknown as Db)) !== null; - }); - if (repaired) settled += 1; + if (await service.settle(candidate)) settled++; } catch (error) { let current: unknown = error; let retryable = false; @@ -3291,7 +3265,7 @@ export function routineService( current = cause.cause; } if (!retryable) throw error; - deferred += 1; + deferred++; } } return { scanned: candidates.length, settled, deferred };