From e55bc2df65d2a73125d1ed4c02fff69bc67c078a Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:19:27 -0500 Subject: [PATCH 1/4] fix(server): schedule only outstanding merge confirmations Co-Authored-By: Paperclip --- ...026-10-30-merged-pr-confirmations-sleep.md | 83 ++++++ .../merged-pr-confirmation-sweep.test.ts | 11 +- .../merged-pr-confirmation-work.test.ts | 278 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 5 +- server/src/index.ts | 17 +- .../chat-confirmation-notifications.ts | 2 + .../services/delivery-work-notifications.ts | 1 + .../src/services/issue-thread-interactions.ts | 101 ++++--- server/src/services/issues.ts | 4 + .../merged-pr-confirmation-notifications.ts | 7 + .../services/merged-pr-confirmation-work.ts | 30 ++ 11 files changed, 474 insertions(+), 65 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..fb8b091f841 --- /dev/null +++ b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md @@ -0,0 +1,83 @@ +# 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 excludes terminal and hidden issues, linked tool actions, +and text without merge intent or a PR reference. The existing fail-closed parser +then excludes additional actions, governed actions, and document/attachment targets. +No accepted or cancelled history keeps a deadline alive. + +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. + +## 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. + +## 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__/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..51195e6157b --- /dev/null +++ b/server/src/__tests__/merged-pr-confirmation-work.test.ts @@ -0,0 +1,278 @@ +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 }): 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("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("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 b645d7437c8..9fc7234b2e6 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -180,6 +180,7 @@ const { const heartbeatServiceFactoryMock = vi.fn(() => heartbeatServiceMock); const issueThreadInteractionServiceMock = { sweepSupersededPendingRequestConfirmations: vi.fn(async () => ({ expired: 0 })), + mergedPullRequestConfirmationCandidates: vi.fn(async () => []), sweepMergedPullRequestConfirmations: vi.fn(async () => ({ checked: 0, candidates: 0, @@ -1334,7 +1335,9 @@ describe("startServer feedback export wiring", () => { expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); - expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); + expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationCandidates).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); diff --git a/server/src/index.ts b/server/src/index.ts index ae229f03518..c1804998ef5 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 { registerProviderTraceRetentionWork } from "./services/provider-trace-retention-work.js"; import { registerRunEvidenceMigrationWork } from "./services/run-evidence-migration-work.js"; import { registerSkillVersionRetentionWork } from "./services/skill-version-retention-work.js"; @@ -1600,19 +1601,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"); - })); - }; + await registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; const gitRestorePointWorker = registerGitRestorePointWork(deliveryWork, gitRestorePointRetentionService(db as any, { storage: storageService, retentionDays: config.gitRestorePointRetentionDays, @@ -2005,7 +1995,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 9195f1fcb12..308a721cbc5 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -97,6 +97,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/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 59c14618342..b306f637e27 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 { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyIssueReviewState } from "./issue-review-state-notifications.js"; import { notifyChatConfirmationResolution } from "./chat-confirmation-notifications.js"; @@ -501,6 +502,7 @@ type IssueResolutionContext = { companyId: string; status: string; executionRunId: string | null; + hiddenAt?: Date | null; workMode: string; assigneeAgentId: string | null; assigneeUserId: string | null; @@ -2074,6 +2076,32 @@ export function issueThreadInteractionService( ); return states; } + async function mergedPullRequestConfirmationCandidates(scope?: MergeConfirmationScope) { + const text = sql`concat_ws(' ', ${issueThreadInteractions.title}, ${issueThreadInteractions.summary}, + ${issueThreadInteractions.payload}->>'prompt', ${issueThreadInteractions.payload}->>'acceptLabel', + ${issueThreadInteractions.payload}->>'detailsMarkdown', ${issueThreadInteractions.payload}->'target'->>'key', + ${issueThreadInteractions.payload}->'target'->>'label', ${issueThreadInteractions.payload}->'target'->>'href')`; + 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, + sql`${text} ~* '(merge|merging)'`, + sql`${text} ~* '(github\\.com/[^ /]+/[^ /]+/pull/[1-9][0-9]*|[a-z0-9_.-]+/[a-z0-9_.-]+#[1-9][0-9]*)'`, + sql`NOT EXISTS (SELECT 1 FROM ${toolActionRequests} WHERE ${toolActionRequests.interactionId} = ${issueThreadInteractions.id})`, + )); + return rows.flatMap(row => { + const references = getMergeConfirmationPullRequestReferences(row.interaction); + return references.length ? [{ ...row, references }] : []; + }); + } + async function getIdempotentInteraction(args: { issueId: string; companyId: string; @@ -2271,6 +2299,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, @@ -2309,6 +2338,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, @@ -2979,6 +3012,7 @@ export function issueThreadInteractionService( await emitInteractionResolvedTelemetry(db, interaction); return interaction; }, + mergedPullRequestConfirmationCandidates, sweepMergedPullRequestConfirmations: async ( mergedHints: Array<{ companyId: string; @@ -2986,51 +3020,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,6 +3065,9 @@ export function issueThreadInteractionService( const refreshedStates = await resolvePullRequestStates([ ...uniqueReferences.values(), ]); + 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, @@ -3075,6 +3078,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"; @@ -3111,12 +3115,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) @@ -3158,7 +3166,7 @@ export function issueThreadInteractionService( } return { - checked: rows.length, + checked: candidates.length, candidates: eligible.length, accepted, woken, @@ -3909,6 +3917,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 42c65bb5d02..4178dc1a35c 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 { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -11491,6 +11492,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 }); + } if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); 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..bfccaeea79f --- /dev/null +++ b/server/src/services/merged-pr-confirmation-work.ts @@ -0,0 +1,30 @@ +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" | "sweepMergedPullRequestConfirmations">, + canRun: () => boolean, +) { + 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) => (await service.mergedPullRequestConfirmationCandidates(scope)) + .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); +} From 7edb163a80d1aca60e6c780c9472befa523fcd5a Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:49:17 -0500 Subject: [PATCH 2/4] fix(server): cancel merge observation HTTP requests on shutdown Bound GitHub object and repository probes, keep readiness independent of external reads, and page pending cards through the owning parser. Co-Authored-By: Paperclip --- ...026-10-30-merged-pr-confirmations-sleep.md | 34 ++++++- .../external-objects-service.test.ts | 9 +- .../merged-pr-confirmation-work.test.ts | 88 +++++++++++++++++++ .../server-startup-feedback-export.test.ts | 17 +++- server/src/index.ts | 4 +- 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 | 38 +++++--- .../services/merged-pr-confirmation-work.ts | 14 ++- 10 files changed, 200 insertions(+), 40 deletions(-) diff --git a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md index fb8b091f841..7b8699456fb 100644 --- a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md +++ b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md @@ -13,10 +13,12 @@ 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 excludes terminal and hidden issues, linked tool actions, -and text without merge intent or a PR reference. The existing fail-closed parser -then excludes additional actions, governed actions, and document/attachment targets. -No accepted or cancelled history keeps a deadline alive. +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 @@ -53,6 +55,14 @@ 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 @@ -68,6 +78,22 @@ provider failure across drain; standby/resume; restart; tenant-scoped and duplic 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 Linux cause is +not established; this is retained as an unresolved CI failure, not called a flake. +The current-head required CI result remains the completion gate. ## Existing limitations outside this migration 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-work.test.ts b/server/src/__tests__/merged-pr-confirmation-work.test.ts index 51195e6157b..95565fd96e7 100644 --- a/server/src/__tests__/merged-pr-confirmation-work.test.ts +++ b/server/src/__tests__/merged-pr-confirmation-work.test.ts @@ -1,3 +1,5 @@ +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"; @@ -262,6 +264,92 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu 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(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 9fc7234b2e6..4e7e3562156 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -180,6 +180,7 @@ 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, @@ -1335,7 +1336,7 @@ describe("startServer feedback export wiring", () => { expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); - expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationCandidates).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); @@ -1424,6 +1425,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 c1804998ef5..86074131dc3 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1601,8 +1601,8 @@ async function startServerWithDatabaseTeardown( : config.workspaceCompactionDelayHours * 60 * 60 * 1000, terminalWorkspaceInspectionBackoffMs: TERMINAL_WORKSPACE_RETRY_MS, }); - await registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations, - () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); const gitRestorePointWorker = registerGitRestorePointWork(deliveryWork, gitRestorePointRetentionService(db as any, { storage: storageService, retentionDays: config.gitRestorePointRetentionDays, 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 b306f637e27..b49ca48d5c1 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -204,6 +204,7 @@ export type IssueThreadInteractionServiceOptions = { resolvePullRequestState?: ( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ) => Promise; wakeup?: InteractionWakeup; pullRequestCacheTtlMs?: number; @@ -2045,10 +2046,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( @@ -2057,6 +2059,7 @@ export function issueThreadInteractionService( companyId: string; reference: GitHubPullRequestReference; }>, + signal?: AbortSignal, ) { const states = new Map(); const pending = entries.slice(); @@ -2064,11 +2067,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); } @@ -2076,11 +2081,7 @@ export function issueThreadInteractionService( ); return states; } - async function mergedPullRequestConfirmationCandidates(scope?: MergeConfirmationScope) { - const text = sql`concat_ws(' ', ${issueThreadInteractions.title}, ${issueThreadInteractions.summary}, - ${issueThreadInteractions.payload}->>'prompt', ${issueThreadInteractions.payload}->>'acceptLabel', - ${issueThreadInteractions.payload}->>'detailsMarkdown', ${issueThreadInteractions.payload}->'target'->>'key', - ${issueThreadInteractions.payload}->'target'->>'label', ${issueThreadInteractions.payload}->'target'->>'href')`; + 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, @@ -2092,14 +2093,24 @@ export function issueThreadInteractionService( scope ? eq(issueThreadInteractions.companyId, scope.companyId) : undefined, scope?.issueId ? eq(issueThreadInteractions.issueId, scope.issueId) : undefined, scope?.interactionId ? eq(issueThreadInteractions.id, scope.interactionId) : undefined, - sql`${text} ~* '(merge|merging)'`, - sql`${text} ~* '(github\\.com/[^ /]+/[^ /]+/pull/[1-9][0-9]*|[a-z0-9_.-]+/[a-z0-9_.-]+#[1-9][0-9]*)'`, + after ? sql`${issueThreadInteractions.id} > ${after}::uuid` : undefined, sql`NOT EXISTS (SELECT 1 FROM ${toolActionRequests} WHERE ${toolActionRequests.interactionId} = ${issueThreadInteractions.id})`, - )); - return rows.flatMap(row => { + )).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: { @@ -3012,6 +3023,7 @@ export function issueThreadInteractionService( await emitInteractionResolvedTelemetry(db, interaction); return interaction; }, + mergedPullRequestConfirmationPage, mergedPullRequestConfirmationCandidates, sweepMergedPullRequestConfirmations: async ( mergedHints: Array<{ @@ -3064,7 +3076,7 @@ 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"); } diff --git a/server/src/services/merged-pr-confirmation-work.ts b/server/src/services/merged-pr-confirmation-work.ts index bfccaeea79f..6a754d5a88a 100644 --- a/server/src/services/merged-pr-confirmation-work.ts +++ b/server/src/services/merged-pr-confirmation-work.ts @@ -9,16 +9,24 @@ export const MERGE_CONFIRMATION_RECONCILE_MS = 5 * 60_000; export function registerMergeConfirmationWork( coordinator: ReturnType, service: Pick, - "mergedPullRequestConfirmationCandidates" | "sweepMergedPullRequestConfirmations">, + "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) => (await service.mergedPullRequestConfirmationCandidates(scope)) - .map(({ interaction }) => ({ id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: Date.now() })), + 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); From 1b12f8f732bd1c6d848797c145b1768f5b76e3a1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:54:51 -0500 Subject: [PATCH 3/4] test(skills): verify Cloud CLI reads through shared object storage Use the real S3 provider with a controlled HTTP fixture and a stable test key. Preserve read-only/no-seeding behavior and verify corrupt, missing, and restored object outcomes under managed Cloud policy. Co-Authored-By: Paperclip --- .../__tests__/managed-skill-content.test.ts | 90 ++++++++++++++++--- 1 file changed, 77 insertions(+), 13 deletions(-) diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index cd4ddb8135f..59566d743e3 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; @@ -79,26 +81,88 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => { const companyId = await company(); - await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const objects = new Map(); + const server = createServer((request, response) => { + void (async () => { + const key = new URL(request.url!, "http://fixture").pathname; + if (request.method === "PUT") { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + let body = Buffer.concat(chunks); + if (request.headers["content-encoding"]?.includes("aws-chunked")) { + const decoded: Buffer[] = []; + let offset = 0; + while (offset < body.length) { + const end = body.indexOf("\r\n", offset); + const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16); + if (size === 0) break; + decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2; + } + body = Buffer.concat(decoded); + } + objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return; + } + const body = objects.get(key); + if (!body) { response.writeHead(404).end(); return; } + response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' }); + response.end(request.method === "HEAD" ? undefined : body); + })().catch(() => { response.writeHead(500).end(); }); + }); + await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address"); + const endpoint = `http://127.0.0.1:${address.port}`; + const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" }; + const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]])); + Object.assign(process.env, credentials); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({ + bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true, + }))); const key = "cloud-runtime-config:v1"; - const run = (phase: string) => promisify(execFile)(process.execPath, [ - "--import", fileURLToPath(import.meta.resolve("tsx")), - fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), - `--phase=${phase}`, `--company=${companyId}`, "--limit=50", - ], { env: { ...process.env, DATABASE_URL: database.connectionString, - PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", - PAPERCLIP_STORAGE_PROVIDER: "local_disk", PAPERCLIP_STORAGE_LOCAL_DIR: path.join(root, "objects") }, - timeout: 60_000, maxBuffer: 1024 * 1024 }); - await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); - expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); - await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); try { + await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const run = (phase: string) => promisify(execFile)(process.execPath, [ + "--import", fileURLToPath(import.meta.resolve("tsx")), + fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), + `--phase=${phase}`, `--company=${companyId}`, "--limit=50", + ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString, + PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", + PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3", + PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1", + PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" }, + timeout: 60_000, maxBuffer: 1024 * 1024 }); + await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); + await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!); expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]); + const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const); const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(verification.results).toEqual([expect.objectContaining({ status: "local_storage" })]); + expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()]).toEqual(before); + const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!; + const archive = objects.get(archiveKey)!; + const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1; + objects.set(archiveKey, corrupted); + const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]); + objects.set(archiveKey, archive); + const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!; + const manifest = objects.get(manifestKey)!; objects.delete(manifestKey); + const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]); + objects.set(manifestKey, manifest); + const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()].sort()).toEqual([...before].sort()); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))) + .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]); } finally { await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key)); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")))); + for (const [key, value] of Object.entries(previousCredentials)) { + if (value === undefined) delete process.env[key]; else process.env[key] = value; + } + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); } }, 180_000); From 51c554275e8786c39c186e008c5d83c0e3ba1b2a Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:55:32 -0500 Subject: [PATCH 4/4] docs: record bounded HTTP and Cloud fixture verification Co-Authored-By: Paperclip --- .../2026-10-30-merged-pr-confirmations-sleep.md | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md index 7b8699456fb..1c691c19509 100644 --- a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md +++ b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md @@ -91,9 +91,18 @@ 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 Linux cause is -not established; this is retained as an unresolved CI failure, not called a flake. -The current-head required CI result remains the completion gate. +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. ## Existing limitations outside this migration