From dd49318d554252ed0cd52134f60caa571aa2cf96 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:01:08 -0500 Subject: [PATCH 1/7] fix(server): settle routine status from committed work Co-Authored-By: Paperclip --- doc/plans/2026-10-30-routine-status-sleep.md | 54 +++++ .../routine-execution-status-work.test.ts | 206 ++++++++++++++++++ .../routine-execution-status-worker.test.ts | 86 ++++++++ .../routine-status-reconciliation.test.ts | 6 +- server/src/index.ts | 17 +- .../services/delivery-work-notifications.ts | 1 + .../execution-control-notifications.ts | 5 +- .../src/services/heartbeat/run-lifecycle.ts | 3 +- .../issue-review-state-notifications.ts | 2 + server/src/services/issues.ts | 11 + .../native-routine-status.test.ts | 6 +- .../routine-execution-status-notifications.ts | 5 + .../services/routine-execution-status-work.ts | 18 ++ server/src/services/routine-run-status.ts | 68 +++++- server/src/services/routines.ts | 41 +--- 15 files changed, 475 insertions(+), 54 deletions(-) create mode 100644 doc/plans/2026-10-30-routine-status-sleep.md create mode 100644 server/src/__tests__/routine-execution-status-work.test.ts create mode 100644 server/src/__tests__/routine-execution-status-worker.test.ts create mode 100644 server/src/services/routine-execution-status-notifications.ts create mode 100644 server/src/services/routine-execution-status-work.ts 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..ea6f9364999 --- /dev/null +++ b/doc/plans/2026-10-30-routine-status-sleep.md @@ -0,0 +1,54 @@ +# 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 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 +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 +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. 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__/routine-execution-status-work.test.ts b/server/src/__tests__/routine-execution-status-work.test.ts new file mode 100644 index 00000000000..7415b63bfd5 --- /dev/null +++ b/server/src/__tests__/routine-execution-status-work.test.ts @@ -0,0 +1,206 @@ +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("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..b88babbbd98 --- /dev/null +++ b/server/src/__tests__/routine-execution-status-worker.test.ts @@ -0,0 +1,86 @@ +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); + 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 worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready; + service.load.mockResolvedValueOnce([entry]); + worker.invalidate("issue"); worker.invalidate("issue"); + await vi.advanceTimersByTimeAsync(1000); + expect(service.load).toHaveBeenLastCalledWith("issue"); 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("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 (_issueId?: 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; + 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("issue"); await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(service.settle).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull(); +}); diff --git a/server/src/__tests__/routine-status-reconciliation.test.ts b/server/src/__tests__/routine-status-reconciliation.test.ts index eba0718ce25..964e43d6035 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") { diff --git a/server/src/index.ts b/server/src/index.ts index ae229f03518..8484b58054f 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,5 @@ +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"; @@ -1560,15 +1562,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 statusCards = statusCardService(db as any); const issues = issueService(db as any); const mergedPullRequestConfirmations = issueThreadInteractionService(db as any, { @@ -1989,7 +1982,6 @@ async function startServerWithDatabaseTeardown( // captures the authoritative set of running heartbeat rows. trackHeartbeatSchedulerWork((async () => { if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - scheduleRoutineStatusReconcile(); trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => { logger.error({ err }, "decision retention sweep failed"); })); @@ -2097,6 +2089,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 9195f1fcb12..22171c435cf 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -114,6 +114,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 22c8913bbcd..33a99774a48 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -1,7 +1,9 @@ +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"; -export async function notifyExecutionRunWork(transaction: object) { +export async function notifyExecutionRunWork(transaction: object, issueId?: string) { + await notifyRoutineExecutionStatus(transaction, issueId); for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { await notifyDeliveryWork(transaction, queue); @@ -14,6 +16,7 @@ export function writeExecutionControl( write: (tx: Db) => PromiseLike, ): Promise { return db.transaction(async transaction => { + await notifyRoutineExecutionStatus(transaction); 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/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 00a27c1faf2..f796da03a5e 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1044,7 +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); + await notifyExecutionRunWork(tx, previous?.nativeIssueId ?? previous?.issueId + ?? readNonEmptyString(parseObject(previous?.contextSnapshot).issueId) ?? 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 679b7e37475..15bc2a24287 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -1,9 +1,11 @@ +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"; 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 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 42c65bb5d02..eeebf326267 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +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"; @@ -10606,6 +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); 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) { @@ -11494,6 +11496,7 @@ export function issueService(db: Db) { if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { 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))) { @@ -11542,6 +11545,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) @@ -11958,6 +11964,7 @@ export function issueService(db: Db) { await notifyDecisionTargetIssueRemoval(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); removedIssue = await tx .delete(issues) .where(eq(issues.id, id)) @@ -12086,6 +12093,7 @@ export function issueService(db: Db) { const updated = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); return tx .update(issues) .set({ @@ -12215,6 +12223,7 @@ export function issueService(db: Db) { const adopted = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); return tx .update(issues) .set(adoptionSet) @@ -12470,6 +12479,7 @@ export function issueService(db: Db) { await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyRoutineExecutionStatus(tx, id); const updated = await tx .update(issues) .set({ @@ -12530,6 +12540,7 @@ export function issueService(db: Db) { } await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); 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/routine-execution-status-notifications.ts b/server/src/services/routine-execution-status-notifications.ts new file mode 100644 index 00000000000..cdd2be826c0 --- /dev/null +++ b/server/src/services/routine-execution-status-notifications.ts @@ -0,0 +1,5 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export function notifyRoutineExecutionStatus(tx: object, issueId?: string) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.routineExecutionStatus, issueId ?? "*"); +} 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..81c8e566a7c --- /dev/null +++ b/server/src/services/routine-execution-status-work.ts @@ -0,0 +1,18 @@ +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"; + +export function registerRoutineExecutionStatusWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + 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() })), + 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..79c1ff9a231 100644 --- a/server/src/services/routine-run-status.ts +++ b/server/src/services/routine-run-status.ts @@ -1,6 +1,7 @@ -import { and, eq } from "drizzle-orm"; +import { notifyRoutineExecutionStatus } 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"; +import { issues, pluginManagedResources, routineRuns, routines } from "@paperclipai/db"; const EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE = "execution_issue_status"; const EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES = ["blocked", "cancelled"] as const; @@ -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({ @@ -67,14 +69,16 @@ 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))) .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 +103,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 +140,60 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string return null; } + +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 function routineExecutionStatusDebtCondition() { + const transientFailure = sql`(${routineRuns.triggerPayload} -> 'transientFailure' ->> 'code' = ${EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE} + 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(issueId?: 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); + } + 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), + routineExecutionOriginCondition(), 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 23787b45249..c391207950b 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -73,7 +73,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, @@ -2001,7 +2001,6 @@ export function routineService( return run; } - let statusReconcileCursor: string | null = null; async function syncRunStatusForIssue(issueId: string, executor: Db = db) { return syncRoutineRunStatusForIssue(executor, issueId); @@ -3245,41 +3244,13 @@ 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(); + 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; @@ -3289,7 +3260,7 @@ export function routineService( current = cause.cause; } if (!retryable) throw error; - deferred += 1; + deferred++; } } return { scanned: candidates.length, settled, deferred }; From 46f2c0bb9a999b414497ba901b64a7c750c7fbc0 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:25:24 -0500 Subject: [PATCH 2/7] 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 From a4ec0d51516cecddf36fde146bdd7979d27dfdd2 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:26:01 -0500 Subject: [PATCH 3/7] test(server): preserve Cloud skill verification fixture coverage Co-Authored-By: Paperclip --- .../__tests__/managed-skill-content.test.ts | 91 ++++++++++++++++++- 1 file changed, 90 insertions(+), 1 deletion(-) diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index 677a3fbda65..c59f4a1277a 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; @@ -78,6 +80,93 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () }); it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => { + const companyId = await company(); + const objects = new Map(); + const server = createServer((request, response) => { + void (async () => { + const key = new URL(request.url!, "http://fixture").pathname; + if (request.method === "PUT") { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + let body = Buffer.concat(chunks); + if (request.headers["content-encoding"]?.includes("aws-chunked")) { + const decoded: Buffer[] = []; + let offset = 0; + while (offset < body.length) { + const end = body.indexOf("\r\n", offset); + const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16); + if (size === 0) break; + decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2; + } + body = Buffer.concat(decoded); + } + objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return; + } + const body = objects.get(key); + if (!body) { response.writeHead(404).end(); return; } + response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' }); + response.end(request.method === "HEAD" ? undefined : body); + })().catch(() => { response.writeHead(500).end(); }); + }); + await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address"); + const endpoint = `http://127.0.0.1:${address.port}`; + const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" }; + const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]])); + Object.assign(process.env, credentials); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({ + bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true, + }))); + const key = "cloud-runtime-config:v1"; + try { + await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const run = (phase: string) => promisify(execFile)(process.execPath, [ + "--import", fileURLToPath(import.meta.resolve("tsx")), + fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), + `--phase=${phase}`, `--company=${companyId}`, "--limit=50", + ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString, + PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", + PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3", + PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1", + PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" }, + timeout: 60_000, maxBuffer: 1024 * 1024 }); + await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); + await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); + const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!); + expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]); + const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const); + const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()]).toEqual(before); + const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!; + const archive = objects.get(archiveKey)!; + const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1; + objects.set(archiveKey, corrupted); + const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]); + objects.set(archiveKey, archive); + const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!; + const manifest = objects.get(manifestKey)!; objects.delete(manifestKey); + const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]); + objects.set(manifestKey, manifest); + const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()].sort()).toEqual([...before].sort()); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))) + .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]); + } finally { + await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key)); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")))); + for (const [key, value] of Object.entries(previousCredentials)) { + if (value === undefined) delete process.env[key]; else process.env[key] = value; + } + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); + } + }, 180_000); + + it("reports a provider mismatch for local versions under persisted Cloud configuration", async () => { const companyId = await company(); const skill = await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); const local = await skillInstallationMaintenance(db).verifyVersions({ companyId, limit: 50 }); @@ -90,7 +179,7 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () ], { env: { ...process.env, DATABASE_URL: database.connectionString, PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", PAPERCLIP_STORAGE_PROVIDER: "s3", PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-cli-fixture", - PAPERCLIP_STORAGE_S3_REGION: "us-east-1", PAPERCLIP_SECRETS_MASTER_KEY: "12345678901234567890123456789012" }, + PAPERCLIP_STORAGE_S3_REGION: "us-east-1", PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64) }, timeout: 60_000, maxBuffer: 1024 * 1024 }); await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); From cc3bc43ada773cdd66d650853a8209df1fcefc02 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:26:36 -0500 Subject: [PATCH 4/7] test(server): retain current master skill fixture Co-Authored-By: Paperclip --- .../__tests__/managed-skill-content.test.ts | 91 +------------------ 1 file changed, 1 insertion(+), 90 deletions(-) diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index c59f4a1277a..677a3fbda65 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -1,5 +1,3 @@ -import { createServer } from "node:http"; -import { createS3StorageProvider } from "../storage/s3-provider.js"; import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; @@ -80,93 +78,6 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () }); it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => { - const companyId = await company(); - const objects = new Map(); - const server = createServer((request, response) => { - void (async () => { - const key = new URL(request.url!, "http://fixture").pathname; - if (request.method === "PUT") { - const chunks: Buffer[] = []; - for await (const chunk of request) chunks.push(Buffer.from(chunk)); - let body = Buffer.concat(chunks); - if (request.headers["content-encoding"]?.includes("aws-chunked")) { - const decoded: Buffer[] = []; - let offset = 0; - while (offset < body.length) { - const end = body.indexOf("\r\n", offset); - const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16); - if (size === 0) break; - decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2; - } - body = Buffer.concat(decoded); - } - objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return; - } - const body = objects.get(key); - if (!body) { response.writeHead(404).end(); return; } - response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' }); - response.end(request.method === "HEAD" ? undefined : body); - })().catch(() => { response.writeHead(500).end(); }); - }); - await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); - const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address"); - const endpoint = `http://127.0.0.1:${address.port}`; - const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" }; - const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]])); - Object.assign(process.env, credentials); - vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({ - bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true, - }))); - const key = "cloud-runtime-config:v1"; - try { - await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); - const run = (phase: string) => promisify(execFile)(process.execPath, [ - "--import", fileURLToPath(import.meta.resolve("tsx")), - fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), - `--phase=${phase}`, `--company=${companyId}`, "--limit=50", - ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString, - PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", - PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3", - PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1", - PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" }, - timeout: 60_000, maxBuffer: 1024 * 1024 }); - await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); - expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); - await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); - const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!); - expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]); - const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const); - const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); - expect([...objects.entries()]).toEqual(before); - const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!; - const archive = objects.get(archiveKey)!; - const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1; - objects.set(archiveKey, corrupted); - const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]); - objects.set(archiveKey, archive); - const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!; - const manifest = objects.get(manifestKey)!; objects.delete(manifestKey); - const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]); - objects.set(manifestKey, manifest); - const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); - expect([...objects.entries()].sort()).toEqual([...before].sort()); - expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))) - .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]); - } finally { - await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key)); - vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")))); - for (const [key, value] of Object.entries(previousCredentials)) { - if (value === undefined) delete process.env[key]; else process.env[key] = value; - } - server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); - } - }, 180_000); - - it("reports a provider mismatch for local versions under persisted Cloud configuration", async () => { const companyId = await company(); const skill = await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); const local = await skillInstallationMaintenance(db).verifyVersions({ companyId, limit: 50 }); @@ -179,7 +90,7 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () ], { env: { ...process.env, DATABASE_URL: database.connectionString, PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", PAPERCLIP_STORAGE_PROVIDER: "s3", PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-cli-fixture", - PAPERCLIP_STORAGE_S3_REGION: "us-east-1", PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64) }, + PAPERCLIP_STORAGE_S3_REGION: "us-east-1", PAPERCLIP_SECRETS_MASTER_KEY: "12345678901234567890123456789012" }, timeout: 60_000, maxBuffer: 1024 * 1024 }); await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); From 9a5c429043cd5927499aa16cb24cbdc6375ec7ac Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:33:00 -0500 Subject: [PATCH 5/7] fix(server): scope lifecycle hints from selected run fields Co-Authored-By: Paperclip --- server/src/services/heartbeat/run-lifecycle.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index ac4e800f279..a1c4fa20fa3 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1044,7 +1044,7 @@ 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 ? { companyId: previous.companyId, agentId: previous.agentId, runId: previous.id, + 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 }); From 2cc2f9ae8cbf355ffba46596b38ce8eb7a04270f Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:39:30 -0500 Subject: [PATCH 6/7] fix(server): serialize direct routine status hooks Co-Authored-By: Paperclip --- doc/plans/2026-10-30-routine-status-sleep.md | 4 ++- .../routine-execution-status-work.test.ts | 34 +++++++++++++++++++ server/src/services/routine-run-status.ts | 8 ++++- 3 files changed, 44 insertions(+), 2 deletions(-) diff --git a/doc/plans/2026-10-30-routine-status-sleep.md b/doc/plans/2026-10-30-routine-status-sleep.md index 271d0b544a7..e852bf94861 100644 --- a/doc/plans/2026-10-30-routine-status-sleep.md +++ b/doc/plans/2026-10-30-routine-status-sleep.md @@ -18,7 +18,9 @@ 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 +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. diff --git a/server/src/__tests__/routine-execution-status-work.test.ts b/server/src/__tests__/routine-execution-status-work.test.ts index 53d7e5c0225..e6bd818852c 100644 --- a/server/src/__tests__/routine-execution-status-work.test.ts +++ b/server/src/__tests__/routine-execution-status-work.test.ts @@ -174,6 +174,40 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.transaction(async tx => { await notifyRoutineExecutionStatus(tx, otherRun.linkedIssueId!); }); await projected(otherRun.id, "completed"); 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; diff --git a/server/src/services/routine-run-status.ts b/server/src/services/routine-run-status.ts index 2a182def8f9..46a2bae4cf5 100644 --- a/server/src/services/routine-run-status.ts +++ b/server/src/services/routine-run-status.ts @@ -51,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, @@ -60,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 @@ -75,6 +80,7 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string .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 || ["received", "skipped", "coalesced"].includes(run.status)) return null; if (issue.status === "done") { From 4b8d9a64ca3873100e0814ea7d89e3e8a11472cf Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:16:05 -0500 Subject: [PATCH 7/7] fix(server): notify routine status after subtree restore Co-Authored-By: Paperclip --- doc/plans/2026-10-30-routine-status-sleep.md | 4 +++- .../issue-tree-control-service.test.ts | 2 +- .../routine-execution-status-work.test.ts | 22 +++++++++++++++++++ server/src/services/issue-tree-control.ts | 2 ++ 4 files changed, 28 insertions(+), 2 deletions(-) diff --git a/doc/plans/2026-10-30-routine-status-sleep.md b/doc/plans/2026-10-30-routine-status-sleep.md index e852bf94861..0afe809bb8a 100644 --- a/doc/plans/2026-10-30-routine-status-sleep.md +++ b/doc/plans/2026-10-30-routine-status-sleep.md @@ -9,7 +9,9 @@ scanned every five minutes, even with no routines or only completed history. `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 and review state writes. It uses the existing +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. 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 index e6bd818852c..0a6ff9339cb 100644 --- a/server/src/__tests__/routine-execution-status-work.test.ts +++ b/server/src/__tests__/routine-execution-status-work.test.ts @@ -1,3 +1,4 @@ +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"; @@ -174,6 +175,27 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu 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!)); diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 0d7633f846a..68ba89f527e 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +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"; @@ -1006,6 +1007,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 });