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),