diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index e5401596a0e..71825b7024c 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -59,7 +59,7 @@ and fences automatic result persistence. It does not modify that draft PR. | Issue title/description, comment and document references | Existing sync methods register object intent before mention replacement and stable identity intent before object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | | Comment/document disappearance | Sync derives the retained mention owners and registers only those companies before deleting obsolete mentions. | | Last mention removal and reattachment | Mention replacement invalidates company deadlines. Retained objects stay available for reattachment. | -| Issue deletion | Issue service registers its company before cascading mention deletion. | +| Issue deletion | The issue service selects distinct mentioned object IDs by company and `sourceIssueId`, then registers only those IDs before the cascade. The source owner includes title, description, comments and documents. | | Company status/archive/deletion, including budget pause/resume | Company and budget services register company reconstruction before status or deletion writes. | | Feature enable/disable | Instance settings registers global reconstruction before the experimental-settings write. | | Plugin manifest/status/uninstall | Plugin registry registers reconstruction before lifecycle writes. Installed plugins acquire support through the ordinary ready transition. | diff --git a/doc/plans/2026-10-30-run-dispatch-sleep.md b/doc/plans/2026-10-30-run-dispatch-sleep.md index e5cd0a9e014..8ad35c128fc 100644 --- a/doc/plans/2026-10-30-run-dispatch-sleep.md +++ b/doc/plans/2026-10-30-run-dispatch-sleep.md @@ -12,6 +12,15 @@ invalidate retries notify the retry worker in their transaction. Promotion uses the existing gate and compare-and-set, and its committed write wakes ordinary queued dispatch. A queued run behind occupied capacity waits for the owning run's lifecycle transition. Configuration and authority changes also wake it. +Retry-schedule notifications are separate from ordinary queue eligibility. +`notifyRunDispatch` wakes only queued dispatch; ordinary issue, agent, budget +and company edits cannot repeatedly scan the global retry schedule. +`notifyScheduledRetryDispatch` wakes retry reconciliation as well as dispatch +for actual retry creation, rescheduling and cancellation. Its producers include +bounded and operator-requested retries, recovery retries, native capacity and +safe-replacement successors, and the existing retry-cancellation transition +boundary. Shared runtime changes do not impose a spacing floor on a saved deadline. + An idle queue has no recurring read or timer-driven dispatch attempt. Notifications batch for one second and carry the changed company to queued dispatch. Known retry deadlines still preempt batching. Cancellation-owned retries stay with their cancellation worker and do not retain a promotion deadline. The supported writer contract is service writes through the shared database diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index 78c56ff9a83..56b0b057af3 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -78,7 +78,15 @@ describe("deferred receipt durable dispatch", () => { return { companyId, agent, issue, source, actorId }; } async function quiet() { - await vi.waitFor(async () => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 }); + await vi.waitFor(async () => { + if (vi.isFakeTimers()) { + // Fake-clock cases have no live execution. Settle finite cleanup + // deadlines explicitly instead of advancing at waitFor's wall speed. + await vi.advanceTimersByTimeAsync(1000); + await Promise.allSettled([...passes]); await immediate(); + } + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 40_000 }); const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"); const selected = reads.mock.calls.length, executed = executes.mock.calls.length; if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); @@ -455,8 +463,8 @@ describe("deferred receipt durable dispatch", () => { const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)); expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId; }, { timeout: 40_000 }); + // Observe this Retry before closing the task and draining its follow-ups. await vi.waitFor(async () => { - await heartbeat.drainActiveRunExecutions(); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("receipt-proof"); diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index 8aab42b35b1..1a878aaea08 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -2,11 +2,12 @@ import { createServer } from "node:http"; import { randomUUID } from "node:crypto"; import { setTimeout as delay } from "node:timers/promises"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { eq, sql } from "drizzle-orm"; +import { and, eq, inArray, sql } from "drizzle-orm"; import { activityLog, approvals, applyPendingMigrations, databaseWorkPending, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, externalObjectMentions, issues, issueComments, plugins } from "@paperclipai/db"; import { externalObjectService, type ExternalObjectResolveResult } from "../services/external-objects.js"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { issueService } from "../services/issues.js"; import { documentService } from "../services/documents.js"; import { budgetService } from "../services/budgets.js"; import { pluginRegistryService } from "../services/plugin-registry.js"; @@ -367,6 +368,94 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { expect(provider).toHaveBeenCalledTimes(2); }); + it("scopes repeated issue deletions to cascaded source objects and cancels their last future mention", async () => { + clock(); const svc = service(); + const own = await fixture(), foreign = await fixture(); + const future = new Date(Date.now() + 86_400_000), removedDeadline = new Date(Date.now() + 1_800_000); + const sharedUrl = "https://example.com/shared-issue-object"; + const [retained] = await db.insert(issues).values({ companyId: own.companyId, title: "Retained", + description: sharedUrl }).returning(); + await svc.syncIssue(retained!.id); + await db.update(issues).set({ description: sharedUrl }).where(eq(issues.id, foreign.issueId)); + await svc.syncIssue(foreign.issueId); + const deletions: Array<{ id: string; affected: string[]; orphaned: string[] }> = []; + for (let index = 0; index < 3; index++) { + const [issue] = await db.insert(issues).values({ companyId: own.companyId, + title: sharedUrl, description: `https://example.com/deleted-${index}` }).returning(); + await svc.syncIssue(issue!.id); + const [comment] = await db.insert(issueComments).values({ companyId: own.companyId, issueId: issue!.id, + body: `https://example.com/comment-${index}` }).returning(); + await svc.syncComment(comment!.id); + const doc = await documentService(db).upsertIssueDocument({ issueId: issue!.id, key: "notes", format: "markdown", + body: `https://example.com/document-${index}` }); + await svc.syncDocument(doc.document.id); + const mentions = await db.select().from(externalObjectMentions).where(and( + eq(externalObjectMentions.companyId, own.companyId), eq(externalObjectMentions.sourceIssueId, issue!.id))); + expect(new Set(mentions.map(mention => mention.sourceKind))).toEqual(new Set(["title", "description", "comment", "document"])); + deletions.push({ id: issue!.id, affected: [...new Set(mentions.map(mention => mention.objectId!))], + orphaned: mentions.filter(mention => mention.sourceKind !== "title").map(mention => mention.objectId!) }); + } + const shared = (await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.sourceIssueId, retained!.id)))[0]!.objectId!; + for (let offset = 0; offset < 1100; offset += 250) { + const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID()); + await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId, + providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future }))); + await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId, + sourceIssueId: own.issueId, sourceKind: "description" as const, objectId }))); + } + await db.update(externalObjects).set({ nextRefreshAt: future }); + for (const entry of deletions) for (const id of entry.orphaned) + await db.update(externalObjects).set({ nextRefreshAt: removedDeadline }).where(eq(externalObjects.id, id)); + await start(svc); + await settle(() => coordinator!.nextWakeAt() === removedDeadline.getTime() && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + expect(provider).not.toHaveBeenCalled(); + const reads: Array<{ query: string; params: unknown[] }> = []; + const selections: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_connection, query, params) => { + if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"')) + reads.push({ query, params: [...params] }); + if (/^select distinct/i.test(query) && query.includes('from "external_object_mentions"')) + selections.push({ query, params: [...params] }); + }; + try { + for (let index = 0; index < deletions.length; index++) { + const entry = deletions[index], first = reads.length; + const result = await issueService(db).remove(entry.id); + expect(result?.id).toBe(entry.id); + const next = index === deletions.length - 1 ? future : removedDeadline; + await settle(() => reads.length > first && coordinator!.nextWakeAt() === next.getTime() + && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + const deletionReads = reads.slice(first); + expect(deletionReads.length).toBeLessThanOrEqual(entry.affected.length); + for (const read of deletionReads) { + expect(read.query).toContain('"external_objects"."id" ='); + expect(read.params).toContain(own.companyId); expect(read.params).not.toContain(foreign.companyId); + expect(read.params.some(value => entry.affected.includes(String(value)))).toBe(true); + } + } + expect(selections).toHaveLength(deletions.length); + for (let index = 0; index < selections.length; index++) { + expect(selections[index]!.query).toContain('"external_object_mentions"."company_id" ='); + expect(selections[index]!.query).toContain('"external_object_mentions"."source_issue_id" ='); + expect(selections[index]!.params).toContain(own.companyId); expect(selections[index]!.params).toContain(deletions[index]!.id); + } + const lastMentions = await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.objectId, shared)); + expect(lastMentions).toHaveLength(1); expect(lastMentions[0]!.sourceIssueId).toBe(retained!.id); + const orphanIds = deletions.flatMap(entry => entry.orphaned); + const orphanRows = await db.select().from(externalObjects).where(inArray(externalObjects.id, orphanIds)); + expect(orphanRows).toHaveLength(orphanIds.length); + for (const orphan of orphanRows) expect(orphan.nextRefreshAt).toEqual(removedDeadline); + expect(await db.select().from(externalObjectMentions).where(inArray(externalObjectMentions.objectId, orphanIds))).toEqual([]); + const foreignRows = await db.select().from(externalObjects).where(eq(externalObjects.companyId, foreign.companyId)); + expect(foreignRows).toHaveLength(1); expect(foreignRows[0]!.nextRefreshAt).toEqual(future); + const count = reads.length; + await vi.advanceTimersByTimeAsync(removedDeadline.getTime() - Date.now() + 1); + expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled(); + expect(coordinator!.nextWakeAt()).toBe(future.getTime()); + } finally { db.$client.options.debug = debug; } + }); + it("keeps warm standby startup SQL-free and resumes a first committed producer", async () => { clock(); admitted = false; reconcile = false; const svc = service(); const selects = vi.spyOn(db, "select"); diff --git a/server/src/__tests__/github-connection-work.test.ts b/server/src/__tests__/github-connection-work.test.ts index 783ed45de43..de620d3f7b5 100644 --- a/server/src/__tests__/github-connection-work.test.ts +++ b/server/src/__tests__/github-connection-work.test.ts @@ -44,7 +44,11 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( beforeAll(async () => { if (databaseUrl) await applyPendingMigrations(databaseUrl); else temporaryDatabase = await startEmbeddedPostgresTestDatabase("paperclip-github-work-"); - db = createDb(databaseUrl ?? temporaryDatabase!.connectionString, { maxConnections: 4 }); + // Only scheduler time is virtual here. Driver housekeeping must not expire + // a real socket connection while that fake clock advances across idle windows. + db = createDb(databaseUrl ?? temporaryDatabase!.connectionString, { + maxConnections: 4, idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0, + }); // This explicitly assigned external test database is exclusive to this suite. await db.transaction(async tx => { await tx.execute(sql`SET LOCAL client_min_messages = warning`); await tx.execute(sql`TRUNCATE companies CASCADE`); }); server = createServer(async (req, res) => { @@ -409,7 +413,8 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( const schedule = f.service.githubConnectionContinuitySchedule.bind(f.service); const load = vi.spyOn(f.service, "githubConnectionContinuitySchedule").mockImplementation(companyId => { loadTimes.push(Date.now()); return schedule(companyId); }); const sweep = vi.spyOn(f.service, "sweepGitHubConnectionContinuity"); - const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); const continuity = registerGitHubConnectionContinuityWork(db, coordinator, f.service, () => true); const events = registerGitHubConnectionEventWork(db, coordinator, githubConnectionEventService(db, { connector: f.connector }), () => true); try { @@ -423,6 +428,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( } await vi.advanceTimersByTimeAsync(10_000); expect(load).not.toHaveBeenCalled(); expect(sweep).not.toHaveBeenCalled(); expect(leases).toBe(1); + expect(errors).toEqual([]); for (let index = 0; index < 30; index++) await db.transaction(async tx => { await notifyGitHubCredentialChange(tx, f.connection); }); expect(leases).toBeLessThanOrEqual(2); await vi.advanceTimersByTimeAsync(6_000); @@ -442,6 +448,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( const quiet = { leases, loads: load.mock.calls.length }; await vi.advanceTimersByTimeAsync(60_000); expect({ leases, loads: load.mock.calls.length }).toEqual(quiet); + expect(errors).toEqual([]); } finally { await events.stop(); await continuity.stop(); await coordinator.stop(); } }); diff --git a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts index f3398d828f8..a9e120387c8 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts @@ -3,7 +3,7 @@ import { PgDialect } from "drizzle-orm/pg-core"; import { afterEach, describe, expect, it, vi } from "vitest"; import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; -import { notifyRunDispatch } from "../services/run-dispatch-notifications.js"; +import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js"; import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js"; const cleanup: Array<() => Promise> = []; @@ -38,6 +38,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean }, () => !state.draining && !state.standby); cleanup.push(() => coordinator.stop()); return { db, state, promote, next, dispatch, coordinator, worker, + notifyRetry: () => db.transaction(tx => notifyScheduledRetryDispatch(tx)), notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)), settle: () => vi.advanceTimersByTimeAsync(1_000) }; } @@ -61,9 +62,9 @@ describe("run dispatch obligations", () => { }); it("rearms earlier work and cancels a future deadline on a committed producer change", async () => { const f = fixture({ due: 1_600_000 }); await f.worker.ready; - f.state.due = 1_005_000; await f.notify(); await f.settle(); + f.state.due = 1_005_000; await f.notifyRetry(); await f.settle(); expect(f.coordinator.nextWakeAt()).toBe(1_005_000); - f.state.due = null; await f.notify(); await f.settle(); + f.state.due = null; await f.notifyRetry(); await f.settle(); expect(f.coordinator.nextWakeAt()).toBeNull(); const before = f.next.mock.calls.length; await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before); @@ -76,7 +77,7 @@ describe("run dispatch obligations", () => { await f.notify(source); await vi.advanceTimersByTimeAsync(10); } - expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2); + expect(f.promote).toHaveBeenCalledOnce(); expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2); await f.settle(); expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" }); @@ -87,7 +88,7 @@ describe("run dispatch obligations", () => { }); it("remembers the first producer arriving during drain without owner SQL", async () => { const f = fixture(); await f.worker.ready; f.state.draining = true; - await f.notify(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000); + await f.notifyRetry(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce(); expect(idleWorkSnapshot().active).toBe(0); f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000); @@ -96,7 +97,7 @@ describe("run dispatch obligations", () => { }); it("releases a failed producer hold during drain and retains recovery for resume", async () => { const f = fixture(); await f.worker.ready; - f.state.fail = true; await f.notify(); await f.settle(); + f.state.fail = true; await f.notifyRetry(); await f.settle(); expect(f.promote).toHaveBeenCalledTimes(2); f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce(); @@ -109,8 +110,8 @@ describe("run dispatch obligations", () => { let release!: () => void; const held = new Promise(resolve => { release = resolve; }); f.promote.mockImplementationOnce(() => held); - await f.notify(); await f.settle(); - await f.notify(); await f.settle(); + await f.notifyRetry(); await f.settle(); + await f.notifyRetry(); await f.settle(); expect(f.promote).toHaveBeenCalledTimes(2); release(); await f.settle(); await vi.advanceTimersByTimeAsync(30_000); @@ -121,7 +122,7 @@ describe("run dispatch obligations", () => { await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).not.toHaveBeenCalled(); f.state.standby = false; await vi.advanceTimersByTimeAsync(30_000); expect(f.promote).toHaveBeenCalledOnce(); expect(f.state.due).toBeNull(); - f.state.due = Date.now() + 60_000; await f.notify(); await f.settle(); await f.worker.stop(); + f.state.due = Date.now() + 60_000; await f.notifyRetry(); await f.settle(); await f.worker.stop(); await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2); }); }); diff --git a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts index bcc8ed23972..1c0e5005a04 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts @@ -2,9 +2,11 @@ import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns } from "@paperclipai/db"; +import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns, issues } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { issueService } from "../services/issues.js"; +import { materializeNativeProviderCapacityRetry } from "../services/native-runtime/native-provider-capacity-retry.js"; import { agentService } from "../services/agents.js"; import { budgetService } from "../services/budgets.js"; import { heartbeatService } from "../services/heartbeat.js"; @@ -24,9 +26,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled let reads: ReturnType; let draining = false; const errors: unknown[] = []; + const retryQueries: string[] = []; beforeAll(async () => { if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-"); db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + db.$client.options.debug = (_connection, query) => { + if (query.startsWith("select") && query.includes('"heartbeat_runs"') && + query.includes("startupCancellation") && query.includes('"scheduled_retry_at"')) retryQueries.push(query); + }; await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); }, 90_000); beforeEach(async () => { @@ -88,6 +95,45 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled await fixture(); await worker!.stop(); await coordinator!.stop(); await start(); await quiet(); }); + it.each(["agent", "issue"] as const)("does not scan global scheduled retries for repeated ordinary %s edits", async kind => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Edit source", status: "backlog" }).returning(); + await start(); await quiet(); + const before = retryQueries.length; + expect(before).toBeGreaterThan(0); // The passive observer saw real startup retry SQL. + for (let index = 0; index < 12; index++) { + if (kind === "agent") await agentService(db).update(f.agentId, { name: `Edited agent ${index}` }); + else await issueService(db).update(issue.id, { title: `Edited issue ${index}` }); + await new Promise(resolve => setTimeout(resolve, 100)); + } + await quiet(); + expect(retryQueries).toHaveLength(before); + }); + it.each([false, true])("observes a native capacity retry only after its transaction settles (rollback=%s)", async rollback => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Capacity retry", status: "in_progress", + assigneeAgentId: f.agentId }).returning(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.runId)); + await start(); await quiet(); + let retryId = ""; + const writing = db.transaction(async tx => { + retryId = await materializeNativeProviderCapacityRetry({ tx: tx as unknown as typeof db, ...f, issueId: issue.id }); + expect(databaseWorkPending(db, DELIVERY_QUEUES.scheduledRetryDispatch)).toBe(true); + expect(await status(retryId)).toBeUndefined(); + if (rollback) throw new Error("native capacity rollback"); + }); + if (rollback) { + await expect(writing).rejects.toThrow("native capacity rollback"); + await quiet(); expect(await status(retryId)).toBeUndefined(); + } else { + await writing; + const retry = await status(retryId); + expect(retry.status).toBe("scheduled_retry"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(retry.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + await heartbeat.cancelRun(retryId, "fixture teardown", { suppressImmediateRecovery: true }); + await quiet(); + } + }); it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => { const f = await fixture(); await start(); await quiet(); const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 }); diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index 9d16ec519ff..25827def7b5 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -24,6 +24,8 @@ import { companies, createDb, environmentLeases, + environments, + sandboxCreationOperations, heartbeatRuns, issues, nativeRunFinalizations, @@ -768,6 +770,43 @@ describeEmbeddedPostgres("native harness backup expiration", () => { await quiet(); expect(await exists(own.current)).toBe(true); }); + it.each([false, true])("sandbox attachment reconciles only its retained root using durable run authority (explicit run: %s)", async explicitRun => { + const own = await seedSession({ runStatus: "running" }); + const unrelated = []; + for (let index = 0; index < 16; index++) unrelated.push(await seedSession({ companyId: own.companyId, + agentId: own.agentId, issueStatus: "todo", runStatus: index % 2 ? "running" : "succeeded" })); + const [environment] = await db.insert(environments).values({ name: randomUUID(), driver: "sandbox" }).returning(); + async function attach(runId: string) { + const operationId = randomUUID(); + await db.insert(sandboxCreationOperations).values({ id: operationId, companyId: own.companyId, + environmentId: environment!.id, originalEnvironmentId: environment!.id, runId, + provider: "controlled", driverKey: "controlled", config: {}, metadata: {}, nextAttemptAt: new Date() }); + return environmentService(db).acquireLease({ creationOperationId: operationId, companyId: own.companyId, + environmentId: environment!.id, ...(explicitRun ? { heartbeatRunId: runId } : {}) }); + } + await startBackups(); await settled(); + const reads = vi.spyOn(fs, "readFile"); + try { + const lease = await attach(own.runId); await settled(); + expect(reads.mock.calls.map(call => String(call[0]))) + .toEqual([path.join(own.root, "control-plane", "control-plane-state.json")]); + expect((await db.select().from(sandboxCreationOperations).where(eq(sandboxCreationOperations.leaseId, lease.id)))[0]) + .toMatchObject({ state: "attached", runId: own.runId }); + reads.mockClear(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: own.companyId, agentId: own.agentId, status: "running" }).returning(); + await attach(run!.id); await settled(); + expect(reads).not.toHaveBeenCalled(); + expect(await exists(own.current)).toBe(true); + for (const other of unrelated) expect(await exists(other.current)).toBe(true); + await quiet(); + } finally { + reads.mockRestore(); + await db.delete(sandboxCreationOperations).where(eq(sandboxCreationOperations.companyId, own.companyId)); + await db.delete(environmentLeases).where(eq(environmentLeases.environmentId, environment!.id)); + await db.delete(environments).where(eq(environments.id, environment!.id)); + } + }); + it("indexes protected session relations so a new run, historical task close and exact process release reactivate only its backup", async () => { const own = await seedSession({ closedDaysAgo: 2 }); const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" }); diff --git a/server/src/__tests__/task-watchdogs-classifier.test.ts b/server/src/__tests__/task-watchdogs-classifier.test.ts index 16ef800c088..71865c75dfe 100644 --- a/server/src/__tests__/task-watchdogs-classifier.test.ts +++ b/server/src/__tests__/task-watchdogs-classifier.test.ts @@ -1,8 +1,6 @@ -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it } from "vitest"; import { classifyTaskWatchdogSubtree, type TaskWatchdogClassifierIssue } from "../services/task-watchdogs.ts"; -import { createIncompleteTaskWatchdogScanReporter } from "../services/task-watchdog-diagnostics.ts"; - const companyId = "company-1"; const sourceId = "source-1"; const childId = "child-1"; @@ -37,39 +35,6 @@ function classify(overrides: Partial { - it("reports an unchanged incomplete scan once across startup and a day of periodic ticks", () => { - const warn = vi.fn(); - const report = createIncompleteTaskWatchdogScanReporter(warn); - report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "startup"); - for (let tick = 0; tick < 2880; tick += 1) { - report({ incomplete: 2, incompleteIssueIds: [childId, sourceId] }, "periodic"); - } - expect(warn).toHaveBeenCalledTimes(1); - expect(warn).toHaveBeenCalledWith( - { incomplete: 2, newIncompleteIssueIds: [sourceId, childId] }, - "startup task-watchdog reconciliation found incomplete subtree scans", - ); - }); - - it("reports new incomplete issues and reports a resolved issue if it becomes incomplete again", () => { - const warn = vi.fn(); - const report = createIncompleteTaskWatchdogScanReporter(warn); - report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "startup"); - report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "periodic"); - expect(warn).toHaveBeenLastCalledWith( - { incomplete: 2, newIncompleteIssueIds: [childId] }, - "periodic task-watchdog reconciliation found incomplete subtree scans", - ); - report({ incomplete: 0, incompleteIssueIds: [] }, "periodic"); - expect(warn).toHaveBeenCalledTimes(2); - report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "periodic"); - expect(warn).toHaveBeenCalledTimes(3); - expect(warn).toHaveBeenLastCalledWith( - { incomplete: 1, newIncompleteIssueIds: [sourceId] }, - "periodic task-watchdog reconciliation found incomplete subtree scans", - ); - }); - it("does not produce a stopped fingerprint from an incomplete scan", () => { const result = classify({ subtreeIncomplete: true }); expect(result.state).toBe("incomplete"); diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index a9985370ae8..68fe46ede14 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -68,7 +68,7 @@ export async function attachSandboxCreation(db: Db, id: string, scope: { company } if (!operation || operation.state !== "pending") throw new Error("Sandbox creation ownership has already transferred"); const lease = await attach(tx as unknown as Db); - await notifyEnvironmentCleanup(tx, { companyId: scope.companyId }); + await notifyEnvironmentCleanup(tx, { companyId: scope.companyId, runId: operation.runId }); await tx.update(sandboxCreationOperations).set({ state: "attached", leaseId: lease.id, updatedAt: new Date() }) .where(eq(sandboxCreationOperations.id, id)); return lease; diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 0264ac23595..f09aad46dec 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,6 +1,6 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; -import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; import { notifyAttentionRetentionWork } from "../retention-work-notifications.js"; +import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js"; import { infrastructureRetryDelayMs } from "../system-recovery-state.js"; import { @@ -1000,8 +1000,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen retryContextSnapshot.previousRunId = continuation.previousRunId; } - await notifyRunDispatch(tx, { companyId: run.companyId }); await notifyAttentionRetentionWork(tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); const wakeupRequest = await tx .insert(agentWakeupRequests) .values({ @@ -1760,7 +1760,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen }; const updated = await db.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: scheduled.run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId }); const row = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 55880801756..1cf77db7926 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -100,6 +100,7 @@ import { heartbeatRuns, routineRuns, executionWorkspaces, + externalObjectMentions, issueApprovals, issueAccessGrants, issueAttachments, @@ -12044,7 +12045,14 @@ export function issueService(db: Db) { if (owner) { await notifyIssueAttentionRetentionWork(tx, { companyId: owner.companyId, issueId: id }); await notifyExternalConversationSettlement(tx, { companyId: owner.companyId }); - await notifyExternalObjectRefresh(tx, owner.companyId); + const mentionedObjects = await tx.selectDistinct({ objectId: externalObjectMentions.objectId }) + .from(externalObjectMentions) + .where(and( + eq(externalObjectMentions.companyId, owner.companyId), + eq(externalObjectMentions.sourceIssueId, id), + isNotNull(externalObjectMentions.objectId), + )); + for (const mention of mentionedObjects) await notifyExternalObjectRefresh(tx, owner.companyId, mention.objectId!); await notifyApprovalWork(tx, owner.companyId); await notifyStatusCardIssueChange(tx, owner.companyId); } diff --git a/server/src/services/native-runtime/native-provider-capacity-retry.ts b/server/src/services/native-runtime/native-provider-capacity-retry.ts index dfef23302a1..cb05729dcc5 100644 --- a/server/src/services/native-runtime/native-provider-capacity-retry.ts +++ b/server/src/services/native-runtime/native-provider-capacity-retry.ts @@ -1,5 +1,5 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; import { notifyAttentionRetentionWork } from "../retention-work-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db"; @@ -51,8 +51,8 @@ export async function materializeNativeProviderCapacityRetry(input: { scheduledRetryAt: dueAt.toISOString(), executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt), }; - await notifyRunDispatch(input.tx, { companyId: run.companyId }); await notifyAttentionRetentionWork(input.tx, { companyId: input.companyId }); + await notifyScheduledRetryDispatch(input.tx, { companyId: run.companyId }); const [wake] = await input.tx.insert(agentWakeupRequests).values({ companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system", reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId, diff --git a/server/src/services/native-runtime/native-safe-replacement.test.ts b/server/src/services/native-runtime/native-safe-replacement.test.ts index 422b23dba6b..491bae17b2b 100644 --- a/server/src/services/native-runtime/native-safe-replacement.test.ts +++ b/server/src/services/native-runtime/native-safe-replacement.test.ts @@ -1,3 +1,4 @@ +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { getExecutionBlocker } from "../execution-blocker.js"; import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; @@ -577,7 +578,12 @@ const support = externalDatabaseUrl }); it("persists exactly one linked successor under competing sweepers and restarts", async () => { const source = await seed(2); + const owner = heartbeatService(db); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const worker = owner.registerRunDispatchWork(coordinator, () => true); + await worker.ready; const now = new Date(); + try { await Promise.all([ reconcileSafeNativeReplacements(db, now), reconcileSafeNativeReplacements(db, now), @@ -608,6 +614,10 @@ const support = externalDatabaseUrl .where(eq(nativeRunFinalizations.runId, source.runId)) )[0]?.failureDetail?.successorRunId, ).toBe(children[0]!.id); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(children[0]!.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + } finally { + await worker.stop(); await coordinator.stop(); await owner.drainActiveRunExecutions(); + } }); it("retries status publication after a crash without dispatching provider work", async () => { await deliverExecutionStatuses(db); diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index da485883cb0..12b556fa779 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; @@ -410,6 +411,7 @@ export async function reconcileSafeNativeReplacements( }) .returning(); await notifyExecutionRunWork(tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: run.companyId, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index df9b7f84333..0b7bfc6db7a 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,4 +1,4 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -2716,7 +2716,7 @@ export function recoveryService( const now = new Date(); const retryAt = readProviderQuotaRetryAt(input.latestRun, now); return database.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: input.issue.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ @@ -3317,7 +3317,7 @@ export function recoveryService( created = Boolean(enqueuedRun); } else { scheduledRun = await db.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: input.issue.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 9661d2d72e0..1b013f4f304 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -7,6 +7,11 @@ export async function notifyRunDispatch(transaction: object, scope?: RunDispatch await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined); const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativeProcessSessions, hint); - await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); } + +/** Only a retry row or deadline change needs to re-read the saved retry schedule. */ +export async function notifyScheduledRetryDispatch(transaction: object, scope?: RunDispatchScope) { + await notifyRunDispatch(transaction, scope); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch); +} diff --git a/server/src/services/scheduled-retry-cancellation-notifications.ts b/server/src/services/scheduled-retry-cancellation-notifications.ts index 24e57de66cb..945bb40e14b 100644 --- a/server/src/services/scheduled-retry-cancellation-notifications.ts +++ b/server/src/services/scheduled-retry-cancellation-notifications.ts @@ -1,9 +1,9 @@ -import { notifyRunDispatch } from "./run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "./run-dispatch-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type ScheduledRetryCancellationScope = { companyId: string; runId?: string }; export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) { - await notifyRunDispatch(transaction, scope); + await notifyScheduledRetryDispatch(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*"); } diff --git a/server/src/services/task-watchdog-diagnostics.ts b/server/src/services/task-watchdog-diagnostics.ts deleted file mode 100644 index ea08eebe2b4..00000000000 --- a/server/src/services/task-watchdog-diagnostics.ts +++ /dev/null @@ -1,16 +0,0 @@ -export function createIncompleteTaskWatchdogScanReporter( - warn: (details: { incomplete: number; newIncompleteIssueIds: string[] }, message: string) => void, -) { - let previousIncompleteIssueIds = new Set(); - return (result: { incomplete: number; incompleteIssueIds: string[] }, phase: "startup" | "periodic") => { - const currentIssueIds = new Set(result.incompleteIssueIds); - const newIncompleteIssueIds = [...currentIssueIds].filter((id) => !previousIncompleteIssueIds.has(id)); - previousIncompleteIssueIds = currentIssueIds; - if (newIncompleteIssueIds.length === 0) return; - warn( - { incomplete: result.incomplete, newIncompleteIssueIds }, - `${phase} task-watchdog reconciliation found incomplete subtree scans`, - ); - }; -} - diff --git a/tests/e2e/agent-chat-projects.spec.ts b/tests/e2e/agent-chat-projects.spec.ts index 994df38746c..89ec94ca54c 100644 --- a/tests/e2e/agent-chat-projects.spec.ts +++ b/tests/e2e/agent-chat-projects.spec.ts @@ -459,12 +459,15 @@ test("plan approval hands the preserved revision to an assigned project task", a { timeout: 60_000 }, ) .toBe(1); + // Task completion is delivered back to the chat as another real turn. + await idle(request, f.chatPath, 3); const task = ( await json(await request.get(`/api/companies/${f.company.id}/issues`)) )[0]; expect(task.parentId).toBeNull(); expect(task.projectId).toBeTruthy(); expect(task.assigneeAgentId).toBe(f.agent.id); + await expect(page.getByText(`Delegated work completed: ${task.identifier}`, { exact: true })).toBeVisible(); const plan = await json( await request.get(`/api/issues/${task.id}/documents/plan`), ); diff --git a/tests/e2e/fixtures/agent-chat.mjs b/tests/e2e/fixtures/agent-chat.mjs index 2a66bdd3111..d85ab355679 100644 --- a/tests/e2e/fixtures/agent-chat.mjs +++ b/tests/e2e/fixtures/agent-chat.mjs @@ -36,6 +36,10 @@ if (!task.conversationAgentId) { }); process.exit(0); } +if (ctx.wakeReason === "chat_task_completed") { + await comment(`Delegated work completed: ${ctx.chatCompletionUpdates.map((update) => update.identifier).join(", ")}`); + process.exit(0); +} const comments = await api(`/issues/${task.id}/comments?order=asc`); const current = comments.find((c) => c.id === ctx.wakeCommentId) ??