From 6d997f889c03b50fd5d880c2d48266e6c7ca729c Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 10 Oct 2026 23:05:23 -0700 Subject: [PATCH] fix: preserve connection delivery through unstarted waits Rearm skipped system continuations atomically, verify absence of execution before exempting response waits from repair, and support already authorized replacement accounts without changing human decisions. Co-Authored-By: Paperclip --- doc/execution-semantics.md | 7 +- .../satisfied-connection-intents.test.ts | 70 ++++++++++++++++++- .../modules/run-dispatch/adapters/postgres.ts | 16 +++++ .../services/connection-intent-delivery.ts | 33 +++++++-- .../src/services/pre-dispatch-review-wait.ts | 13 +++- .../services/satisfied-connection-intents.ts | 5 +- 6 files changed, 133 insertions(+), 11 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 8121c51e793..58e352d9a17 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1618,7 +1618,12 @@ New requests notify the connection worker so this reconciliation also starts aft listing policy with issue, project, and run context. System-retired requests are rechecked against the next run's responsible user and current access, and saved continuations defer while another question or linked approval remains pending. -Those checks run again during queued and final dispatch admission. +Those checks run again during queued and final dispatch admission. An unstarted +system continuation cancelled for a new response wait rearms its delivery in the +same transaction. Verified absence of execution keeps that wait out of workspace +repair incidents. Historical retirements do not bind all future retries to an +old account; an already authorized replacement can refresh the system observation +with its own audit record before delivery. Human decisions remain unchanged. ## Responses submitted during an active run diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts index d645edf4030..a2c95450d20 100644 --- a/server/src/__tests__/satisfied-connection-intents.test.ts +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -1,8 +1,8 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, approvals, authUsers, companies, companyMemberships, connectionGrants, connectionGrantDelegations, - connectionIntentDeliveries, createDb, environmentLeases, heartbeatRuns, issueApprovals, issueRecoveryActions, issueThreadInteractions, issues, projects, toolApplications, +import { activityLog, agentWakeupRequests, agents, approvals, authUsers, companies, companyMemberships, connectionGrants, connectionGrantDelegations, + connectionIntentDeliveries, createDb, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions, issueThreadInteractions, issues, projects, toolApplications, toolCatalogEntries, toolConnectionInstalls, toolConnections, toolProfileBindings, toolProfiles, toolPolicies, } from "@paperclipai/db"; import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js"; @@ -12,6 +12,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { materializeNativeInteractionResponses } from "../services/native-runtime/native-interaction-bridge.js"; import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; +import { legacyExecutionNeedsReconciliationWithEvidence } from "../services/legacy-execution-recovery.js"; import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; const support = await getEmbeddedPostgresTestSupport(); @@ -155,6 +156,71 @@ describePostgres("already available connection requests", () => { expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" }); }); + async function replaceAccount(f: Awaited>) { + await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, f.connection.id)); + await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, f.grant.id)); + const [connection] = await db.insert(toolConnections).values({ ...f.connection, id: randomUUID(), uid: "sentry-replacement", name: "Replacement Sentry" }).returning(); + await db.insert(connectionGrants).values({ ...f.grant, id: randomUUID(), connectionId: connection!.id, status: "active" }); + await db.insert(toolConnectionInstalls).values({ companyId: f.companyId, connectionId: connection!.id, targetType: "agent", targetId: f.agentId }); + await db.insert(toolCatalogEntries).values({ ...f.tool, id: randomUUID(), connectionId: connection!.id }); + return connection!; + } + + it("allows a later approved replacement instead of permanently binding retries to an old account", async () => { + const f = await seed(); const run = await retry(f); + await satisfiedConnectionIntentService(db).sweepPending(); + const replacement = await replaceAccount(f); + await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, kind: "connection_intent", + status: "accepted", payload: f.intent.payload, addresseeUserId: "connection-owner", resolvedByUserId: "connection-owner", + result: { version: 1, outcome: "accepted", connectionId: replacement.id }, resolvedAt: new Date() }); + expect(await createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ companyId: f.companyId, runId: run.id, now: new Date() })).toMatchObject({ allowed: true }); + }); + + it("refreshes only the system observation for an already authorized replacement", async () => { + const f = await seed(); const run = await retry(f); + await satisfiedConnectionIntentService(db).sweepPending(); + const replacement = await replaceAccount(f); + const input = { db, companyId: f.companyId, issueId: f.issueId, runId: run.id, agentId: f.agentId, interactionIds: [f.intent.id] }; + await expect(materializeNativeInteractionResponses(input)).rejects.toMatchObject({ code: "native_interaction_unresolved" }); + await connectionIntentDeliveryService(db, { wakeup: vi.fn().mockResolvedValue(null) }).sweepPending(); + expect((await db.select().from(issueThreadInteractions))[0]).toMatchObject({ status: "expired", resolvedByUserId: null, + result: { outcome: "expired", connectionId: replacement.id } }); + expect(await materializeNativeInteractionResponses(input)).toEqual([expect.objectContaining({ + response: expect.objectContaining({ result: expect.objectContaining({ connectionId: replacement.id }) }) })]); + expect(await db.select().from(activityLog)).toEqual(expect.arrayContaining([expect.objectContaining({ actorType: "system", + details: expect.objectContaining({ previousConnectionId: f.connection.id, connectionId: replacement.id }) })])); + }); + + it("rearms an unstarted wait without manufacturing a workspace repair hold", async () => { + const f = await seed(); const run = await retry(f); + await satisfiedConnectionIntentService(db).sweepPending(); + const key = `connection-intent:${f.intent.id}:expired`; + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "automation", status: "claimed", idempotencyKey: key, runId: run.id }).returning(); + await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }); + await db.update(heartbeatRuns).set({ status: "queued", wakeupRequestId: wake!.id, resultJson: null, + contextSnapshot: { issueId: f.issueId, interactionId: f.intent.id, connectionIntentResolution: "existing_connection", + interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", wakeReason: "issue_commented", source: "connection_intent.resolved" } }).where(eq(heartbeatRuns.id, run.id)); + await db.update(connectionGrants).set({ status: "revoked" }); + expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id, + expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" }); + const cancelled = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]; + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, cancelled)).toBe(false); + expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).toBeNull(); + expect((await db.select().from(agentWakeupRequests))[0].status).toBe("skipped"); + await db.update(connectionGrants).set({ status: "active" }); + const wakeup = vi.fn(async () => { + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", status: "queued", idempotencyKey: key }); + return null; + }); + await connectionIntentDeliveryService(db, { wakeup }).sweepPending(); + expect(wakeup).toHaveBeenCalledOnce(); + expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).not.toBeNull(); + await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: run.id, agentId: f.agentId, + seq: 2, eventType: "adapter.invoke" }); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, cancelled)).toBe(true); + }); + 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 }); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 58fd6e5e595..3a8828f8498 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -6,6 +6,7 @@ import type { Db } from "@paperclipai/db"; import { agentWakeupRequests, agents, + connectionIntentDeliveries, heartbeatRuns, issueRecoveryActions, issueComments, @@ -19,6 +20,7 @@ import { evaluateAgentInvokabilityFromDb } from "../../../services/agent-invokab import { budgetService, budgetServiceInTransaction } from "../../../services/budgets.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; import { isHeartbeatWakeOnDemandEnabled } from "../../../services/heartbeat-policy.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { connectionContinuationPendingResponse } from "../../../services/satisfied-connection-intents.js"; import { collectDispositionRepairSourceState } from "../../../services/recovery/disposition-repair.js"; import { legacyDispositionEpisode, legacyDispositionFingerprint } from "../../../services/recovery/legacy-continuation.js"; @@ -926,6 +928,20 @@ export function createPostgresRunDispatchAdapter( ); } + const context = parseObject(row.contextSnapshot); + if (decision.errorCode === "issue_waiting_for_response" && context.connectionIntentResolution === "existing_connection" + && context.source === "connection_intent.resolved" && context.interactionKind === "connection_intent" + && typeof context.interactionId === "string") { + // The partial wake-key uniqueness fence permits another delivery after + // this skipped receipt. Rearm atomically so restored access or a later + // answer can resume the untouched request after a restart. + await tx.update(connectionIntentDeliveries).set({ deliveredAt: null, nextAttemptAt: now }).where(and( + eq(connectionIntentDeliveries.companyId, row.companyId), + eq(connectionIntentDeliveries.interactionId, context.interactionId), + )); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connection); + } + await tx .update(issues) .set({ diff --git a/server/src/services/connection-intent-delivery.ts b/server/src/services/connection-intent-delivery.ts index e7f6d99400b..04fc7d020df 100644 --- a/server/src/services/connection-intent-delivery.ts +++ b/server/src/services/connection-intent-delivery.ts @@ -126,7 +126,7 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick { + const [task] = await tx.select().from(issues).where(and(eq(issues.id, loaded.issue.id), eq(issues.companyId, claimed.companyId))).for("update"); + const [card] = await tx.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, interactionId), eq(issueThreadInteractions.companyId, claimed.companyId), + eq(issueThreadInteractions.issueId, loaded.issue.id))).for("update"); + if (!task || !card || card.status !== "expired" || card.result?.outcome !== "expired") return null; + const ready = await findSatisfiedToolConnection(tx as unknown as Db, task, { ...card, status: "pending" }); + if (!ready) return null; + if (ready.id !== (card.result as { connectionId: string }).connectionId) { + const previousConnectionId = (card.result as { connectionId: string }).connectionId; + // This is a system observation of existing access. It grants nothing + // and never changes or impersonates a human decision. + const [updated] = await tx.update(issueThreadInteractions).set({ result: { ...card.result, + connectionId: ready.id }, updatedAt: new Date() }).where(eq(issueThreadInteractions.id, card.id)).returning(); + Object.assign(card, updated); + await logActivity(tx as unknown as Db, { companyId: claimed.companyId, actorType: "system", actorId: "connection-reconciliation", + action: "issue.thread_interaction_resolved", entityType: "issue", entityId: task.id, + details: { interactionId, interactionKind: "connection_intent", status: "expired", connectionId: ready.id, + previousConnectionId, resolutionSource: "existing_connection_refreshed" } }, publications); + } + return { task, card, waiting: await connectionContinuationPendingResponse(tx as unknown as Db, task, card.addresseeUserId, card.sourceRunId) }; + }); + for (const publication of publications) publishActivity(publication); + if (!current || current.waiting) return; + loaded.issue = current.task; + loaded.interaction = current.card; + interaction = current.card; } if (interaction.status === "accepted" && payload?.purpose === "ai" && loaded.issue.status === "blocked") { const restored = await restoreAiBlockedTask(loaded); diff --git a/server/src/services/pre-dispatch-review-wait.ts b/server/src/services/pre-dispatch-review-wait.ts index 08ea49b498e..2f5413b8f6f 100644 --- a/server/src/services/pre-dispatch-review-wait.ts +++ b/server/src/services/pre-dispatch-review-wait.ts @@ -6,16 +6,23 @@ import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; type Run = typeof heartbeatRuns.$inferSelect; type DispatchFields = "startedAt" | "runtimeModeResolvedAt" | "processPid" | "processGroupId" | "processStartedAt" | "nativeIssueId" | "nativeSessionId" | "sessionIdAfter" | - "controllerBootId" | "controllerLeaseExpiresAt" | "executionStage"; + "controllerBootId" | "controllerLeaseExpiresAt" | "executionStage" | "contextSnapshot"; /** The queued-run gate owns this receipt before execution authority is claimed. - * A deliberate review wait has no provider actions to reconcile. The error code + * A deliberate review or connection response wait has no provider actions to reconcile. The error code * alone is not proof; missing or conflicting dispatch evidence retains the hold. */ export function isPreDispatchReviewWait( run: Pick & Partial>, ): boolean { return run.runtimeMode === "legacy" && run.status === "cancelled" && - run.errorCode === "issue_continuation_waiting_on_review" && + (run.errorCode === "issue_continuation_waiting_on_review" || + (run.errorCode === "issue_waiting_for_response" && + run.contextSnapshot?.connectionIntentResolution === "existing_connection" && + run.contextSnapshot?.source === "connection_intent.resolved" && + run.contextSnapshot?.interactionKind === "connection_intent" && + run.contextSnapshot?.interactionStatus === "expired" && + run.contextSnapshot?.mutation === "interaction" && + typeof run.contextSnapshot?.interactionId === "string")) && run.resultJson?.stopReason === run.errorCode && run.resultJson?.timeoutSource === "stale_queued_run_gate" && run.resultJson?.workspaceRestoreFailure !== "restore_unsafe_archive" && diff --git a/server/src/services/satisfied-connection-intents.ts b/server/src/services/satisfied-connection-intents.ts index 43a9b540601..49d295ae97f 100644 --- a/server/src/services/satisfied-connection-intents.ts +++ b/server/src/services/satisfied-connection-intents.ts @@ -56,7 +56,10 @@ export async function connectionContinuationPendingResponse(db: Db, task: Task, if (retired && (interaction.payload as { requestingAgentId?: string }).requestingAgentId !== task.assigneeAgentId) continue; if (interaction.addresseeUserId !== responsibleUserId) return "interaction"; const connection = await findSatisfiedToolConnection(db, task, { ...interaction, status: "pending" }, policyRunId); - if (!connection || (retired && connection.id !== (interaction.result as { connectionId: string }).connectionId)) return "interaction"; + // Historical system retirements describe a service need, not a permanent + // binding to an account the user may later replace. Delivery and native + // response projection still validate their own exact recorded connection. + if (!connection) return "interaction"; } const [approval] = await db.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, and( eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, issueApprovals.companyId),