From ea46d0846a8b42bea33be26a90509946c630fed7 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 01:16:58 -0500 Subject: [PATCH] fix(server): wake governed responses from interaction settlement Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 10 ++ ...26-governed-review-response-sleep-audit.md | 53 +++++++++ ...ssue-thread-interactions-telemetry.test.ts | 9 +- .../satisfied-connection-intents.test.ts | 25 ++++- .../services/chat-response-availability.ts | 2 + .../src/services/issue-thread-interactions.ts | 23 +++- .../external-chat-wait.integration.test.ts | 102 ++++++++++++++++++ .../services/satisfied-connection-intents.ts | 2 + .../tool-access/oauth/cloud-callback.ts | 10 +- .../tool-access/oauth/direct-callback.ts | 10 +- .../src/services/tool-access/oauth/state.ts | 59 +++++----- 11 files changed, 262 insertions(+), 43 deletions(-) diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 71bb42dc887..70891ec8608 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2400,3 +2400,13 @@ pass, with server types passing. The owner audit lists remaining interaction, ta authority and coordinator producer evidence. Registration, periodic removal, query cost, full verification and publication remain outstanding; the full stack scope and older unresolved gates remain unchanged. + +Slice 49 interaction checkpoint: five interaction misses plus task-status and +OAuth-denial misses are reproduced and fixed. Task-scoped notifications cover +existing resolution transactions and direct OAuth/connection cleanup writers. +The complete owner suite passes 180 cases; interaction/telemetry/satisfied-connection/ +generic-OAuth consumers pass 249 cases, and server types pass. Skipped telemetry, +external pool-fixture and callback-narrowing failures and repairs are preserved +in the owner audit. Execution-state-only restoration, real coordinator release, +query cost, registration/periodic removal and final acceptance remain. No PR yet, +no merge, and the full original stack scope remains open. diff --git a/doc/plans/2026-10-26-governed-review-response-sleep-audit.md b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md index ad6a147a224..f3e11b286f4 100644 --- a/doc/plans/2026-10-26-governed-review-response-sleep-audit.md +++ b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md @@ -241,3 +241,56 @@ Further producer audit findings, not completed migration claims: The worker remains unregistered and periodic repair is unchanged. These commits are checkpoints, not PR readiness or whole-instance sleep acceptance. + +## Interaction and task-authority checkpoint + +Five real interaction producer cases failed before wiring: skip, withdrawal, +rejecting suggestions, completing item verdicts and declining a connection intent. +They now pass through task-scoped notifications before the existing transaction's +write. Checkbox and administrative settlement use the same narrow topic when +the existing question/confirmation milestone fanout does not apply. Comment +supersession of item verdicts now owns its notification and row mutation together; +connection-intent ownership expiry receives its caller's issue transaction. + +Task status restoration and OAuth denial also reproduced missed wakes. Task-scoped +response-availability changes now reach governed-response recovery. Direct and +Cloud OAuth confirmation acceptance and automatic satisfied-connection retirement +notify the affected task before their writes. OAuth terminal denial owns the +signal and card mutation in one transaction; connection-intent retry phase changes +remain pending and do not wake this worker. Existing policy remains authoritative: +the attempted task-assignment restoration was rejected because externally bound +tasks cannot be reassigned through issueService. Its test now verifies that +restriction rather than bypassing it. Endpoint/identity configuration already +uses the company milestone boundary for supported authority restoration. + +Current evidence: all 180 owner cases pass, including OAuth outer rollback/commit, +plus 106 interaction service, nine telemetry, 50 satisfied-connection, and 84 +generic MCP/OAuth cases. Server typecheck passes. The first interaction consumer +run skipped the nine telemetry cases because embedded PostgreSQL was unavailable; +they now run through the standard validated external-database test option and pass. +The satisfied-connection suite initially passed 49/50: its one-connection-pool +fixture still used temp.connectionString after enabling external DB mode. That +fixture now uses the selected URL and all 50 pass. The first OAuth typecheck caught +nullable interactionId captured by a transaction callback; binding the checked ID +locally repaired it without changing policy. Preserved evidence: +/tmp/sleep49-interactions-before.log, /tmp/sleep49-interactions-after.log, +/tmp/sleep49-interactions-consumers.log, /tmp/sleep49-authority-before.log, +/tmp/sleep49-authority-after.log, /tmp/sleep49-authority-types.log, +/tmp/sleep49-telemetry-consumers.log, /tmp/sleep49-satisfied-consumers.log, +/tmp/sleep49-satisfied-consumers-v2.log, /tmp/sleep49-oauth.log, +/tmp/sleep49-authority-final.log and /tmp/sleep49-authority-final-types.log. + +Further audit: change-consent consumes already accepted metadata and cannot clear +a competing pending card. Confirmation-from-comment delegates resolution to the +covered interaction service. Native interaction audit records change summaries +only. Superseded native review retirement commits with a task-scoped native status +decision. Tool-review projection and secret-proposal settlement already signal +task milestones. Board claim and trusted-header membership restoration also +signal existing company milestones; compatibility membership backfill is agent +only. No new global live-authority registration dependency was found in governed +review authorization, which reads its durable binding/principal/destination proof. + +Still inspect execution-state-only transitions (pending policy state can block a +response without changing task status), prove real coordinator lease/retry release, +and finish candidate query-cost/lifecycle/wiring acceptance. The worker remains +unregistered and both periodic entrances remain intact. diff --git a/server/src/__tests__/issue-thread-interactions-telemetry.test.ts b/server/src/__tests__/issue-thread-interactions-telemetry.test.ts index d3db5df11d3..78bd4a1ca70 100644 --- a/server/src/__tests__/issue-thread-interactions-telemetry.test.ts +++ b/server/src/__tests__/issue-thread-interactions-telemetry.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -29,7 +30,8 @@ vi.mock("../telemetry.js", () => ({ getTelemetryClient: telemetryMocks.getTelemetryClient, })); -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_INTERACTION_TELEMETRY_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("issueThreadInteractionService telemetry", () => { @@ -38,8 +40,8 @@ describeEmbeddedPostgres("issueThreadInteractionService telemetry", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-interaction-telemetry-"); - db = createDb(tempDb.connectionString); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-interaction-telemetry-"); + db = createDb(externalUrl ?? tempDb!.connectionString); interactionsSvc = issueThreadInteractionService(db); }, 20_000); @@ -65,6 +67,7 @@ describeEmbeddedPostgres("issueThreadInteractionService telemetry", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts index 36bfab8c9d9..44348c5544d 100644 --- a/server/src/__tests__/satisfied-connection-intents.test.ts +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -19,16 +20,17 @@ import { idleWorkSnapshot } from "../services/task-admission.js"; import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; import { reserveRunBudget } from "../services/budget-reservations.js"; -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_SATISFIED_CONNECTION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; describePostgres("already available connection requests", () => { let db!: ReturnType; let temp!: Awaited>; beforeAll(async () => { - temp = await startEmbeddedPostgresTestDatabase("paperclip-satisfied-connections-"); - db = createDb(temp.connectionString); - await db.insert(authUsers).values({ id: "connection-owner", name: "Owner", email: "owner@example.test", createdAt: new Date(), updatedAt: new Date() }); + if (!externalUrl) temp = await startEmbeddedPostgresTestDatabase("paperclip-satisfied-connections-"); + db = createDb(externalUrl ?? temp.connectionString); + await db.insert(authUsers).values({ id: "connection-owner", name: "Owner", email: "owner@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); }, 20_000); beforeEach(async () => { await db.execute(sql`truncate table companies cascade`); }); afterAll(async () => { await db.$client.end({ timeout: 1 }); await temp?.cleanup(); }); @@ -59,6 +61,19 @@ describePostgres("already available connection requests", () => { return run!; } + it("wakes governed response recovery only when automatic cleanup retires a connection card", async () => { + const f = await seed(); + const notified = vi.fn(); + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeReviewResponse, notified); + try { + expect(await satisfiedConnectionIntentService(db).sweepPending()).toMatchObject({ satisfied: 1 }); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, f.intent.id)))[0].status).toBe("expired"); + expect(notified).toHaveBeenCalledOnce(); + expect(await satisfiedConnectionIntentService(db).sweepPending()).toMatchObject({ satisfied: 0 }); + expect(notified).toHaveBeenCalledOnce(); + } finally { unsubscribe(); } + }); + it("ignores an already available connection before the worker retires its card", async () => { const f = await seed(); const run = await retry(f); expect(await createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ companyId: f.companyId, runId: run.id, now: new Date() })).toMatchObject({ allowed: true }); @@ -356,7 +371,7 @@ describePostgres("already available connection requests", () => { it("records a system completion and one durable continuation with a one-connection pool", async () => { const f = await seed(); - const single = createDb(temp.connectionString, { maxConnections: 1 }); + const single = createDb(externalUrl ?? temp.connectionString, { maxConnections: 1 }); try { const outcomes = await Promise.all([satisfiedConnectionIntentService(single).sweepPending(), satisfiedConnectionIntentService(db).sweepPending()]); expect(outcomes.reduce((sum, outcome) => sum + outcome.satisfied, 0)).toBe(1); diff --git a/server/src/services/chat-response-availability.ts b/server/src/services/chat-response-availability.ts index d1459d65d9d..8779eab4a27 100644 --- a/server/src/services/chat-response-availability.ts +++ b/server/src/services/chat-response-availability.ts @@ -1,8 +1,10 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string }; export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) { + if (scope.issueId) await notifyNativeReviewResponse(tx, { companyId: scope.companyId, issueId: scope.issueId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); } diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 92d58eccbee..409f4e3345e 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; import { normalizeEscapedLineBreaks } from "@paperclipai/shared/validators/text"; @@ -1899,6 +1900,7 @@ async function expireStaleRequestConfirmationTarget( }); const expireInTransaction = async (tx: Db | any) => { if (args.row.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.row); + else await notifyNativeReviewResponse(tx, args.row); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -2177,6 +2179,7 @@ export function issueThreadInteractionService( const result = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, args.issue, args.actor); if (args.current.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.current); + else await notifyNativeReviewResponse(tx, args.current); await args.mutationOptions?.beforeResolveInTransaction?.(tx); // Policy mutations and review transitions use the same issue-row lock, // so the authoritative review policy and requester are stable through @@ -2461,6 +2464,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, args.issue, args.actor); if (args.current.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.current); + else await notifyNativeReviewResponse(tx, args.current); await args.mutationOptions?.beforeResolveInTransaction?.(tx); const issueContext = await tx .select({ @@ -2821,6 +2825,7 @@ export function issueThreadInteractionService( : "expired"; const resolvedAt = now(); const updated = await db.transaction(async (tx) => { + await notifyNativeReviewResponse(tx, current); const [row] = await tx .update(issueThreadInteractions) .set({ @@ -3253,6 +3258,7 @@ export function issueThreadInteractionService( for (const row of rows) { const updated = await db.transaction(async (tx) => { if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [resolved] = await tx .update(issueThreadInteractions) @@ -3339,6 +3345,7 @@ export function issueThreadInteractionService( for (const { row, replacementInteractionId } of supersededRows) { const updated = await db.transaction(async (tx) => { if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [updatedRow] = await tx .update(issueThreadInteractions) @@ -4026,6 +4033,7 @@ export function issueThreadInteractionService( await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); + await notifyNativeReviewResponse(tx, { companyId: issue.companyId, issueId: issue.id }); const resolvedAt = new Date(); const [claimed] = await tx .update(issueThreadInteractions) @@ -4262,6 +4270,7 @@ export function issueThreadInteractionService( complete, items, } satisfies RequestItemVerdictsResult; + if (complete) await notifyNativeReviewResponse(tx, { companyId: issue.companyId, issueId: issue.id }); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -4323,6 +4332,7 @@ export function issueThreadInteractionService( const [updated] = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); + await notifyNativeReviewResponse(tx, { companyId: issue.companyId, issueId: issue.id }); return tx.update(issueThreadInteractions) .set({ status: "rejected", @@ -4410,6 +4420,7 @@ export function issueThreadInteractionService( for (const row of superseded) { const updated = await db.transaction(async (tx) => { if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "connection_intent") { await tx .delete(toolOauthStates) @@ -4586,6 +4597,7 @@ export function issueThreadInteractionService( if (!sampleConfirmationRow) continue; const updatedInteractions = await db.transaction(async (tx) => { if (commentRows.some(row => row.kind === "request_confirmation")) await notifyChatRunMilestones(tx, sampleConfirmationRow); + else await notifyNativeReviewResponse(tx, sampleConfirmationRow); await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const updatedRows = await tx .update(issueThreadInteractions) @@ -4620,7 +4632,9 @@ export function issueThreadInteractionService( } for (const row of itemVerdictRows) { - const [updated] = await db + const [updated] = await db.transaction(async tx => { + await notifyNativeReviewResponse(tx, row); + return tx .update(issueThreadInteractions) .set({ status: "expired", @@ -4637,11 +4651,13 @@ export function issueThreadInteractionService( ), ) .returning(); + }); if (updated) expired.push(hydrateInteraction(updated)); } for (const row of connectionIntentRows) { const [updated] = await db.transaction(async (tx) => { + await notifyNativeReviewResponse(tx, row); await tx .delete(toolOauthStates) .where(eq(toolOauthStates.interactionId, row.id)); @@ -4728,6 +4744,7 @@ export function issueThreadInteractionService( }); const updated = await db.transaction(async (tx) => { if (row.kind === "request_confirmation") await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -4785,6 +4802,7 @@ export function issueThreadInteractionService( eq(issueThreadInteractions.status, "pending"), )); const expired: typeof issueThreadInteractions.$inferSelect[] = []; + if (pending.length) await notifyNativeReviewResponse(db, { companyId: issue.companyId, issueId: issue.id }); for (const row of pending) { const payload = row.payload as { purpose?: string; serviceName?: string } | null; const subject = payload?.purpose === "ai" @@ -4846,6 +4864,7 @@ export function issueThreadInteractionService( const updated = await db .transaction(async (tx) => { if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "connection_intent") { await tx .delete(toolOauthStates) @@ -4937,6 +4956,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); if (["ask_user_questions", "request_confirmation"].includes(current.kind)) await notifyChatRunMilestones(tx, current); + else await notifyNativeReviewResponse(tx, current); await resolveLinkedToolActionRequests(tx, current, { status: "cancelled", fromStatuses: ["pending", "approved"], @@ -5198,6 +5218,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); if (["ask_user_questions", "request_confirmation"].includes(current.kind)) await notifyChatRunMilestones(tx, current); + else await notifyNativeReviewResponse(tx, current); await resolveLinkedToolActionRequests(tx, current, { status: "cancelled", fromStatuses: ["pending", "approved"], diff --git a/server/src/services/native-runtime/external-chat-wait.integration.test.ts b/server/src/services/native-runtime/external-chat-wait.integration.test.ts index 69bd158e5eb..86415417bb2 100644 --- a/server/src/services/native-runtime/external-chat-wait.integration.test.ts +++ b/server/src/services/native-runtime/external-chat-wait.integration.test.ts @@ -38,6 +38,7 @@ import { statusDecisions, toolApplications, toolConnections, + toolOauthStates, chatVoiceSessions, chatVoiceReplies, chatVoiceToolCalls, workspaceOperations, } from "@paperclipai/db"; @@ -49,6 +50,7 @@ import type { import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { finalizeNativeRun, recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js"; import { configureAgentLifecycle, createAgentLifecycle, processAgentLifecycle } from "../agent-lifecycle.js"; +import { createOAuthStateStore } from "../tool-access/oauth/state.js"; import { approvalRecords } from "../approvals.js"; import { issueApprovalService } from "../issue-approvals.js"; import { nativeReviewResponseWork } from "./native-review-response-work.js"; @@ -2962,6 +2964,106 @@ describe("native external-chat response wait", () => { } finally { await lifecycle.stop(); await coordinator.stop(); } }); + it.each(["skip", "withdraw", "reject_suggestions", "complete_verdicts", "decline_connection"] as const)( + "reactivates governed responses after competing interaction %s", async action => { + const fixture = await seedRecoverableReviewResponse(); + const service = issueThreadInteractionService(db); + const issue = { id: fixture.issueId, companyId: fixture.companyId }; + const actor = { userId: fixture.userId }; + const competing = action === "decline_connection" + ? await service.createConnectionIntent(issue, { sourceRunId: fixture.runId, + addresseeUserId: fixture.userId, idempotencyKey: `review-competing:${randomUUID()}`, + payload: { version: 1, serviceSlug: "notion", serviceName: "Notion", serviceLogoUrl: null, + requestingAgentId: fixture.agentId, requestingAgentName: "Agent", phase: "requested" } }) + : action === "complete_verdicts" + ? await service.create(issue, { kind: "request_item_verdicts", payload: { version: 1, + prompt: "Review this item", items: [{ id: "one", label: "One" }] } }, actor) + : await service.create(issue, { kind: "suggest_tasks", payload: { version: 1, + tasks: [{ clientKey: "child", title: "A suggested task" }] } }, actor); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + if (action === "skip") await service.skipInteraction(issue, competing.id, {}, actor); + else if (action === "withdraw") await service.withdrawInteraction(issue, competing.id, {}, actor); + else if (action === "reject_suggestions") await service.rejectInteraction(issue, competing.id, {}, actor); + else if (action === "complete_verdicts") await service.submitItemVerdicts(issue, competing.id, + { verdicts: [{ id: "one", verdict: "approve" }] }, actor); + else await service.resolveConnectionIntent(issue, competing.id, { version: 1, outcome: "declined" }, actor); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it.each(["status", "assignment"] as const)("preserves governed response policy through task %s changes", async change => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(issues).set(change === "status" ? { status: "blocked" } : { assigneeAgentId: null }) + .where(eq(issues.id, fixture.issueId)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + if (change === "assignment") { + await expect(issueService(db).update(fixture.issueId, { assigneeAgentId: fixture.agentId })) + .rejects.toThrow("Agent assignment cannot change while this task is bound to an external channel"); + expect(await comments()).toHaveLength(0); + } else { + await issueService(db).update(fixture.issueId, { status: "in_review" }); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + } + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("reactivates governed responses after OAuth confirmation denial", async () => { + const fixture = await seedRecoverableReviewResponse(); + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.companyId, fixture.companyId)); + const competing = await issueThreadInteractionService(db).create({ id: fixture.issueId, companyId: fixture.companyId }, + { kind: "request_confirmation", payload: { version: 1, prompt: "Connect your account", target: { + type: "custom", key: `connection:${connection.id}`, revisionId: randomUUID(), label: "Connect account" } } }, + { userId: fixture.userId }); + const [state] = await db.insert(toolOauthStates).values({ state: randomUUID(), companyId: fixture.companyId, + connectionId: connection.id, codeVerifier: "test-verifier", issueId: fixture.issueId, interactionId: competing.id, + subjectUserId: fixture.userId, expiresAt: new Date(Date.now() + 60_000) }).returning(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + const rollback = new Error("rollback OAuth denial"); + await expect(db.transaction(async tx => { + await createOAuthStateStore(tx as unknown as typeof db).rejectPendingOAuthInteraction(state, + { actorType: "user", actorId: fixture.userId }); + expect(await comments()).toHaveLength(0); + throw rollback; + })).rejects.toBe(rollback); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, competing.id)))[0].status).toBe("pending"); + await db.transaction(async tx => { + await createOAuthStateStore(tx as unknown as typeof db).rejectPendingOAuthInteraction(state, + { actorType: "user", actorId: fixture.userId }); + expect(await comments()).toHaveLength(0); + }); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + it("reconstructs scheduled governed responses at startup and leaves their history quiet", async () => { const fixture = await seedRecoverableReviewResponse(); const errors: unknown[] = []; diff --git a/server/src/services/satisfied-connection-intents.ts b/server/src/services/satisfied-connection-intents.ts index f6a3eb39ef4..88682b5f16d 100644 --- a/server/src/services/satisfied-connection-intents.ts +++ b/server/src/services/satisfied-connection-intents.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { and, asc, eq, gt, inArray, isNull, or } from "drizzle-orm"; import { approvals, companyMemberships, connectionIntentDeliveries, issueApprovals, issueThreadInteractions, issues, type Db } from "@paperclipai/db"; import { connectionIntentPayloadSchema } from "@paperclipai/shared"; @@ -109,6 +110,7 @@ export function satisfiedConnectionIntentService(db: Db) { } const connection = await findSatisfiedToolConnection(txDb, task, intent); if (!connection) return false; + await notifyNativeReviewResponse(tx, candidate); const now = new Date(); // Retire the obsolete request without inventing a human approval. await tx.update(issueThreadInteractions).set({ status: "expired", resolvedAt: now, updatedAt: now, diff --git a/server/src/services/tool-access/oauth/cloud-callback.ts b/server/src/services/tool-access/oauth/cloud-callback.ts index 82c727df61f..9d11708a685 100644 --- a/server/src/services/tool-access/oauth/cloud-callback.ts +++ b/server/src/services/tool-access/oauth/cloud-callback.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "../../native-review-response-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { managedConnectorProfile } from "./client-settings.js"; import type { ActorInfo } from "../types.js"; @@ -407,18 +408,19 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { personalCredential || agentCredential ? credentialSecretRefs : [], tx, ); - const linkedInteractionKind = stateRow.interactionId + const linkedInteraction = stateRow.interactionId ? await tx - .select({ kind: issueThreadInteractions.kind }) + .select({ kind: issueThreadInteractions.kind, issueId: issueThreadInteractions.issueId }) .from(issueThreadInteractions) .where(eq(issueThreadInteractions.id, stateRow.interactionId)) .limit(1) - .then((rows) => rows[0]?.kind ?? null) + .then((rows) => rows[0] ?? null) : null; if ( stateRow.interactionId && - linkedInteractionKind === "request_confirmation" + linkedInteraction?.kind === "request_confirmation" ) { + await notifyNativeReviewResponse(tx, { companyId: connection.companyId, issueId: linkedInteraction.issueId }); await tx .update(issueThreadInteractions) .set({ diff --git a/server/src/services/tool-access/oauth/direct-callback.ts b/server/src/services/tool-access/oauth/direct-callback.ts index eb4f53022ad..23edba17fa0 100644 --- a/server/src/services/tool-access/oauth/direct-callback.ts +++ b/server/src/services/tool-access/oauth/direct-callback.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "../../native-review-response-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { credentialScope } from "./client-settings.js"; import { withoutOAuthRefreshLease } from "./refresh-leases.js"; @@ -373,18 +374,19 @@ export function createDirectOAuthCallback(db: Db, dependencies: Dependencies) { updatedAt: new Date(), }) .where(eq(toolApplications.id, connection.applicationId)); - const linkedInteractionKind = stateRow.interactionId + const linkedInteraction = stateRow.interactionId ? await tx - .select({ kind: issueThreadInteractions.kind }) + .select({ kind: issueThreadInteractions.kind, issueId: issueThreadInteractions.issueId }) .from(issueThreadInteractions) .where(eq(issueThreadInteractions.id, stateRow.interactionId)) .limit(1) - .then((rows) => rows[0]?.kind ?? null) + .then((rows) => rows[0] ?? null) : null; if ( stateRow.interactionId && - linkedInteractionKind === "request_confirmation" + linkedInteraction?.kind === "request_confirmation" ) { + await notifyNativeReviewResponse(tx, { companyId: connection.companyId, issueId: linkedInteraction.issueId }); await tx .update(issueThreadInteractions) .set({ diff --git a/server/src/services/tool-access/oauth/state.ts b/server/src/services/tool-access/oauth/state.ts index 2ab81656edc..cb498c2ce73 100644 --- a/server/src/services/tool-access/oauth/state.ts +++ b/server/src/services/tool-access/oauth/state.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "../../native-review-response-notifications.js"; import type { ActorInfo } from "../types.js"; import { and, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -124,17 +125,19 @@ async function rejectPendingOAuthInteraction( stateRow: typeof toolOauthStates.$inferSelect, actor: ActorInfo | undefined, ) { - if (!stateRow.interactionId) return; + const interactionId = stateRow.interactionId; + if (!interactionId) return; const now = new Date(); const linked = await db .select({ kind: issueThreadInteractions.kind, + issueId: issueThreadInteractions.issueId, payload: issueThreadInteractions.payload, }) .from(issueThreadInteractions) .where( and( - eq(issueThreadInteractions.id, stateRow.interactionId), + eq(issueThreadInteractions.id, interactionId), eq(issueThreadInteractions.companyId, stateRow.companyId), ), ) @@ -152,36 +155,40 @@ async function rejectPendingOAuthInteraction( }) .where( and( - eq(issueThreadInteractions.id, stateRow.interactionId), + eq(issueThreadInteractions.id, interactionId), eq(issueThreadInteractions.companyId, stateRow.companyId), eq(issueThreadInteractions.status, "pending"), ), ); return; } - await db - .update(issueThreadInteractions) - .set({ - status: "rejected", - result: { - version: 1, - outcome: "rejected", - // Paperclip's own words: the provider's explanation is untrusted and - // this reason is rendered in the thread (PAP-17108). - reason: - "Authorization was declined or cancelled in the provider's window", - }, - resolvedByUserId: actor?.actorType === "user" ? actor.actorId : null, - resolvedAt: now, - updatedAt: now, - }) - .where( - and( - eq(issueThreadInteractions.id, stateRow.interactionId), - eq(issueThreadInteractions.companyId, stateRow.companyId), - eq(issueThreadInteractions.status, "pending"), - ), - ); + if (!linked) return; + await db.transaction(async tx => { + await notifyNativeReviewResponse(tx, { companyId: stateRow.companyId, issueId: linked.issueId }); + await tx + .update(issueThreadInteractions) + .set({ + status: "rejected", + result: { + version: 1, + outcome: "rejected", + // Paperclip's own words: the provider's explanation is untrusted and + // this reason is rendered in the thread (PAP-17108). + reason: + "Authorization was declined or cancelled in the provider's window", + }, + resolvedByUserId: actor?.actorType === "user" ? actor.actorId : null, + resolvedAt: now, + updatedAt: now, + }) + .where( + and( + eq(issueThreadInteractions.id, interactionId), + eq(issueThreadInteractions.companyId, stateRow.companyId), + eq(issueThreadInteractions.status, "pending"), + ), + ); + }); } return { peekOAuthState, validateOAuthState, consumeOAuthState, rejectPendingOAuthInteraction }; }