From 6484f96c28bfb11a793ce4be6cc5ac15643a5a02 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sat, 31 Oct 2026 11:43:18 -0700 Subject: [PATCH 1/2] fix(workspaces): inspect storage only when needed (#18410) --- doc/DEVELOPING.md | 26 +-- doc/plugins/PLUGIN_RUNNER_EXECUTION.md | 15 ++ maps/agent-access-map/workspaces.md | 2 + maps/feature-map/workspaces.md | 1 + packages/shared/src/types/plugin.ts | 1 + packages/shared/src/validators/plugin.test.ts | 11 ++ packages/shared/src/validators/plugin.ts | 1 + .../execution-workspaces-service.test.ts | 10 +- .../plugin-workspaces.integration.test.ts | 153 +++++++++++++++++- .../workspace-git-inspection-cache.test.ts | 47 ------ server/src/services/execution-workspaces.ts | 142 ++++++++-------- server/src/services/plugin-registry.ts | 24 ++- server/src/services/plugin-workspaces.ts | 7 + .../workspace-git-inspection-cache.ts | 30 ---- 14 files changed, 305 insertions(+), 165 deletions(-) delete mode 100644 server/src/__tests__/workspace-git-inspection-cache.test.ts delete mode 100644 server/src/services/workspace-git-inspection-cache.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 0f28897914f..3aa15e16b20 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -110,8 +110,9 @@ and the [agent case reference](../skills/paperclip/references/cases.md). Task detail, heartbeat context, and closed-workspace admission guards read stored workspace metadata. They do not inspect Git or refresh live runtime health. Task context includes persisted runtime service state; delivery state is unknown -until an explicit workspace read or lifecycle inspection computes it. Use the -workspace detail and close-readiness APIs for current Git safety checks. +until an explicit close-readiness or lifecycle inspection computes it. Workspace +detail reads also return unknown delivery state without running Git or contacting +a storage provider. Use close-readiness for current Git safety checks. ## Deployment Modes @@ -3601,12 +3602,13 @@ Every delivery and request still checks current permissions. Grant revocation, deleted tasks, missing runs, and cross-company run IDs remain fail-closed. The developer-server health poll also stops in hidden tabs. -Workspace detail responses share concurrent Git inspections and reuse their -results for up to thirty seconds after completion. Workspace identity and metadata -changes invalidate the cached result. The cache holds at most 256 -entries. Close-readiness checks and final cleanup validation request separate -Git scans that cannot join an older in-flight scan. A display result never -authorizes worktree removal. +Workspace detail responses return stored metadata and runtime service information +without inspecting Git, resolving delivery PRs, or reconciling remote storage. +This also keeps operation-history reads, login handoffs, runtime controls and +metadata updates from inspecting storage through their common workspace lookup. +Delivery state stays `unknown` on these reads. Explicit close-readiness and final +cleanup validation still inspect current Git state; neither uses a display result +to authorize removal. The terminal-workspace reaper uses the database work coordinator. Startup scans recover existing workspace rows in bounded pages. Issue status and tree changes, @@ -3615,7 +3617,13 @@ their transaction before writing. The coordinator wakes after the outer transaction settles and holds idle admission while writes or cleanup are active. Empty and nonterminal-only lanes have no recurring timer or query. Terminal trees schedule their actual cooldown deadline; unresolved delivery or cleanup eligibility -is checked again after five minutes. Nonterminal trees, cooldowns, and active runs +is checked again after five minutes. Providers that explicitly declare +`supportsMergeAncestry: false` are inspected for automatic cleanup only when a +matching merged PR can provide delivery evidence. With no matching PR work +product, the unchanged workspace has no delivery-inspection deadline; existing +work-product notifications and startup recovery reconsider it. Open or unknown +remote PRs and unfinished cleanup retain their retries. Missing declarations keep +the previous ancestry-inspection behavior. Nonterminal trees, cooldowns, and active runs are checked before Git. The worker shares Git readiness across compatible rows that point at one directory within a bounded page. Company, branch, base ref, and runtime ownership must match. Each row keeps its own task and delivery diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index 9e6f746fc77..3142a596818 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -173,6 +173,21 @@ The existing agent-independent `inspect` and `remove` operations continue to protect storage maintenance. Providers that do not declare finalization version 1 retain the inspection-based completion path. No stored binding migration is required. +A provider whose inspection cannot prove merge ancestry may declare +`workspaceProvider.supportsMergeAncestry: false`. Omission preserves the existing +ancestry inspection behavior. For an opted-out provider, automatic cleanup first +checks for a matching PR work product. No matching PR means no storage inspection +or recurring cleanup deadline for that workspace; work-product and workspace +notifications reconsider it, and startup reconstructs the decision. Provider +manifest changes, activation, and company re-enablement also wake cleanup, so a +provider that later gains ancestry inspection can reconsider parked workspaces. +A relevant open or unknown PR retains the existing bounded retry because external merge +state can change without a local notification. Only a merged PR with the expected +branch and a known HEAD triggers inspection; Core still compares that HEAD with +fresh storage evidence before archive. Removal's fresh data-safety and lifecycle +checks are unchanged. Explicit close-readiness and pending lifecycle recovery +continue to inspect even without a PR. + ## Warm task retention A driver declares `warmTaskVersion: 1`, implements `onEnvironmentResumeLease`, diff --git a/maps/agent-access-map/workspaces.md b/maps/agent-access-map/workspaces.md index 2dc6ba5ad8b..ee7451d2d62 100644 --- a/maps/agent-access-map/workspaces.md +++ b/maps/agent-access-map/workspaces.md @@ -268,6 +268,8 @@ reconcile-branch requires board. Policy: Company/project/workspace visibility and runtime authority; active run, trust preset and task ownership where required. +Workspace metadata reads do not perform storage inspection or certify delivery. +The explicit close-readiness API and destructive cleanup retain live safety checks. Evidence: `source-only` — [implementation](../../server/src/routes/workspace-runtime-service-authz.ts#L76) at the diff --git a/maps/feature-map/workspaces.md b/maps/feature-map/workspaces.md index e82349ce27c..9d8923a67e2 100644 --- a/maps/feature-map/workspaces.md +++ b/maps/feature-map/workspaces.md @@ -181,4 +181,5 @@ Capability ID: `workspaces/closure`. Scope: ui Storybook: Close-readiness checks, pending changes/active runs, blocked/pending/error close, retry close and eligible reopening. Production: Confirm owned workspace shutdown, preserved files/Git state and permitted lifecycle transition/reopening; Browser Use session closure is unrelated. +Workspace detail and operation-history reads do not inspect Git or remote storage. Delivery is unknown until explicit close readiness or lifecycle inspection. Verify that these reads work during an inspection outage, while destructive close still refuses unverified data. Sources: Source: [ExecutionWorkspaceDetail.tsx](../../ui/src/pages/ExecutionWorkspaceDetail.tsx), [ExecutionWorkspaceCloseDialog.tsx](../../ui/src/components/ExecutionWorkspaceCloseDialog.tsx), [execution-workspaces.ts](../../server/src/routes/execution-workspaces.ts). Tests: [execution-workspaces-routes.test.ts](../../server/src/__tests__/execution-workspaces-routes.test.ts), [execution-workspaces-service.test.ts](../../server/src/__tests__/execution-workspaces-service.test.ts), [execution-workspace-reopen.test.ts](../../server/src/__tests__/execution-workspace-reopen.test.ts) diff --git a/packages/shared/src/types/plugin.ts b/packages/shared/src/types/plugin.ts index f5740487394..794f6cecdf4 100644 --- a/packages/shared/src/types/plugin.ts +++ b/packages/shared/src/types/plugin.ts @@ -719,6 +719,7 @@ export interface PaperclipPluginManifestV1 { workspaceProvider?: { /** Supports bounded run completion separately from removal-safety inspection. */ finalizationVersion?: 1; + supportsMergeAncestry?: boolean; freshWorkspaceVersion?: 1; modes: Array<"shared_workspace" | "isolated_workspace" | "operator_branch">; strategies: Array<"project_primary" | "git_worktree">; diff --git a/packages/shared/src/validators/plugin.test.ts b/packages/shared/src/validators/plugin.test.ts index 1331dadc16c..e145e15cb98 100644 --- a/packages/shared/src/validators/plugin.test.ts +++ b/packages/shared/src/validators/plugin.test.ts @@ -33,6 +33,17 @@ describe("plugin capability constants", () => { }); describe("plugin manifest validators", () => { + it("preserves explicit workspace ancestry support without changing legacy declarations", () => { + const manifest = { id: "example.storage", apiVersion: 1, version: "1.0.0", displayName: "Storage", description: "Storage", author: "Tests", + categories: ["workspace"], capabilities: ["workspaces.manage", "events.subscribe"], resourceLifecycle: ["project"], + entrypoints: { worker: "worker.js" }, workspaceProvider: { modes: ["isolated_workspace"], strategies: ["git_worktree"] } }; + expect(pluginManifestV1Schema.parse(manifest).workspaceProvider?.supportsMergeAncestry).toBeUndefined(); + for (const supportsMergeAncestry of [true, false]) { + expect(pluginManifestV1Schema.parse({ ...manifest, workspaceProvider: { ...manifest.workspaceProvider, supportsMergeAncestry } }) + .workspaceProvider?.supportsMergeAncestry).toBe(supportsMergeAncestry); + } + expect(pluginManifestV1Schema.safeParse({ ...manifest, workspaceProvider: { ...manifest.workspaceProvider, supportsMergeAncestry: "false" } }).success).toBe(false); + }); it("limits environment skills to unique declarations owned by the plugin", () => { const manifest = { ...buildSandboxProviderManifest({ skillKeys: ["desktop"] }), capabilities: ["environment.drivers.register", "skills.managed"], diff --git a/packages/shared/src/validators/plugin.ts b/packages/shared/src/validators/plugin.ts index 1713f5ef38c..841fcdc23a1 100644 --- a/packages/shared/src/validators/plugin.ts +++ b/packages/shared/src/validators/plugin.ts @@ -808,6 +808,7 @@ export const pluginManifestV1Schema = z.object({ agentLifecycle: z.literal(true).optional(), workspaceProvider: z.object({ finalizationVersion: z.literal(1).optional(), + supportsMergeAncestry: z.boolean().optional(), freshWorkspaceVersion: z.literal(1).optional(), modes: z.array(z.enum(["shared_workspace", "isolated_workspace", "operator_branch"])).min(1).max(3), strategies: z.array(z.enum(["project_primary", "git_worktree"])).min(1).max(2), diff --git a/server/src/__tests__/execution-workspaces-service.test.ts b/server/src/__tests__/execution-workspaces-service.test.ts index 8a4a523fa26..0d6ec7dff2f 100644 --- a/server/src/__tests__/execution-workspaces-service.test.ts +++ b/server/src/__tests__/execution-workspaces-service.test.ts @@ -575,7 +575,7 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { expect(await service.nextTerminalWorkspaceSweepAt()).toBe(sourceDone.getTime() + 7 * 86_400_000); }); - it("does not use cached display Git state to remove a worktree", async () => { + it("checks current Git state before removal after a metadata read", async () => { const seed = await seedTerminalWorkspace({ mergedPr: true }); await svc.getById(seed.executionWorkspaceId); await fs.writeFile(path.join(seed.worktreePath, "unsaved.txt"), "keep this work\n"); @@ -945,7 +945,7 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { expect(workspace).toMatchObject({ status: "archived", cleanupReason: "issue_terminal" }); }, 20_000); - it("reads company-scoped execution configuration without live Git inspection", async () => { + it("reads workspace details and company-scoped execution configuration without live Git inspection", async () => { const seed = await seedAncestryTerminalWorkspace(); const service = executionWorkspaceService(db, { inspectGitCloseReadiness: async () => { throw new Error("Git inspection unavailable"); }, @@ -970,7 +970,11 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { await expect(service.getStoredById(randomUUID(), seed.executionWorkspaceId)) .rejects.toThrow("Task workspace access is no longer available"); expect(await service.getStoredById(seed.companyId, randomUUID())).toBeNull(); - await expect(service.getById(seed.executionWorkspaceId)).rejects.toThrow("Git inspection unavailable"); + await expect(service.getById(seed.executionWorkspaceId)).resolves.toMatchObject({ + id: seed.executionWorkspaceId, deliveryState: "unknown", config, + runtimeServices: [{ id: runtimeServiceId, status: "running" }], + }); + await expect(service.getCloseReadiness(seed.executionWorkspaceId)).rejects.toThrow("Git inspection unavailable"); }); it("derives ancestry from the existing comparison and stays unknown for an invalid base", async () => { diff --git a/server/src/__tests__/plugin-workspaces.integration.test.ts b/server/src/__tests__/plugin-workspaces.integration.test.ts index 0d6a92f3c99..b838a7b7408 100644 --- a/server/src/__tests__/plugin-workspaces.integration.test.ts +++ b/server/src/__tests__/plugin-workspaces.integration.test.ts @@ -8,6 +8,7 @@ import { executionWorkspaces, heartbeatRuns, issues, + issueWorkProducts, plugins, pluginCompanySettings, } from "@paperclipai/db"; @@ -19,6 +20,7 @@ import { type WorkspaceWorkers, } from "../services/plugin-workspaces.js"; import { executionWorkspaceService } from "../services/execution-workspaces.js"; +import { pluginRegistryService } from "../services/plugin-registry.js"; import { projectService } from "../services/projects.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { @@ -43,6 +45,127 @@ const support = await getEmbeddedPostgresTestSupport(); }); const fixture = (companyId?: string) => createPluginWorkspaceFixture(db, companyId); + it.each([true, undefined])("wakes parked cleanup when a provider upgrade enables ancestry (%s)", async (supportsMergeAncestry) => { + const f = await fixture(); + await preparePluginWorkspace(db, f.workers, f.input); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + const svc = executionWorkspaceService(db, { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0 }); + await svc.sweepTerminalWorkspaces(100); + expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); + expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(false); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, (event, hints) => { + events.push(event); + if (event === "settled") svc.terminalWorkspaceWorkSettled(hints); + }); + try { + await pluginRegistryService(db).update(f.pluginId, { manifest: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry } } }); + expect(events).toEqual(["intent", "settled"]); + } finally { unsubscribe(); } + await svc.sweepTerminalWorkspaces(100); + expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(true); + expect(f.calls.some(call => call.operation.kind === "remove")).toBe(true); + }); + it("wakes cleanup only for provider contract changes and activation", async () => { + const f = await fixture(); + const registry = pluginRegistryService(db); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, event => events.push(event)); + try { + await registry.update(f.pluginId, { manifest: { ...plugin.manifestJson, description: "Updated description" } }); + await registry.updateStatus(f.pluginId, { status: "disabled" }); + expect(events).toEqual([]); + await registry.updateStatus(f.pluginId, { status: "ready" }); + expect(events.splice(0)).toEqual(["intent", "settled"]); + await registry.updateStatus(f.pluginId, { status: "ready" }); + await registry.upsertCompanySettings(f.pluginId, f.companyId, { enabled: false, settingsJson: {} }); + expect(events).toEqual([]); + await registry.upsertCompanySettings(f.pluginId, f.companyId, { enabled: true, settingsJson: {} }); + expect(events.splice(0)).toEqual(["intent", "settled"]); + await registry.upsertCompanySettings(f.pluginId, f.companyId, { enabled: true, settingsJson: {} }); + expect(events).toEqual([]); + await db.transaction(async tx => { + await pluginRegistryService(tx as unknown as typeof db).update(f.pluginId, { + manifest: { ...plugin.manifestJson, workspaceProvider: undefined }, + }); + expect(events).toEqual(["intent"]); + }); + expect(events.splice(0)).toEqual(["intent", "settled"]); + await registry.updateStatus(f.pluginId, { status: "disabled" }); + await registry.updateStatus(f.pluginId, { status: "ready" }); + await registry.update(f.pluginId, { manifest: { ...plugin.manifestJson, workspaceProvider: undefined, description: "Other plugin" } }); + expect(events).toEqual([]); + } finally { unsubscribe(); } + }); + it("keeps PR-only storage idle until delivery can qualify for fresh inspection", async () => { + const f = await fixture(); + const workspace = (await preparePluginWorkspace(db, f.workers, f.input))!; + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + const originalCall = f.workers.call.bind(f.workers); + f.workers.call = async (...args) => { + const value = await originalCall(...args); + return (args[2] as any).operation.kind === "inspect" ? { ...value as object, isMergedIntoBase: null } : value; + }; + let clock = new Date(); + let state: "open" | "merged" = "open"; + let headSha = "b".repeat(40); + let pullRequestReads = 0; + const options = { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0, + terminalWorkspaceInspectionBackoffMs: 300_000, now: () => clock, + resolvePullRequestDetails: async () => { pullRequestReads += 1; return { state, headRef: workspace.branchName, headSha }; } }; + const svc = executionWorkspaceService(db, options); + const inspections = () => f.calls.filter(call => call.operation.kind === "inspect").length; + await svc.sweepTerminalWorkspaces(100); + expect(inspections()).toBe(0); + expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); + svc.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`])); + await svc.sweepTerminalWorkspaces(100); + expect(inspections()).toBe(0); + expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); + await executionWorkspaceService(db, options).sweepTerminalWorkspaces(100); + expect(inspections()).toBe(0); + + await db.insert(issueWorkProducts).values({ companyId: f.companyId, issueId: f.issueId, + type: "pull_request", provider: "github", title: "Unrelated work", status: "active", + url: "https://github.com/other/repository/pull/1" }); + svc.terminalWorkspaceWorkSettled(new Set([`issue:${f.issueId}`])); + await svc.sweepTerminalWorkspaces(100); + expect(inspections()).toBe(0); + expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); + + await db.insert(issueWorkProducts).values({ companyId: f.companyId, issueId: f.issueId, + type: "pull_request", provider: "github", title: "Work", status: "active", + url: "https://github.com/example/repo/pull/1" }); + svc.terminalWorkspaceWorkSettled(new Set([`issue:${f.issueId}`])); + await svc.sweepTerminalWorkspaces(100); + expect(inspections()).toBe(0); + expect(await svc.nextTerminalWorkspaceSweepAt()).not.toBeNull(); + const readsBeforeUnrelatedWake = pullRequestReads; + svc.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`])); + await svc.sweepTerminalWorkspaces(100); + expect(pullRequestReads).toBe(readsBeforeUnrelatedWake); + expect(inspections()).toBe(0); + + state = "merged"; + clock = new Date(clock.getTime() + 300_001); + await svc.sweepTerminalWorkspaces(100); + expect(inspections()).toBe(1); + expect(f.calls.some(call => call.operation.kind === "remove")).toBe(false); + headSha = "a".repeat(40); + clock = new Date(clock.getTime() + 300_001); + await svc.sweepTerminalWorkspaces(100); + expect(f.calls.some(call => call.operation.kind === "remove")).toBe(true); + }); it("uses bounded finalization only for opted-in providers, preserving maintenance inspection", async () => { const f = await fixture(); const workspace = (await preparePluginWorkspace(db, f.workers, f.input))!; @@ -339,6 +462,24 @@ const support = await getEmbeddedPostgresTestSupport(); .where(eq(executionWorkspaces.id, reservation.id)); expect(recorded.providerRef).not.toBeNull(); }); + it("reconciles an unfinished reservation before applying PR-only cleanup eligibility", async () => { + const f = await fixture(); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId)); + f.lose(); + await expect(preparePluginWorkspace(db, f.workers, f.input)).rejects.toThrow("incomplete"); + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + f.dirty(); + await executionWorkspaceService(db, { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0 }).sweepTerminalWorkspaces(100); + expect(f.calls.filter(call => call.operation.kind === "prepare")).toHaveLength(2); + expect(f.calls.filter(call => call.operation.kind === "inspect")).toHaveLength(1); + expect(f.calls.some(call => call.operation.kind === "remove")).toBe(false); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, task.executionWorkspaceId!)); + expect(workspace.providerRef).not.toBeNull(); + }); it("fails closed for a disabled plugin, stopped run, and cross-company source", async () => { const f = await fixture(); await db.insert(pluginCompanySettings).values({ @@ -506,13 +647,14 @@ const support = await getEmbeddedPostgresTestSupport(); binding: { workspaceRef: restored.providerRef }, }); }); - it("uses provider inspection for remote workspace display and shares repeated reads", async () => { + it("reads workspace metadata without provider inspection", async () => { const f = await fixture(); const workspace = await preparePluginWorkspace(db, f.workers, f.input); await db .update(issues) .set({ status: "done" }) .where(eq(issues.id, f.issueId)); + f.workers.call = async () => { throw new Error("Storage unavailable"); }; const svc = executionWorkspaceService(db, { pluginWorkerManager: f.workers, inspectGitCloseReadiness: async () => { @@ -520,14 +662,14 @@ const support = await getEmbeddedPostgresTestSupport(); }, }); expect(await svc.getById(workspace!.id)).toMatchObject({ - deliveryState: "merged_by_ancestry", + deliveryState: "unknown", }); expect(await svc.getById(workspace!.id)).toMatchObject({ - deliveryState: "merged_by_ancestry", + deliveryState: "unknown", }); expect( f.calls.filter((call) => call.operation.kind === "inspect"), - ).toHaveLength(1); + ).toHaveLength(0); }); it("replays a durable removal after a lost result without inspecting the missing checkout", async () => { const f = await fixture(); @@ -641,6 +783,9 @@ const support = await getEmbeddedPostgresTestSupport(); }); it("retries an ambiguous deletion through the existing terminal sweep", async () => { const f = await fixture(); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId)); const workspace = await preparePluginWorkspace(db, f.workers, f.input); f.loseRemoval(); await db diff --git a/server/src/__tests__/workspace-git-inspection-cache.test.ts b/server/src/__tests__/workspace-git-inspection-cache.test.ts deleted file mode 100644 index c31b1b57be5..00000000000 --- a/server/src/__tests__/workspace-git-inspection-cache.test.ts +++ /dev/null @@ -1,47 +0,0 @@ -import type { ExecutionWorkspace } from "@paperclipai/shared"; -import { afterEach, expect, it, vi } from "vitest"; -import { createWorkspaceGitInspectionCache } from "../services/workspace-git-inspection-cache.js"; - -const workspace = { id: "workspace", companyId: "company", cwd: "/repo", baseRef: "master" } as ExecutionWorkspace; -afterEach(() => vi.useRealTimers()); - -it("coalesces concurrent display reads and expires after thirty seconds", async () => { - vi.useFakeTimers(); - const inspect = vi.fn(async () => ({ dirty: false })); - const read = createWorkspaceGitInspectionCache(inspect); - await Promise.all(Array.from({ length: 100 }, () => read(workspace))); - expect(inspect).toHaveBeenCalledTimes(1); - await read(workspace); - expect(inspect).toHaveBeenCalledTimes(1); - vi.advanceTimersByTime(29_999); - await read(workspace); - expect(inspect).toHaveBeenCalledTimes(1); - vi.advanceTimersByTime(1); - await read(workspace); - expect(inspect).toHaveBeenCalledTimes(2); - // Callers that authorize cleanup retain the uncached inspector. - await inspect(); - expect(inspect).toHaveBeenCalledTimes(3); -}); - -it("does not share results across companies, paths, base refs or workspace revisions", async () => { - const inspect = vi.fn(async () => null); - const read = createWorkspaceGitInspectionCache(inspect); - await read(workspace); - await read({ ...workspace, companyId: "other" }); - await read({ ...workspace, cwd: "/other" }); - await read({ ...workspace, baseRef: "other" }); - await read({ ...workspace, updatedAt: new Date() }); - expect(inspect).toHaveBeenCalledTimes(5); -}); - -it("retries failed inspections and bounds retained entries", async () => { - const inspect = vi.fn(async () => null).mockRejectedValueOnce(new Error("failed")); - const read = createWorkspaceGitInspectionCache(inspect); - await expect(read(workspace)).rejects.toThrow("failed"); - await read(workspace); - expect(inspect).toHaveBeenCalledTimes(2); - for (let i = 0; i < 256; i++) await read({ ...workspace, id: String(i) }); - await read(workspace); - expect(inspect).toHaveBeenCalledTimes(259); -}); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 15ff32f292e..b8048d2c976 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -1,6 +1,6 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { effectiveProjectWorkspacePolicy } from "./project-workspace-provider.js"; -import { inspectPluginWorkspace, removePluginWorkspace, restorePluginWorkspace, type WorkspaceWorkers } from "./plugin-workspaces.js"; +import { inspectPluginWorkspace, removePluginWorkspace, restorePluginWorkspace, pluginWorkspaceSupportsMergeAncestry, type WorkspaceWorkers } from "./plugin-workspaces.js"; import { acquireExecutionWorkspaceLifecycleLock, withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; import { measureRunTransaction } from "./execution-transaction-timing.js"; import { measureRunPhase } from "./execution-preparation-timing.js"; @@ -31,7 +31,6 @@ import { retryIdempotentDatabaseOperation } from "../database-retry.js"; import { taskWorkspaceSelectionSchema, type TaskWorkspaceSelection, type TaskWorkspaceIntent } from "@paperclipai/shared"; import { authorizationService, executionWorkspaceReadSqlCondition, projectReadSqlCondition, issueReadSqlCondition, type AuthorizationActor } from "./authorization.js"; import type { SQL } from "drizzle-orm"; -import { createWorkspaceGitInspectionCache } from "./workspace-git-inspection-cache.js"; import { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js"; import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; import { notifyTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./terminal-workspace-work.js"; @@ -340,10 +339,7 @@ export type ExecutionWorkspaceServiceOptions = { // Test hook. Runs after compaction moves directories out of the worktree and // before it re-checks the guard under the lifecycle lock. afterWorkspaceCompactionMove?: (workspace: ExecutionWorkspaceRow) => Promise; - inspectGitCloseReadiness?: (workspace: ExecutionWorkspace) => Promise<{ - git: ExecutionWorkspaceCloseGitReadiness | null; - warnings: string[]; - }>; + inspectGitCloseReadiness?: typeof inspectGitCloseReadiness; }; function parseGitHubRepository(repoUrl: string | null) { @@ -1370,8 +1366,6 @@ type WorkspaceOverviewIssueRow = WorkspaceOverviewLinkedIssue & { executionWorkspaceId: string; }; -const inspectGitForDisplay = createWorkspaceGitInspectionCache(inspectGitCloseReadiness); - async function hasPendingTaskWorkspaceRecovery(tx: DbTransaction, input: { companyId: string; issueId: string; excludeRunId?: string; }) { @@ -1431,7 +1425,7 @@ const WORKSPACE_COMPACTION_TRASH_PURGE_INTERVAL_MS = 30 * 60 * 1000; export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServiceOptions = {}) { async function inspectWorkspace(workspace: ExecutionWorkspace) { - if (workspace.providerType !== "plugin") return inspectGitCloseReadiness(workspace); + if (workspace.providerType !== "plugin") return (opts.inspectGitCloseReadiness ?? inspectGitCloseReadiness)(workspace); const [row] = await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, workspace.id), eq(executionWorkspaces.companyId, workspace.companyId))); if (!row) throw notFound("Workspace is unavailable"); let state; @@ -1729,13 +1723,6 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic ...(selectedWorkspace ? { projectWorkspaceId: selectedWorkspace.projectWorkspaceId } : {}) }; } - const inspectLocalDisplay = opts.inspectGitCloseReadiness - ? createWorkspaceGitInspectionCache(opts.inspectGitCloseReadiness) - : inspectGitForDisplay; - const inspectProviderDisplay = createWorkspaceGitInspectionCache(inspectWorkspace); - const inspectDisplay = (workspace: ExecutionWorkspace) => workspace.providerType === "plugin" - ? inspectProviderDisplay(workspace) - : inspectLocalDisplay(workspace); const recoveryActionsSvc = issueRecoveryActionService(db); const resolvePullRequestDetails = opts.resolvePullRequestDetails ?? createPullRequestMergeDetailsResolver(db); const now = opts.now ?? (() => new Date()); @@ -1915,14 +1902,14 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic ))); } - async function listDeliveryPullRequestProducts( - workspace: Pick, + async function listDeliveryPullRequestReferences( + workspace: ExecutionWorkspaceRow, ) { const sourceIssueId = workspace.sourceIssueId; if (!sourceIssueId) return []; // A read-only lookup: rebuild it after a transient pooled disconnect. - return retryIdempotentDatabaseOperation(() => db + const products = await retryIdempotentDatabaseOperation(() => db .select({ id: issueWorkProducts.id, url: issueWorkProducts.url, @@ -1939,6 +1926,27 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic )) .orderBy(desc(issueWorkProducts.updatedAt)) .limit(100)); + return products.flatMap(product => extractGitHubPullRequestReferences([ + product.url, product.externalId, product.title, product.summary, + product.metadata ? JSON.stringify(product.metadata) : null, + ])).filter(reference => pullRequestMatchesWorkspaceRepository(reference, workspace)); + } + + async function cachedDeliveryPullRequestDetails(workspace: ExecutionWorkspaceRow, reference: GitHubPullRequestReference) { + const key = `${workspace.companyId}:${reference.owner.toLowerCase()}/${reference.repo.toLowerCase()}#${reference.number}`; + const cached = pullRequestStateCache.get(key); + if (cached && now().getTime() - cached.checkedAtMs < pullRequestStateCacheTtlMs) return cached.details; + const details = await resolvePullRequestDetails(workspace.companyId, reference); + setBoundedPullRequestCacheEntry(pullRequestStateCache, key, { details, checkedAtMs: now().getTime() }); + return details; + } + + async function hasMergedDeliveryPullRequest(workspace: ExecutionWorkspaceRow, references: GitHubPullRequestReference[]) { + for (const reference of references) { + const details = await cachedDeliveryPullRequestDetails(workspace, reference); + if (details.state === "merged" && details.headRef === workspace.branchName && details.headSha) return true; + } + return false; } async function assessDelivery( @@ -1970,48 +1978,23 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic : null; if (sourceIssueTerminal) { - const products = await listDeliveryPullRequestProducts(workspace); - for (const product of products) { - const references = extractGitHubPullRequestReferences([ - product.url, - product.externalId, - product.title, - product.summary, - product.metadata ? JSON.stringify(product.metadata) : null, - ]); - if (references.length === 0) continue; - for (const reference of references) { - if (!pullRequestMatchesWorkspaceRepository(reference, workspace)) continue; - const key = `${workspace.companyId}:${reference.owner.toLowerCase()}/${reference.repo.toLowerCase()}#${reference.number}`; - const cached = pullRequestStateCache.get(key); - let details; - if (cached && now().getTime() - cached.checkedAtMs < pullRequestStateCacheTtlMs) { - details = cached.details; - } else { - details = await resolvePullRequestDetails(workspace.companyId, reference); - setBoundedPullRequestCacheEntry( - pullRequestStateCache, - key, - { details, checkedAtMs: now().getTime() }, - ); - } - if ( - details.state === "merged" - && details.headRef === workspace.branchName - && details.headSha === workspaceHeadSha - && workspaceHeadSha !== null - ) { - mergedPullRequest = true; - break; - } - if ( - details.state === "unknown" - || (details.state === "merged" && (!details.headRef || !details.headSha || !workspaceHeadSha)) - ) { - pullRequestStateUnknown = true; - } + for (const reference of await listDeliveryPullRequestReferences(workspace)) { + const details = await cachedDeliveryPullRequestDetails(workspace, reference); + if ( + details.state === "merged" + && details.headRef === workspace.branchName + && details.headSha === workspaceHeadSha + && workspaceHeadSha !== null + ) { + mergedPullRequest = true; + break; + } + if ( + details.state === "unknown" + || (details.state === "merged" && (!details.headRef || !details.headSha || !workspaceHeadSha)) + ) { + pullRequestStateUnknown = true; } - if (mergedPullRequest) break; } } @@ -2059,13 +2042,6 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } } - async function hydrateWorkspace(row: ExecutionWorkspaceRow, runtimeServices: WorkspaceRuntimeService[] = []) { - const workspace = toExecutionWorkspace(row, runtimeServices); - const { git } = await inspectDisplay(workspace); - const assessment = await assessDelivery(row, git); - return toExecutionWorkspace(row, runtimeServices, assessment.deliveryState); - } - async function workspaceHasActiveRun(workspace: Pick) { const linkedIssues = await db .select({ @@ -3059,7 +3035,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic projectWorkspaceId: row.projectWorkspaceId, }); const runtimeServicesByWorkspaceId = await loadEffectiveRuntimeServicesByExecutionWorkspace(db, row.companyId, [row]); - return hydrateWorkspace( + return toExecutionWorkspace( row, (runtimeServicesByWorkspaceId.get(row.id) ?? []).map(toRuntimeService), ); @@ -3691,6 +3667,25 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic else result.skippedActiveRun += 1; continue; } + let deliveryReferences: GitHubPullRequestReference[] | null = null; + if (workspace.providerType === "plugin" && !isClosedExecutionWorkspaceStatus(workspace.status) + && !reopenPending && !workspace.metadata?.workspaceProviderRemoval && !workspace.metadata?.workspaceProviderRestoration) { + try { + if (!(await pluginWorkspaceSupportsMergeAncestry(db, opts.pluginWorkerManager, workspace))) { + deliveryReferences = await listDeliveryPullRequestReferences(workspace); + if (deliveryReferences.length === 0) { + result.skippedUndelivered += 1; + if (compactionPending && compactedTerminalAt.get(workspace.id) !== latestTerminalAt) + terminalWorkspaceDeadlines.set(workspace.id, compactionDueAt); + else terminalWorkspaceDeadlines.delete(workspace.id); + continue; + } + } + } catch { + result.skippedUndelivered += 1; + continue; + } + } if (opts.terminalWorkspaceInspectionBackoffMs) { const revision = createHash("sha256").update(JSON.stringify([ workspace.updatedAt, workspace.metadata, @@ -3707,6 +3702,17 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic nextAt: now().getTime() + opts.terminalWorkspaceInspectionBackoffMs, }); } + if (deliveryReferences !== null) { + try { + if (!(await hasMergedDeliveryPullRequest(workspace, deliveryReferences))) { + result.skippedUndelivered += 1; + continue; + } + } catch { + result.skippedUndelivered += 1; + continue; + } + } const workspacePath = readNullableString(workspace.providerRef) ?? readNullableString(workspace.cwd); const createdByRuntime = workspace.providerType === "git_worktree" ? isRuntimeOwnedGitBranch(workspace.metadata) diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index 239c3c50490..4be581f4394 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -88,6 +88,14 @@ export function pluginRegistryService(db: Db) { .then((rows) => rows[0] ?? null); } + async function writeWorkspaceProviderChange(reconsiderWorkspaces: boolean, write: (database: Db) => Promise): Promise { + if (!reconsiderWorkspaces) return write(db); + return db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.terminalWorkspaces); + return write(tx as unknown as Db); + }); + } + async function nextInstallOrder(): Promise { const result = await db .select({ maxOrder: sql`coalesce(max(${plugins.installOrder}), 0)` }) @@ -239,12 +247,14 @@ export function pluginRegistryService(db: Db) { setClause.categories = data.manifest.categories; } - return db + const workspaceProviderChanged = data.manifest !== undefined + && !isDeepStrictEqual(plugin.manifestJson.workspaceProvider, data.manifest.workspaceProvider); + return writeWorkspaceProviderChange(workspaceProviderChanged, database => database .update(plugins) .set(setClause) .where(eq(plugins.id, id)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); }, // ----- Status --------------------------------------------------------- @@ -254,7 +264,9 @@ export function pluginRegistryService(db: Db) { const plugin = await getById(id); if (!plugin) throw notFound("Plugin not found"); - return db + const workspaceProviderActivated = Boolean(plugin.manifestJson.workspaceProvider) + && plugin.status !== "ready" && input.status === "ready"; + return writeWorkspaceProviderChange(workspaceProviderActivated, database => database .update(plugins) .set({ status: input.status, @@ -263,7 +275,7 @@ export function pluginRegistryService(db: Db) { }) .where(eq(plugins.id, id)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); }, // ----- Uninstall / Remove -------------------------------------------- @@ -479,6 +491,10 @@ export function pluginRegistryService(db: Db) { )) .then((rows) => rows[0] ?? null); + if (plugin.manifestJson.workspaceProvider && existing?.enabled === false && input.enabled === true) { + await notifyDeliveryWork(db, DELIVERY_QUEUES.terminalWorkspaces); + } + if (existing) { return db .update(pluginCompanySettings) diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index ca04b081e48..09316a0be45 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -70,6 +70,13 @@ async function provider( return plugin; } +export async function pluginWorkspaceSupportsMergeAncestry(db: Db, workers: WorkspaceWorkers | undefined, workspace: Workspace) { + const pluginId = readWorkspaceProviderBinding(workspace.metadata)?.pluginId; + if (!pluginId) return true; + const selected = await provider(db, workers, workspace.companyId, pluginId); + return selected.manifestJson.workspaceProvider?.supportsMergeAncestry !== false; +} + /** No paths or provider credentials are accepted from task callers. */ export async function callWorkspaceProvider( db: Db, diff --git a/server/src/services/workspace-git-inspection-cache.ts b/server/src/services/workspace-git-inspection-cache.ts deleted file mode 100644 index 6fad32e4766..00000000000 --- a/server/src/services/workspace-git-inspection-cache.ts +++ /dev/null @@ -1,30 +0,0 @@ -import type { ExecutionWorkspace } from "@paperclipai/shared"; - -/** Short-lived display cache only. Destructive operations must inspect afresh. */ -export function createWorkspaceGitInspectionCache(inspect: (workspace: ExecutionWorkspace) => Promise) { - const entries = new Map }>(); - return (workspace: ExecutionWorkspace): Promise => { - const key = JSON.stringify([ - workspace.companyId, workspace.id, workspace.updatedAt, workspace.providerType, - workspace.providerRef, workspace.cwd, workspace.repoUrl, workspace.baseRef, - workspace.branchName, workspace.metadata, - ]); - const now = Date.now(); - const existing = entries.get(key); - if (existing && existing.expiresAt > now) return existing.promise; - for (const [candidate, entry] of entries) { - if (entry.expiresAt <= now) entries.delete(candidate); - } - if (entries.size >= 256) entries.delete(entries.keys().next().value!); - const entry = { expiresAt: Number.POSITIVE_INFINITY, promise: Promise.resolve().then(() => inspect(workspace)) }; - entries.set(key, entry); - entry.promise = entry.promise.then((result) => { - entry.expiresAt = Date.now() + 30_000; - return result; - }, (error) => { - if (entries.get(key) === entry) entries.delete(key); - throw error; - }); - return entry.promise; - }; -} From c32ecf665705559f3e58acb175971ec5178a959a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:44:46 -0500 Subject: [PATCH 2/2] fix(server): schedule only outstanding merge confirmations (#18382) Reviewed and verified at bc22051c56b9bef7d492e88ad8f6b5dcd773f6f5. Co-Authored-By: Paperclip --- ...026-10-30-merged-pr-confirmations-sleep.md | 132 ++++++ .../external-objects-service.test.ts | 9 +- .../merged-pr-confirmation-sweep.test.ts | 11 +- .../merged-pr-confirmation-work.test.ts | 444 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 20 +- server/src/index.ts | 17 +- .../chat-confirmation-notifications.ts | 2 + .../services/delivery-work-notifications.ts | 1 + server/src/services/external-objects.ts | 1 + .../github-external-object-provider.ts | 18 +- .../src/services/github-pull-request-merge.ts | 17 +- .../src/services/issue-thread-interactions.ts | 125 +++-- server/src/services/issues.ts | 4 + .../merged-pr-confirmation-notifications.ts | 7 + .../services/merged-pr-confirmation-work.ts | 38 ++ 15 files changed, 761 insertions(+), 85 deletions(-) create mode 100644 doc/plans/2026-10-30-merged-pr-confirmations-sleep.md create mode 100644 server/src/__tests__/merged-pr-confirmation-work.test.ts create mode 100644 server/src/services/merged-pr-confirmation-notifications.ts create mode 100644 server/src/services/merged-pr-confirmation-work.ts diff --git a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md new file mode 100644 index 00000000000..c22c826f22e --- /dev/null +++ b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md @@ -0,0 +1,132 @@ +# Merged pull request confirmation scheduling + +The merged-PR confirmation duty previously ran from every heartbeat maintenance +callback, including when there were no merge requests. It loaded all pending +confirmation cards and then applied the merge-only parser in memory. + +The interaction service still owns merge-only eligibility, provider access, +automatic acceptance, current resolver authority, audit, and continuation. +The worker registers the existing scoped-deadline helper with a dedicated delivery +topic. Shared scheduling receives interaction/company identities and deadlines; +it receives no GitHub payloads. No database migration is required. + +## Durable truth and producers + +Pending `request_confirmation` interaction rows remain the durable work source. +The candidate query pages pending cards in bounded groups of 100 and excludes +terminal and hidden issues and linked tool actions. The existing fail-closed +parser is the sole text eligibility source: it requires merge intent and a PR +reference and excludes additional actions, governed actions, and document/attachment targets. +No accepted or cancelled history keeps a deadline alive. A full page without any +merge candidates still advances its keyset cursor; later candidates are not lost. + +Ordinary creation registers the topic before inserting a confirmation in its +transaction. Existing confirmation-resolution notifications invalidate its issue +before acceptance, rejection, withdrawal, supersession, and administrative expiry. +Issue status and visibility changes invalidate that issue before their update. +The signal helper joins caller-owned outer transactions; nested savepoint release +is not a commit. Startup reconstructs existing candidates. Direct tool, secret, +and OAuth interaction inserts do not create eligible merge-only work. Company +portability does not import pending interaction rows. + +The supported writer model is the instance's ordinary service/API process. +Independent SQL writers must use the same transaction-aware notification contract +or explicitly invalidate/restart the owning worker. Process-local notifications +do not establish a cross-replica ingress guarantee. No permanent database listener +or empty recovery scan is added. + +## External observation and retries + +The existing normalized `applyPullRequestEvent` to +`sweepMergedPullRequestConfirmations` seam remains available. A normalized merge +hint checks matching references in its company; unrelated tenants and PRs do not +cause provider reads. Existing provider/binding authorization remains unchanged. + +A pending candidate can reference a PR whose external changes have no dependable +webhook coverage. Each such candidate retains a five-minute observation deadline. +Resolution, withdrawal, supersession, hiding, or terminal issue state removes the +obligation. Provider/read failures and unknown states retain the candidate, using +the existing scoped helper's 30-second exponential backoff capped at five minutes. +These retries do not scan completed history. Removing candidate reconciliation +requires a proven durable delivery/replay guarantee for every referenced PR, +including the first event after idle and supported non-Cloud installations. + +This is candidate-backed external observation. It does not eliminate all polling, +register a Cloud wake alarm, or prove that the host/database suspends. Existing +sleep-safety blockers remain in force. + +Startup registers this observation without waiting for live GitHub requests. +The internal resolver invocation accepts an optional host-owned abort signal; +that signal is not part of external-object records or plugin/RPC payloads. +The GitHub owner combines cancellation with a ten-second request budget and +passes it to the actual object fetch and repository-visibility probes, including +response body reads. Shutdown aborts the request and waits for its cancellation; +no post-abort acceptance is committed. A timeout leaves the durable card pending. + +## Acceptance evidence + +Focused tests use real PostgreSQL, the actual service producers, transaction +signals, coordinator, scheduler, acceptance transaction, and persisted audit and +chat continuation outbox. GitHub state and the direct heartbeat wake callback are +controlled simulation boundaries. The tests do not claim live GitHub delivery or +real agent execution. + +Cases cover empty/configured idle and retained terminal history with no recurring +SQL/provider work across old cadence windows; the first ordinary creation; +outer commit/rollback; exact candidate and failure deadlines; withdrawal; +provider failure across drain; standby/resume; restart; tenant-scoped and duplicate +merge hints; latest/superseded cards during provider observation; hidden/terminal +issues; URL targets; no-continuation policy; and shutdown during an in-flight read. +Existing startup and normalized GitHub event tests exercise the changed consumers. +Controlled local HTTP tests hold headers, partial response bodies, 404 probes, +and public-visibility probes. They observe server-side response close after +shutdown within one second, rather than only racing a promise. A shortened +request budget times out a real request; the saved card then accepts after +restart when that same HTTP fixture serves a merged response. Existing GitHub +provider and external-object service tests preserve their authorization behavior. + +## Verification follow-up + +The original head passed full-workspace typecheck and build. A later required +Linux CI shard failed the unchanged managed-skill maintenance CLI fixture in +`managed-skill-content.test.ts`: it returned `unavailable_or_corrupt_package` +where `local_storage` was expected. Both the isolated case and the complete +unchanged 17-test file pass against fresh local PostgreSQL. The original Linux cause is +not established; that failure is retained and is not called a flake. + +After merging current master, the same fixture failed locally because the new +managed-Cloud policy from PR #18350 correctly refuses local disk. A separate +test-only prerequisite commit changes that fixture to the real S3 provider with +a controlled HTTP endpoint, valid fixture credentials, and a stable test key. +It verifies remote reads without seeding or mutating runtime configuration, +rejects corrupt and missing objects, and recovers after restoring canonical bytes. +The full existing 17-test file passes with these failure controls. Production +configuration validation remains unchanged. Current-head required CI remains +the completion gate. + +## Cancellation during acceptance + +The registered worker passes its optional cancellation signal into the canonical +confirmation acceptance transaction. It rechecks after the issue lock, before +mutation and before commit. Cancellation while an awaited write is blocked rolls +back the interaction, audit and transactional delivery effects. Manual callers +without a signal retain their existing acceptance and rollback behavior. + +A PostgreSQL lock wait itself remains unabortable by this signal: stop waits for +the query to unwind. Once the lock is released, cancelled acceptance leaves the +card pending for restart. The existing HTTP request timeout and prompt stop abort +remain unchanged. Actual lock and write-trigger controls verify pending state, +no post-stop wake, and exactly one acceptance/audit/wake on restart. + +## Existing limitations outside this migration + +Direct non-chat continuation still uses the existing postcommit `wakeup` call. +If that call fails after acceptance commits, the previous periodic sweep also +skipped the accepted row and did not reconstruct the missing wake. This migration +preserves that behavior; a general durable interaction-continuation contract is +a separate follow-up. Real chat bindings retain the existing transactional +`interaction_wakeup` outbox, verified through ordinary creation and acceptance. + +The separate review-authority issue tracked in public issue #17704 and PR #17711 +remains owned there. The worker isolates candidate failures and preserves current +review authorization; it does not grant a system actor new review authority. diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index e9022599779..af3b4667deb 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; @@ -26,7 +27,8 @@ import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -709,8 +711,8 @@ describeEmbeddedPostgres("externalObjectService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); + db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 }); }, 20_000); afterEach(async () => { @@ -725,6 +727,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/merged-pr-confirmation-sweep.test.ts b/server/src/__tests__/merged-pr-confirmation-sweep.test.ts index 1e4d56436e7..d9cf9880357 100644 --- a/server/src/__tests__/merged-pr-confirmation-sweep.test.ts +++ b/server/src/__tests__/merged-pr-confirmation-sweep.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { inArray } from "drizzle-orm"; @@ -131,7 +132,8 @@ describe("merged pull-request confirmation extraction", () => { }); }); -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MERGE_SWEEP_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("merged pull-request confirmation sweep", () => { @@ -139,8 +141,8 @@ describeEmbeddedPostgres("merged pull-request confirmation sweep", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-merged-pr-confirmations-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-merged-pr-confirmations-"); + db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 }); }, 20_000); afterEach(async () => { @@ -153,6 +155,7 @@ describeEmbeddedPostgres("merged pull-request confirmation sweep", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -276,7 +279,7 @@ describeEmbeddedPostgres("merged pull-request confirmation sweep", () => { repo: "paperclip", number: 39, }])).resolves.toEqual({ - checked: 6, + checked: 3, candidates: 3, accepted: 2, woken: 2, diff --git a/server/src/__tests__/merged-pr-confirmation-work.test.ts b/server/src/__tests__/merged-pr-confirmation-work.test.ts new file mode 100644 index 00000000000..60d1a5ffcb7 --- /dev/null +++ b/server/src/__tests__/merged-pr-confirmation-work.test.ts @@ -0,0 +1,444 @@ +import { createServer } from "node:http"; +import { createPullRequestMergeStateResolver } from "../services/github-pull-request-merge.js"; +import { randomUUID } from "node:crypto"; +import { setTimeout as delay } from "node:timers/promises"; +import { eq, sql } from "drizzle-orm"; +import { activityLog, agents, chatActions, chatConversations, chatEndpoints, chatPublications, companies, createDb, issues, issueThreadInteractions, toolApplications, toolConnections, type Db } from "@paperclipai/db"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; +import { issueService } from "../services/issues.js"; +import { registerMergeConfirmationWork, MERGE_CONFIRMATION_RECONCILE_MS } from "../services/merged-pr-confirmation-work.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MERGE_WORK_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("merge confirmation work with PostgreSQL and simulated GitHub", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let companyId: string, issueId: string, agentId: string; + let coordinator: ReturnType | undefined; + let draining = false, standby = false; + const errors: unknown[] = []; + const provider = vi.fn(async (_companyId: string, _reference: { number: number }, _signal?: AbortSignal): Promise<"open" | "merged" | "unknown"> => "merged"); + const wakeup = vi.fn(async () => ({ id: "simulated-durable-wake" })); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("merged-pr-work-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + errors.length = 0; draining = false; standby = false; + provider.mockReset().mockResolvedValue("merged"); wakeup.mockClear(); + companyId = randomUUID(); agentId = randomUUID(); issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Merge worker", issuePrefix: `M${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle", adapterType: "codex_local" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Merge task", status: "todo", assigneeAgentId: agentId }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(chatActions); await db.delete(chatPublications); await db.delete(chatConversations); await db.delete(chatEndpoints); + await db.delete(toolConnections); await db.delete(toolApplications); + await db.delete(activityLog); await db.delete(issueThreadInteractions); await db.delete(issues); + await db.delete(agents); await db.delete(companies); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + const service = (database: Db = db) => issueThreadInteractionService(database, { resolvePullRequestState: provider, wakeup }); + const input = (number = 39) => ({ kind: "request_confirmation" as const, title: `Merge paperclipai/paperclip#${number}?`, + resolverPolicy: "human_only" as const, continuationPolicy: "wake_assignee_on_accept" as const, + payload: { version: 1 as const, prompt: "Merge the PR?" } }); + const create = (database: Db = db, number = 39) => service(database).create({ id: issueId, companyId }, input(number), { agentId }); + const stored = (id: string) => db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, id)).then(rows => rows[0]); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining && !standby, + canReconcile: () => !standby, onError: error => errors.push(error) }); + const worker = registerMergeConfirmationWork(coordinator, service(), () => !draining && !standby); + await worker.ready; + return worker; + } + async function waitFor(check: () => Promise | void) { + for (let attempt = 0; attempt < 100; attempt++) { + await vi.advanceTimersByTimeAsync(1000); + await delay(10); + try { await check(); return; } catch (error) { if (attempt === 99) throw error; } + } + } + const settle = () => waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + async function quiet() { + await waitFor(() => { expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); }); + const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); + const calls = provider.mock.calls.length; + await vi.advanceTimersByTimeAsync(3 * MERGE_CONFIRMATION_RECONCILE_MS); + expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + expect(provider).toHaveBeenCalledTimes(calls); + reads.mockRestore(); transactions.mockRestore(); executes.mockRestore(); + } + + it("settles empty and configured idle, then ordinary creation accepts, audits, wakes once and becomes quiet with retained history", async () => { + await start(); await quiet(); expect(provider).not.toHaveBeenCalled(); + const card = await create(); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); + await quiet(); + expect(wakeup).toHaveBeenCalledTimes(1); + expect(wakeup).toHaveBeenCalledWith(agentId, expect.objectContaining({ idempotencyKey: `interaction:${card.id}:accepted`, allowRunCoalescing: false })); + expect(await db.select().from(activityLog)).toEqual(expect.arrayContaining([expect.objectContaining({ + action: "issue.thread_interaction_accepted", actorId: "system:pr-merged", details: expect.objectContaining({ interactionId: card.id }), + })])); + expect(errors).toEqual([]); + }); + + it("reconciles an open candidate at its deadline, never early, then stops after merge", async () => { + provider.mockResolvedValue("open"); await start(); const card = await create(); + await waitFor(() => expect(provider).toHaveBeenCalledTimes(1)); await settle(); + const dueAt = coordinator!.nextWakeAt()!; + expect(dueAt).toBeGreaterThan(Date.now()); + await vi.advanceTimersByTimeAsync(dueAt - Date.now() - 1); expect(provider).toHaveBeenCalledTimes(1); + provider.mockResolvedValue("merged"); await vi.advanceTimersByTimeAsync(1); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(provider).toHaveBeenCalledTimes(2); expect(wakeup).toHaveBeenCalledTimes(1); + }); + + it("manual cancellation removes the candidate deadline without provider work", async () => { + provider.mockResolvedValue("open"); await start(); const card = await create(); + await waitFor(() => expect(provider).toHaveBeenCalledTimes(1)); await settle(); + await service().withdrawInteraction({ id: issueId, companyId }, card.id, {}, { agentId }); + await quiet(); expect(provider).toHaveBeenCalledTimes(1); expect(wakeup).not.toHaveBeenCalled(); + expect((await stored(card.id)).status).toBe("cancelled"); + }); + + it("does not accept before caller-owned outer commit and forgets a rolled-back create", async () => { + await start(); await quiet(); + let release!: () => void, created!: () => void; + const barrier = new Promise(resolve => { release = resolve; }); + const inserted = new Promise(resolve => { created = resolve; }); + let id = ""; + const transaction = db.transaction(async tx => { + id = (await create(tx as unknown as Db)).id; created(); await barrier; + }); + await inserted; + await vi.advanceTimersByTimeAsync(10_000); expect(provider).not.toHaveBeenCalled(); + expect(await stored(id)).toBeUndefined(); release(); await transaction; + await waitFor(async () => expect((await stored(id)).status).toBe("accepted")); await quiet(); + const prior = provider.mock.calls.length; + await expect(db.transaction(async tx => { await create(tx as unknown as Db, 41); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await quiet(); expect(provider).toHaveBeenCalledTimes(prior); expect(errors).toEqual([]); + }); + + it("retains provider failures across drain, resumes without overlap, and clears retry after success", async () => { + provider.mockRejectedValue(new Error("GitHub unavailable")); await start(); const card = await create(); + await waitFor(() => expect(errors).toHaveLength(1)); await settle(); + expect((await stored(card.id)).status).toBe("pending"); + draining = true; + const reads = vi.spyOn(db, "select"), calls = provider.mock.calls.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(calls); reads.mockRestore(); + draining = false; provider.mockResolvedValue("merged"); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(wakeup).toHaveBeenCalledTimes(1); + }); + + it("keeps first work committed during drain and startup during standby until resume", async () => { + standby = true; await start(); const card = await create(); + await vi.advanceTimersByTimeAsync(60_000); expect(provider).not.toHaveBeenCalled(); + standby = false; draining = true; + await vi.advanceTimersByTimeAsync(60_000); expect(provider).not.toHaveBeenCalled(); + draining = false; + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + }); + + it("reconstructs pending cards on restart and ignores terminal or hidden tasks", async () => { + provider.mockResolvedValue("open"); const card = await create(); await start(); + await waitFor(() => expect(provider).toHaveBeenCalledTimes(1)); await settle(); + await coordinator!.stop(); coordinator = undefined; provider.mockResolvedValue("merged"); + await start(); await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + const hidden = await db.transaction(async tx => { + const card = await create(tx as unknown as Db, 41); + await issueService(tx as unknown as Db).update(issueId, { hiddenAt: new Date() }); + return card; + }); + await quiet(); expect((await stored(hidden.id)).status).toBe("pending"); + await issueService(db).update(issueId, { hiddenAt: null }); + await waitFor(async () => expect((await stored(hidden.id)).status).toBe("accepted")); await quiet(); + }); + + it("preserves a real chat binding's durable continuation outbox on automatic acceptance", async () => { + const applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID(); + await db.insert(toolApplications).values({ id: applicationId, companyId, applicationKey: `chat:slack:${endpointId}`, name: "Slack", type: "chat", status: "active" }); + await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Slack", uid: endpointId, + connectionPurpose: "channel", transport: "chat_sdk", status: "active", enabled: true }); + await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "slack", publicId: endpointId, + assignedAgentId: agentId, status: "active" }); + const [conversation] = await db.insert(chatConversations).values({ companyId, endpointId, issueId, + externalConversationId: "merge-channel", externalThreadId: "merge-thread", externalLabel: "Merge thread", state: "active" }).returning(); + await start(); await quiet(); const card = await create(); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(await db.select().from(chatActions).where(eq(chatActions.kind, "interaction_wakeup"))).toEqual([ + expect.objectContaining({ companyId, endpointId, conversationId: conversation.id, providerActionId: `interaction_wakeup:${card.id}`, + payload: expect.objectContaining({ interactionId: card.id, issueId, agentId, interactionStatus: "accepted" }) }), + ]); + expect(await db.select().from(chatPublications)).toEqual(expect.arrayContaining([ + expect.objectContaining({ idempotencyKey: `interaction:${card.id}:${endpointId}`, state: "cancelled" }), + ])); + expect(errors).toEqual([]); + }); + + it("does not lose a newer card created during provider observation or wake the superseded card", async () => { + let release!: () => void, observed!: () => void; + const barrier = new Promise(resolve => { release = resolve; }); + const observing = new Promise(resolve => { observed = resolve; }); + provider.mockImplementationOnce(async () => { observed(); await barrier; return "merged"; }); + await start(); const old = await create(); + await vi.advanceTimersByTimeAsync(1000); await observing; + const current = await create(db, 41); + expect((await stored(old.id)).status).toBe("expired"); + release(); + await waitFor(async () => expect((await stored(current.id)).status).toBe("accepted")); await quiet(); + expect(wakeup).toHaveBeenCalledTimes(1); + expect(wakeup).toHaveBeenCalledWith(agentId, expect.objectContaining({ idempotencyKey: `interaction:${current.id}:accepted` })); + expect(errors).toEqual([]); + }); + + it("keeps unknown provider results as retryable debt and does not act before retry", async () => { + provider.mockResolvedValue("unknown"); await start(); const card = await create(); + await waitFor(() => expect(errors).toHaveLength(1)); + const next = coordinator!.nextWakeAt()!; + await vi.advanceTimersByTimeAsync(next - Date.now() - 1); expect(provider).toHaveBeenCalledTimes(1); + provider.mockResolvedValue("merged"); await vi.advanceTimersByTimeAsync(1); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(provider).toHaveBeenCalledTimes(2); + }); + + it("honors terminal issues, pure merge URL targets and continuationPolicy none", async () => { + await start(); await quiet(); + const target = await service().create({ id: issueId, companyId }, { ...input(), title: "Merge the linked PR?", continuationPolicy: "none", + payload: { version: 1, prompt: "Merge the PR?", target: { type: "custom", key: "github-pr-39", href: "https://github.com/paperclipai/paperclip/pull/39" } } }, { agentId }); + await waitFor(async () => expect((await stored(target.id)).status).toBe("accepted")); await quiet(); + expect(wakeup).not.toHaveBeenCalled(); + provider.mockClear(); + const terminal = await db.transaction(async tx => { + const card = await create(tx as unknown as Db, 41); + await issueService(tx as unknown as Db).update(issueId, { status: "cancelled" }); + return card; + }); + await quiet(); expect((await stored(terminal.id)).status).toBe("expired"); + expect(provider).not.toHaveBeenCalled(); expect(wakeup).not.toHaveBeenCalled(); + }); + + it("rechecks hidden state under the acceptance lock when visibility changes during a provider read", async () => { + let release!: () => void, observed!: () => void; + const barrier = new Promise(resolve => { release = resolve; }); + const observing = new Promise(resolve => { observed = resolve; }); + provider.mockImplementationOnce(async () => { observed(); await barrier; return "merged"; }); + await start(); const card = await create(); await vi.advanceTimersByTimeAsync(1000); await observing; + await issueService(db).update(issueId, { hiddenAt: new Date() }); release(); await quiet(); + expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled(); + await issueService(db).update(issueId, { hiddenAt: null }); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + }); + + it("unrelated pending confirmations do not retain worker SQL or provider work", async () => { + await start(); await quiet(); + await service().create({ id: issueId, companyId }, { ...input(), title: "Approve the plan?", payload: { version: 1, prompt: "Approve the plan?" } }, { agentId }); + await quiet(); expect(provider).not.toHaveBeenCalled(); expect(wakeup).not.toHaveBeenCalled(); + expect(errors).toEqual([]); + }); + + it("preserves manual acceptance without cancellation and rolls back a failed caller mutation", async () => { + const card = await create(); + const issue = { id: issueId, companyId, projectId: null, goalId: null }; + await expect(service().acceptInteraction(issue, card.id, {}, { userId: "board" }, { + afterResolveInTransaction: async () => { throw new Error("manual acceptance rollback"); }, + })).rejects.toThrow("manual acceptance rollback"); + expect((await stored(card.id)).status).toBe("pending"); + const accepted = await service().acceptInteraction(issue, card.id, {}, { userId: "board" }); + expect(accepted.interaction.status).toBe("accepted"); + expect((await stored(card.id)).status).toBe("accepted"); + }); + + it.each(["issue lock", "acceptance write"] as const)("keeps a stopped %s acceptance pending and accepts once after restart", async phase => { + vi.useRealTimers(); + const card = await create(); + let signal: AbortSignal | undefined; + provider.mockImplementation(async (_companyId, _reference, currentSignal) => { signal = currentSignal; return "merged"; }); + let reserved: Awaited> | undefined = await db.$client.reserve(); + const [{ pid }] = await reserved<{ pid: number }[]>`select pg_backend_pid() as pid`; + if (phase === "issue lock") { + await reserved.unsafe("begin"); + await reserved`select id from issues where id=${issueId} for update`; + } else { + await db.$client.unsafe(`create function maintenance_hold_merge_acceptance() returns trigger language plpgsql as $$begin + if new.id = '${card.id}'::uuid and new.status = 'accepted' then perform pg_advisory_xact_lock(930582); end if; return new; end$$`); + await db.$client.unsafe(`create trigger maintenance_hold_merge_acceptance before update on issue_thread_interactions for each row execute function maintenance_hold_merge_acceptance()`); + await reserved`select pg_advisory_lock(930582)`; + } + let stopping: Promise | undefined; + try { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = registerMergeConfirmationWork(coordinator, service(), () => true); + let blocked: { query: string } | undefined; + for (let attempt = 0; attempt < 200; attempt++) { + [blocked] = await db.$client<{ query: string }[]>`select query from pg_stat_activity + where datname=current_database() and wait_event_type='Lock' and ${pid} = any(pg_blocking_pids(pid))`; + if (blocked) break; + await delay(10); + } + expect(blocked?.query).toContain(phase === "issue lock" ? '"issues"' : '"issue_thread_interactions"'); + let stopped = false; + stopping = coordinator.stop().then(() => { stopped = true; }); + await delay(20); + expect(signal?.aborted).toBe(true); + expect(stopped).toBe(false); + expect((await stored(card.id)).status).toBe("pending"); + const releasedAt = performance.now(); + if (phase === "issue lock") await reserved.unsafe("commit"); + else await reserved`select pg_advisory_unlock(930582)`; + reserved.release(); reserved = undefined; + await stopping; await worker.ready; coordinator = undefined; + expect(performance.now() - releasedAt).toBeLessThan(1000); + expect((await stored(card.id)).status).toBe("pending"); + expect(wakeup).not.toHaveBeenCalled(); + expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(0); + expect(await db.select().from(chatActions)).toHaveLength(0); + await start(); + expect((await stored(card.id)).status).toBe("accepted"); + expect(wakeup).toHaveBeenCalledTimes(1); + expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(1); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + await quiet(); + expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + } finally { + if (reserved) { + if (phase === "issue lock") await reserved.unsafe("rollback"); + else await reserved`select pg_advisory_unlock(930582)`; + reserved.release(); + } + await stopping; + await coordinator?.stop(); coordinator = undefined; + if (phase === "acceptance write") { + await db.$client.unsafe("drop trigger if exists maintenance_hold_merge_acceptance on issue_thread_interactions"); + await db.$client.unsafe("drop function if exists maintenance_hold_merge_acceptance()"); + } + } + }); + + it("awaits an in-flight observation at shutdown and reconstructs its unaccepted card on restart", async () => { + let release!: () => void, observed!: () => void; + const barrier = new Promise(resolve => { release = resolve; }); + const observing = new Promise(resolve => { observed = resolve; }); + provider.mockImplementationOnce(async () => { observed(); await barrier; return "merged"; }); + await start(); const card = await create(); await vi.advanceTimersByTimeAsync(1000); await observing; + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await delay(10); expect(stopped).toBe(false); release(); await stopping; coordinator = undefined; + expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled(); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(2 * MERGE_CONFIRMATION_RECONCILE_MS); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await start(); await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + }); + + it.each(["headers", "body", "not-found probe", "public visibility probe"] as const)("cancels an actual unresponsive HTTP %s read promptly at stop", async phase => { + vi.useRealTimers(); + const card = await create(); + let observed!: () => void, closed!: () => void; + const requested = new Promise(resolve => { observed = resolve; }); + const disconnected = new Promise(resolve => { closed = resolve; }); + const signals: AbortSignal[] = []; + const server = createServer((request, response) => { + if (phase === "not-found probe" && request.url?.includes("/pulls/")) { response.writeHead(404).end(); return; } + response.on("close", closed); + if (phase === "body") { response.writeHead(200, { "content-type": "application/json" }); response.write('{"merged":'); } + observed(); + }); + 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 test HTTP address"); + const resolveState = createPullRequestMergeStateResolver(db, { + requestTimeoutMs: 10_000, + credentialProvider: () => phase === "public visibility probe" ? { token: "fixture", source: "server_env", secretName: null } : null, + fetch: (url, init) => { signals.push(init!.signal!); return fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init); }, + }); + const real = issueThreadInteractionService(db, { resolvePullRequestState: resolveState, wakeup }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const startupAt = performance.now(); + const worker = registerMergeConfirmationWork(coordinator, real, () => true); + expect(performance.now() - startupAt).toBeLessThan(100); + await Promise.race([requested, delay(2000).then(() => { throw new Error("HTTP observation did not start"); })]); + const stoppingAt = performance.now(); + await coordinator.stop(); coordinator = undefined; + expect(performance.now() - stoppingAt).toBeLessThan(1000); + await worker.ready; + await Promise.race([disconnected, delay(1000).then(() => { throw new Error("HTTP request remained open after stop"); })]); + expect(signals.at(-1)?.aborted).toBe(true); + expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled(); expect(errors).toEqual([]); + } finally { + await coordinator?.stop(); coordinator = undefined; + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); + } + }); + + it("times out a real unresponsive HTTP read and recovers its durable candidate on restart", async () => { + vi.useRealTimers(); const card = await create(); + let serveMerge = false, closed!: () => void; + const disconnected = new Promise(resolve => { closed = resolve; }); + const server = createServer((_request, response) => { + if (serveMerge) { response.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify({ merged: true, state: "closed" })); return; } + response.on("close", closed); + }); + 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 test HTTP address"); + const real = issueThreadInteractionService(db, { resolvePullRequestState: createPullRequestMergeStateResolver(db, { + requestTimeoutMs: 100, tokenProvider: () => null, + fetch: (url, init) => fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init), + }), wakeup }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const startedAt = performance.now(); + await registerMergeConfirmationWork(coordinator, real, () => true).ready; + expect(performance.now() - startedAt).toBeLessThan(2000); + await Promise.race([disconnected, delay(1000).then(() => { throw new Error("Timed out HTTP request remained open"); })]); + expect(errors).toHaveLength(1); expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled(); + expect(coordinator.nextWakeAt()).not.toBeNull(); + await coordinator.stop(); coordinator = undefined; serveMerge = true; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerMergeConfirmationWork(coordinator, real, () => true).ready; + expect((await stored(card.id)).status).toBe("accepted"); expect(wakeup).toHaveBeenCalledTimes(1); + } finally { + await coordinator?.stop(); coordinator = undefined; + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); + } + }); + + it("pages beyond a full pending-card page without duplicating the merge parser in SQL", async () => { + const card = await create(); + const id = BigInt(`0x${card.id.replaceAll("-", "")}`); + await db.insert(issueThreadInteractions).values(Array.from({ length: 100 }, (_, index) => { + const hex = (id - BigInt(index + 1)).toString(16).padStart(32, "0"); + return { id: `${hex.slice(0,8)}-${hex.slice(8,12)}-${hex.slice(12,16)}-${hex.slice(16,20)}-${hex.slice(20)}`, + companyId, issueId, kind: "request_confirmation", title: "Approve the plan?", payload: { version: 1 as const, prompt: "Approve the plan?" } }; + })); + await start(); expect(provider).not.toHaveBeenCalled(); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(provider).toHaveBeenCalledTimes(1); expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + }); + + it("limits normalized merge hints to the owning company and matching PR and tolerates duplicates", async () => { + provider.mockResolvedValue("open"); const own = await create(); + const foreignCompany = randomUUID(), foreignIssue = randomUUID(); + await db.insert(companies).values({ id: foreignCompany, name: "Other company", issuePrefix: "OTH" }); + await db.insert(issues).values({ id: foreignIssue, companyId: foreignCompany, title: "Other", status: "todo" }); + const foreign = await service().create({ id: foreignIssue, companyId: foreignCompany }, input(), { userId: "board" }); + const local = service(); + const hint = [{ companyId, owner: "PAPERCLIPAI", repo: "paperclip", number: 39 }]; + expect((await local.sweepMergedPullRequestConfirmations(hint)).accepted).toBe(1); + expect((await stored(own.id)).status).toBe("accepted"); expect((await stored(foreign.id)).status).toBe("pending"); + expect(provider).not.toHaveBeenCalled(); + expect((await local.sweepMergedPullRequestConfirmations(hint)).accepted).toBe(0); expect(wakeup).toHaveBeenCalledTimes(1); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index cf49503dfbf..b78ec44664d 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -186,6 +186,8 @@ const { const heartbeatServiceFactoryMock = vi.fn(() => heartbeatServiceMock); const issueThreadInteractionServiceMock = { sweepSupersededPendingRequestConfirmations: vi.fn(async () => ({ expired: 0 })), + mergedPullRequestConfirmationPage: vi.fn(async () => ({ rows: [], complete: true })), + mergedPullRequestConfirmationCandidates: vi.fn(async () => []), sweepMergedPullRequestConfirmations: vi.fn(async () => ({ checked: 0, candidates: 0, @@ -1488,7 +1490,9 @@ describe("startServer feedback export wiring", () => { expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); - expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); + expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalledTimes(1); + expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).not.toHaveBeenCalled(); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.mergedPullRequestConfirmations)).toBe(true); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); const routineWorkerIndex = deliveryWorkRegisterMock.mock.calls.findIndex(([queue]) => queue === DELIVERY_QUEUES.routineSchedules); @@ -1575,6 +1579,20 @@ describe("startServer feedback export wiring", () => { } }); + it("becomes ready without waiting for the merge-confirmation observation", async () => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); + let release!: () => void; + const blocked = new Promise<{ rows: never[]; complete: boolean }>(resolve => { release = () => resolve({ rows: [], complete: true }); }); + issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage.mockReturnValueOnce(blocked); + const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(() => 1 as unknown as ReturnType); + let started = false; + const starting = startServer().then(server => { started = true; return server; }); + try { + await vi.waitFor(() => expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalled()); + await settleStartupHousekeeping(); expect(started).toBe(true); expect(getStartupRecoveryState().phase).toBe("ready"); + } finally { release(); await starting; timer.mockRestore(); } + }); + it.each(["retained cleanup", "workspace finalization", "environment settlement", "native run projection", "native result commit", "native recovery notices", "external conversation settlement"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); let releaseCleanup!: () => void; diff --git a/server/src/index.ts b/server/src/index.ts index c6e420b2748..0cd8308747c 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerMergeConfirmationWork } from "./services/merged-pr-confirmation-work.js"; import { registerRoutineExecutionStatusWork } from "./services/routine-execution-status-work.js"; import { routineExecutionStatusService } from "./services/routine-run-status.js"; import { attentionRetentionService, registerAttentionRetentionWork } from "./services/attention-retention.js"; @@ -1593,19 +1594,8 @@ async function startServerWithDatabaseTeardown( : config.workspaceCompactionDelayHours * 60 * 60 * 1000, terminalWorkspaceInspectionBackoffMs: TERMINAL_WORKSPACE_RETRY_MS, }); - const scheduleMergedPullRequestConfirmationSweep = () => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(mergedPullRequestConfirmations - .sweepMergedPullRequestConfirmations() - .then((result) => { - if (result.accepted > 0) { - logger.info(result, "accepted merge confirmations for merged pull requests"); - } - }) - .catch((err) => { - logger.error({ err }, "merged pull-request confirmation sweep failed"); - })); - }; + registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); const gitRestorePointWorker = registerGitRestorePointWork(deliveryWork, gitRestorePointRetentionService(db as any, { storage: storageService, retentionDays: config.gitRestorePointRetentionDays, @@ -1946,7 +1936,6 @@ async function startServerWithDatabaseTeardown( scheduleExternalObjectRefreshSweep(new Date()); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - scheduleMergedPullRequestConfirmationSweep(); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); diff --git a/server/src/services/chat-confirmation-notifications.ts b/server/src/services/chat-confirmation-notifications.ts index e55d8c0d7ac..02cdfbee8c8 100644 --- a/server/src/services/chat-confirmation-notifications.ts +++ b/server/src/services/chat-confirmation-notifications.ts @@ -1,3 +1,4 @@ +import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type ChatConfirmationScope = { companyId: string; issueId?: string }; @@ -10,6 +11,7 @@ export async function notifyChatConfirmationWork(tx: object, scope: ChatConfirma /** Interaction resolution can release a callback parked while its human input * was pending. Register on the owning transaction before its resolution write. */ export async function notifyChatConfirmationResolution(tx: object, interaction: ChatConfirmationScope & { kind: string }) { + if (interaction.kind === "request_confirmation") await notifyMergeConfirmationWork(tx, interaction); if (["request_confirmation", "request_checkbox_confirmation", "suggest_tasks", "connection_intent"].includes(interaction.kind)) { await notifyChatConfirmationWork(tx, interaction); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 654244eb1bf..562c9ed4f50 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -110,6 +110,7 @@ export const DELIVERY_QUEUES = { browser: "browser-use", email: "email-channels", museMaintenance: "muse-maintenance", + mergedPullRequestConfirmations: "merged-pull-request-confirmations", mcpEvents: "mcp-events", dotEvents: "dot-events", grokDispatch: "grok-dispatch", diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index 56a8fb91834..306f043d753 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -84,6 +84,7 @@ export interface ExternalObjectResolver { resolve(input: { companyId: string; object: ExternalObjectRecord; + signal?: AbortSignal; }): Promise; } diff --git a/server/src/services/github-external-object-provider.ts b/server/src/services/github-external-object-provider.ts index 6dedac97efb..83b47354035 100644 --- a/server/src/services/github-external-object-provider.ts +++ b/server/src/services/github-external-object-provider.ts @@ -15,6 +15,7 @@ type FetchLike = (url: string, init?: RequestInit) => Promise; export interface GitHubExternalObjectProviderOptions { fetch?: FetchLike; + requestTimeoutMs?: number; tokenProvider?: (companyId: string) => Promise | string | null; /** * Resolve the credential and its source. Prefer this over `tokenProvider` when the @@ -243,12 +244,12 @@ type RepoProbeOutcome = { async function probeRepoVisibility( identity: GitHubObjectIdentity, - opts: { fetch: FetchLike; headers?: Record | undefined }, + opts: { fetch: FetchLike; headers?: Record | undefined; signal?: AbortSignal }, ): Promise { try { const probe = await opts.fetch( `${gitHubApiBase(identity.host)}/repos/${encodeURIComponent(identity.owner)}/${encodeURIComponent(identity.repo)}`, - { headers: opts.headers }, + { headers: opts.headers, signal: opts.signal }, ); if (probe.ok) { const body = await safeJson(probe); @@ -427,12 +428,13 @@ export function createGitHubExternalObjectProvider( companyId: string, identity: GitHubObjectIdentity, headers: Record, + signal: AbortSignal, ): Promise { const credentialFingerprint = createHash("sha256").update(headers.authorization ?? "anonymous").digest("hex"); const key = `${companyId}\u0000${credentialFingerprint}\u0000${identity.owner.toLowerCase()}\u0000${identity.repo.toLowerCase()}`; const cached = probeCache.get(key); if (cached && cached.expiresAt > Date.now()) return cached.outcome; - const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (outcome.visibility === "visible" || outcome.visibility === "hidden") { probeCache.set(key, { outcome, expiresAt: Date.now() + GITHUB_OBJECT_TTL_SECONDS * 1000 }); } @@ -464,7 +466,9 @@ export function createGitHubExternalObjectProvider( return { providerKey: "github", objectType, - async resolve({ companyId, object }) { + async resolve({ companyId, object, signal: ownerSignal }) { + const deadline = AbortSignal.timeout(opts.requestTimeoutMs ?? 10_000); + const signal = ownerSignal ? AbortSignal.any([ownerSignal, deadline]) : deadline; const identity = parseGitHubObject(object); if (!identity || identity.objectType !== objectType) { return { @@ -509,7 +513,7 @@ export function createGitHubExternalObjectProvider( // "public" verdict must not authorize a later read of its private contents. // The cache is still used below to classify a 404, where a stale verdict can // only make the object look deleted — it can never expose private data. - const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (probe.visibility === "hidden") { return { ok: false, @@ -554,7 +558,7 @@ export function createGitHubExternalObjectProvider( let response: Response; try { - response = await fetchImpl(url, { headers }); + response = await fetchImpl(url, { headers, signal }); } catch { return { ok: false, @@ -573,7 +577,7 @@ export function createGitHubExternalObjectProvider( // Probe the repository metadata to classify: if the repo itself is visible // the object truly is gone; if it is not, surface credentials-needed // instead of permanently freezing a reference as "Not found". - const probe = await cachedProbeRepoVisibility(companyId, identity, headers); + const probe = await cachedProbeRepoVisibility(companyId, identity, headers, signal); if (probe.visibility === "visible") { return { ok: true, snapshot: notFoundSnapshot(identity, etag) }; } diff --git a/server/src/services/github-pull-request-merge.ts b/server/src/services/github-pull-request-merge.ts index bb92d37d7c2..1cdff01dc1f 100644 --- a/server/src/services/github-pull-request-merge.ts +++ b/server/src/services/github-pull-request-merge.ts @@ -1,5 +1,5 @@ import type { Db } from "@paperclipai/db"; -import { createGitHubExternalObjectProvider } from "./github-external-object-provider.js"; +import { createGitHubExternalObjectProvider, type GitHubExternalObjectProviderOptions } from "./github-external-object-provider.js"; export type GitHubPullRequestReference = { host: "github.com"; @@ -25,11 +25,13 @@ export type PullRequestMergeDetails = { export type PullRequestMergeStateResolver = ( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ) => Promise; export type PullRequestMergeDetailsResolver = ( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ) => Promise; export const PULL_REQUEST_CACHE_MAX_ENTRIES = 1_000; @@ -86,14 +88,15 @@ function readRecord(value: unknown): Record | null { : null; } -export function createPullRequestMergeDetailsResolver(db: Db): PullRequestMergeDetailsResolver { - const resolver = createGitHubExternalObjectProvider(db).resolvers +export function createPullRequestMergeDetailsResolver(db: Db, options?: GitHubExternalObjectProviderOptions): PullRequestMergeDetailsResolver { + const resolver = createGitHubExternalObjectProvider(db, options).resolvers .find((candidate) => candidate.objectType === "pull_request") ?? null; - return async (companyId, reference) => { + return async (companyId, reference, signal) => { if (!resolver) return { state: "unknown", headRef: null, headSha: null }; const result = await resolver.resolve({ companyId, + signal, object: { externalId: `${reference.owner}/${reference.repo}#pull/${reference.number}`, sanitizedCanonicalUrl: `https://github.com/${reference.owner}/${reference.repo}/pull/${reference.number}`, @@ -123,7 +126,7 @@ export function createPullRequestMergeDetailsResolver(db: Db): PullRequestMergeD }; } -export function createPullRequestMergeStateResolver(db: Db): PullRequestMergeStateResolver { - const resolveDetails = createPullRequestMergeDetailsResolver(db); - return async (companyId, reference) => (await resolveDetails(companyId, reference)).state; +export function createPullRequestMergeStateResolver(db: Db, options?: GitHubExternalObjectProviderOptions): PullRequestMergeStateResolver { + const resolveDetails = createPullRequestMergeDetailsResolver(db, options); + return async (companyId, reference, signal) => (await resolveDetails(companyId, reference, signal)).state; } diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index f4d5abbc5e2..194cddbfee9 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1,3 +1,4 @@ +import { notifyMergeConfirmationWork, type MergeConfirmationScope } from "./merged-pr-confirmation-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyIssueReviewState } from "./issue-review-state-notifications.js"; @@ -204,6 +205,7 @@ export type IssueThreadInteractionServiceOptions = { resolvePullRequestState?: ( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ) => Promise; wakeup?: InteractionWakeup; pullRequestCacheTtlMs?: number; @@ -502,6 +504,7 @@ type IssueResolutionContext = { companyId: string; status: string; executionRunId: string | null; + hiddenAt?: Date | null; workMode: string; assigneeAgentId: string | null; assigneeUserId: string | null; @@ -2044,10 +2047,11 @@ export function issueThreadInteractionService( async function resolvePullRequestState( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ): Promise { if (opts.resolvePullRequestState) - return opts.resolvePullRequestState(companyId, reference); - return defaultPullRequestStateResolver?.(companyId, reference) ?? "unknown"; + return opts.resolvePullRequestState(companyId, reference, signal); + return defaultPullRequestStateResolver?.(companyId, reference, signal) ?? "unknown"; } async function resolvePullRequestStates( @@ -2056,6 +2060,7 @@ export function issueThreadInteractionService( companyId: string; reference: GitHubPullRequestReference; }>, + signal?: AbortSignal, ) { const states = new Map(); const pending = entries.slice(); @@ -2063,11 +2068,13 @@ export function issueThreadInteractionService( await Promise.all( Array.from({ length: workerCount }, async () => { while (pending.length > 0) { + signal?.throwIfAborted(); const entry = pending.shift(); if (!entry) return; const state = await resolvePullRequestState( entry.companyId, entry.reference, + signal, ); states.set(entry.key, state); } @@ -2075,6 +2082,38 @@ export function issueThreadInteractionService( ); return states; } + async function mergedPullRequestConfirmationPage(scope?: MergeConfirmationScope, after?: string) { + const rows = await db.select({ interaction: issueThreadInteractions, issue: { + id: issues.id, companyId: issues.companyId, projectId: issues.projectId, goalId: issues.goalId, + status: issues.status, assigneeAgentId: issues.assigneeAgentId, + } }).from(issueThreadInteractions).innerJoin(issues, and( + eq(issueThreadInteractions.issueId, issues.id), eq(issueThreadInteractions.companyId, issues.companyId), + )).where(and( + eq(issueThreadInteractions.kind, "request_confirmation"), eq(issueThreadInteractions.status, "pending"), + isNull(issues.hiddenAt), sql`${issues.status} NOT IN ('done', 'cancelled')`, + scope ? eq(issueThreadInteractions.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(issueThreadInteractions.issueId, scope.issueId) : undefined, + scope?.interactionId ? eq(issueThreadInteractions.id, scope.interactionId) : undefined, + after ? sql`${issueThreadInteractions.id} > ${after}::uuid` : undefined, + sql`NOT EXISTS (SELECT 1 FROM ${toolActionRequests} WHERE ${toolActionRequests.interactionId} = ${issueThreadInteractions.id})`, + )).orderBy(asc(issueThreadInteractions.id)).limit(100); + return { rows: rows.flatMap(row => { + const references = getMergeConfirmationPullRequestReferences(row.interaction); + return references.length ? [{ ...row, references }] : []; + }), after: rows.at(-1)?.interaction.id, complete: rows.length < 100 }; + } + + async function mergedPullRequestConfirmationCandidates(scope?: MergeConfirmationScope) { + const candidates: Awaited>["rows"] = []; + let after: string | undefined; + while (true) { + const page = await mergedPullRequestConfirmationPage(scope, after); + candidates.push(...page.rows); + if (page.complete) return candidates; + after = page.after; + } + } + async function getIdempotentInteraction(args: { issueId: string; companyId: string; @@ -2239,10 +2278,12 @@ export function issueThreadInteractionService( input: AcceptIssueThreadInteraction; actor: InteractionActor; mutationOptions?: InteractionResolutionMutationOptions; + signal?: AbortSignal; }): Promise<{ interaction: IssueThreadInteraction; continuationIssue: IssueWakeTarget | null; }> { + args.signal?.throwIfAborted(); const expired = await expireStaleRequestConfirmationTarget(db, { row: args.current, actor: args.actor, @@ -2272,6 +2313,7 @@ export function issueThreadInteractionService( companyId: issues.companyId, status: issues.status, executionRunId: issues.executionRunId, + hiddenAt: issues.hiddenAt, workMode: issues.workMode, assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, @@ -2284,6 +2326,7 @@ export function issueThreadInteractionService( .for("update") .then((rows: IssueResolutionContext[]) => rows[0] ?? null); + args.signal?.throwIfAborted(); if (!issueContext || issueContext.companyId !== args.issue.companyId) { throw notFound("Issue not found"); } @@ -2310,6 +2353,10 @@ export function issueThreadInteractionService( "Interaction has already been resolved", ); } + if (args.actor.systemId === "system:pr-merged" && + (issueContext.hiddenAt || getMergeConfirmationPullRequestReferences(lockedCurrent).length === 0)) { + throw conflict("Merge confirmation is no longer actionable"); + } await assertRequestConfirmationResolutionAllowedUnderLock( tx as unknown as Db, issueContext, @@ -2332,6 +2379,7 @@ export function issueThreadInteractionService( : undefined; if (lockedCurrent.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, lockedCurrent.companyId, [lockedCurrent.id]); + args.signal?.throwIfAborted(); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -2503,6 +2551,7 @@ export function issueThreadInteractionService( }); } + args.signal?.throwIfAborted(); return { interaction: acceptedInteraction, continuationIssue, @@ -2980,6 +3029,8 @@ export function issueThreadInteractionService( await emitInteractionResolvedTelemetry(db, interaction); return interaction; }, + mergedPullRequestConfirmationPage, + mergedPullRequestConfirmationCandidates, sweepMergedPullRequestConfirmations: async ( mergedHints: Array<{ companyId: string; @@ -2987,51 +3038,17 @@ export function issueThreadInteractionService( repo: string; number: number; }> = [], + scope?: MergeConfirmationScope, + signal?: AbortSignal, ) => { - const rows = await db - .select({ - interaction: issueThreadInteractions, - issue: { - id: issues.id, - companyId: issues.companyId, - projectId: issues.projectId, - goalId: issues.goalId, - status: issues.status, - assigneeAgentId: issues.assigneeAgentId, - }, - }) - .from(issueThreadInteractions) - .innerJoin(issues, eq(issueThreadInteractions.issueId, issues.id)) - .where( - and( - eq(issueThreadInteractions.kind, "request_confirmation"), - eq(issueThreadInteractions.status, "pending"), - ), - ); - - const candidates = rows.flatMap((row) => { - const references = getMergeConfirmationPullRequestReferences( - row.interaction, - ); - return references.length > 0 ? [{ ...row, references }] : []; - }); - if (candidates.length === 0) { - return { checked: rows.length, candidates: 0, accepted: 0, woken: 0 }; - } - - const candidateIds = candidates.map(({ interaction }) => interaction.id); - const linkedToolActions = await db - .select({ interactionId: toolActionRequests.interactionId }) - .from(toolActionRequests) - .where(inArray(toolActionRequests.interactionId, candidateIds)); - const toolActionInteractionIds = new Set( - linkedToolActions - .map((row) => row.interactionId) - .filter((value): value is string => Boolean(value)), - ); - const eligible = candidates.filter( - ({ interaction }) => !toolActionInteractionIds.has(interaction.id), - ); + const candidates = await mergedPullRequestConfirmationCandidates(scope ?? + (mergedHints.length === 1 ? { companyId: mergedHints[0]!.companyId } : undefined)); + const eligible = mergedHints.length ? candidates.filter(candidate => mergedHints.some(hint => + hint.companyId === candidate.issue.companyId && candidate.references.some(reference => + hint.owner.toLowerCase() === reference.owner.toLowerCase() && + hint.repo.toLowerCase() === reference.repo.toLowerCase() && hint.number === reference.number), + )) : candidates; + if (eligible.length === 0) return { checked: 0, candidates: 0, accepted: 0, woken: 0 }; const checkedAt = now().getTime(); const cacheTtlMs = opts.pullRequestCacheTtlMs ?? 5 * 60 * 1000; @@ -3065,7 +3082,10 @@ export function issueThreadInteractionService( const refreshedStates = await resolvePullRequestStates([ ...uniqueReferences.values(), - ]); + ], signal); + if (scope && [...refreshedStates.values()].some(state => state === "unknown")) { + throw new Error("Merge confirmation provider state is unavailable"); + } for (const [key, state] of refreshedStates) { setBoundedPullRequestCacheEntry(pullRequestStateCache, key, { state, @@ -3076,6 +3096,7 @@ export function issueThreadInteractionService( let accepted = 0; let woken = 0; for (const candidate of eligible) { + if (signal?.aborted) break; const allMerged = candidate.references.every((reference) => { const key = `${candidate.issue.companyId}:${reference.owner.toLowerCase()}/${reference.repo.toLowerCase()}#${reference.number}`; return pullRequestStateCache.get(key)?.state === "merged"; @@ -3087,6 +3108,7 @@ export function issueThreadInteractionService( resolved = await acceptRequestConfirmation({ issue: candidate.issue, current: candidate.interaction, + signal, input: {}, actor: { systemId: "system:pr-merged", @@ -3112,12 +3134,16 @@ export function issueThreadInteractionService( if (resolved.interaction.status !== "accepted") continue; accepted += 1; - const wakeIssue = resolved.continuationIssue ?? candidate.issue; + const wakeIssue = resolved.continuationIssue ?? await db.select({ + id: issues.id, assigneeAgentId: issues.assigneeAgentId, status: issues.status, + }).from(issues).where(and(eq(issues.id, candidate.issue.id), + eq(issues.companyId, candidate.issue.companyId), isNull(issues.hiddenAt))).then(rows => rows[0]); const shouldWake = resolved.interaction.continuationPolicy === "wake_assignee" || resolved.interaction.continuationPolicy === "wake_assignee_on_accept"; if ( !opts.wakeup || + !wakeIssue || !shouldWake || !wakeIssue.assigneeAgentId || isTerminalIssueStatus(wakeIssue.status) @@ -3159,7 +3185,7 @@ export function issueThreadInteractionService( } return { - checked: rows.length, + checked: candidates.length, candidates: eligible.length, accepted, woken, @@ -3906,6 +3932,7 @@ export function issueThreadInteractionService( } else { await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id }); } + if (data.kind === "request_confirmation") await notifyMergeConfirmationWork(tx, { companyId: issue.companyId, issueId: issue.id }); const [row] = await tx .insert(issueThreadInteractions) .values({ diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 1bbdacbbbc2..d15ff12cde6 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js"; import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import { notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; @@ -11543,6 +11544,9 @@ export function issueService(db: Db) { || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); } + if (patch.status !== undefined || patch.hiddenAt !== undefined) { + await notifyMergeConfirmationWork(tx, { companyId: receiptExisting.companyId, issueId: id }); + } await notifyIssueAttentionRetentionWork(tx, { companyId: receiptExisting.companyId, issueId: id, parentId: patch.parentId, originKind: patch.originKind, originId: patch.originId }); if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined || patch.parentId !== undefined diff --git a/server/src/services/merged-pr-confirmation-notifications.ts b/server/src/services/merged-pr-confirmation-notifications.ts new file mode 100644 index 00000000000..34df4ecffb0 --- /dev/null +++ b/server/src/services/merged-pr-confirmation-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type MergeConfirmationScope = { companyId: string; issueId?: string; interactionId?: string }; + +export function notifyMergeConfirmationWork(tx: object, scope: MergeConfirmationScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.mergedPullRequestConfirmations, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, interactionId: scope.interactionId })); +} diff --git a/server/src/services/merged-pr-confirmation-work.ts b/server/src/services/merged-pr-confirmation-work.ts new file mode 100644 index 00000000000..6a754d5a88a --- /dev/null +++ b/server/src/services/merged-pr-confirmation-work.ts @@ -0,0 +1,38 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { issueThreadInteractionService } from "./issue-thread-interactions.js"; +import type { MergeConfirmationScope } from "./merged-pr-confirmation-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; + +export const MERGE_CONFIRMATION_RECONCILE_MS = 5 * 60_000; + +export function registerMergeConfirmationWork( + coordinator: ReturnType, + service: Pick, + "mergedPullRequestConfirmationCandidates" | "mergedPullRequestConfirmationPage" | "sweepMergedPullRequestConfirmations">, + canRun: () => boolean, +) { + const cursors = new Map(); + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.mergedPullRequestConfirmations, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as MergeConfirmationScope, + matches: (entry, scope?: MergeConfirmationScope) => !scope || (entry.companyId === scope.companyId && + (!scope.issueId || entry.issueId === scope.issueId) && (!scope.interactionId || entry.id === scope.interactionId)), + load: async (scope?: MergeConfirmationScope) => { + const key = scope ? JSON.stringify(scope) : "*"; + const page = await service.mergedPullRequestConfirmationPage(scope, cursors.get(key)); + if (page.complete) cursors.delete(key); + else cursors.set(key, page.after!); + return { complete: page.complete, rows: page.rows.map(({ interaction }) => ({ + id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: Date.now(), + })) }; + }, + check: async (entry, signal) => { + const scope = { companyId: entry.companyId, issueId: entry.issueId, interactionId: entry.id }; + await service.sweepMergedPullRequestConfirmations([], scope, signal); + if (signal?.aborted) return entry; + const remaining = await service.mergedPullRequestConfirmationCandidates(scope); + return remaining.length ? { ...entry, dueAt: Date.now() + MERGE_CONFIRMATION_RECONCILE_MS } : null; + }, + }, canRun); +}