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 <noreply@paperclip.ing>

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
 

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<ReturnType<typeof seed>>) {
+    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 });

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({

server/src/services/connection-intent-delivery.ts

@@ -126,7 +126,7 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick<Heartbea
     const [loaded] = await db.select({ interaction: issueThreadInteractions, issue: issues })
       .from(issueThreadInteractions).innerJoin(issues, eq(issues.id, issueThreadInteractions.issueId))
       .where(and(eq(issueThreadInteractions.id, interactionId), eq(issueThreadInteractions.companyId, claimed.companyId), eq(issues.companyId, claimed.companyId)));
-    const interaction = loaded?.interaction;
+    let interaction = loaded?.interaction;
     const retiredForAvailableConnection = interaction?.status === "expired"
       && interaction.result?.outcome === "expired" && "connectionId" in interaction.result
       && typeof interaction.result.connectionId === "string";
@@ -153,9 +153,34 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick<Heartbea
       if (!ready) return;
     }
     if (retiredForAvailableConnection) {
-      const ready = await findSatisfiedToolConnection(db, loaded.issue, { ...interaction, status: "pending" });
-      if (!ready || ready.id !== (interaction.result as { connectionId: string }).connectionId) return;
-      if (await connectionContinuationPendingResponse(db, loaded.issue, userId, interaction.sourceRunId)) return;
+      const publications: ActivityPublication[] = [];
+      const current = await db.transaction(async tx => {
+        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);

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<Run, "runtimeMode" | "status" | "errorCode" | "resultJson"> & Partial<Pick<Run, DispatchFields>>,
 ): 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" &&

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