From 5847f047c7329e81296f1345cb54b9850056dfef Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 11 Oct 2026 09:30:45 -0700 Subject: [PATCH] fix(connections): retire requests for already available tools (#15856) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Connection cards let users provide the access an agent needs. > - Users can complete setup in Apps while an earlier card stays pending. > - A stale card can then cancel recovery despite valid existing access. > - This change verifies existing access and expires the obsolete request. > - It records a system outcome and sends one durable continuation. > - Approval, identity, permission, and workspace recovery gates remain in place. ## Linked Issues or Issue Description **What happened?** An ordinary tool connection request stayed pending after the addressed user connected the service in Apps. A later conversation retry was cancelled with `issue_waiting_for_response`. **Expected behavior** A request expires when the same user and requesting agent already have usable access. The agent gets a durable tool refresh. The system records no new human approval. **Steps to reproduce** 1. Have an agent request a tool connection on an active task. 2. Connect the service through Apps with the addressed user's existing grant, agent install, and allowed tools. 3. Let a conversation run stop and retry. The pending card still blocks the retry on the base branch. Related: #12345 introduced the connection request flow. This fix uses its existing access checks and delivery queue. ## What Changed - Verify company, task ownership, user identity, membership, health, grants, installs, and task tool policy before expiring an ordinary request. - Preserve AI account, channel, additional permission, and upstream provider requests on their existing paths. - Save the system outcome, audit event, and continuation in one transaction. Notify an idle delivery worker when a request commits. - Recheck access and other questions or approvals before delivery, retry, final dispatch, and native projection. Refresh system observations for an authorized replacement account. - Rearm delivery when access changes before provider work starts. Verify the host's admission receipt against execution evidence before allowing retry without a repair incident. - Require the current delivery claim when marking an attempt complete. An old completion cannot overwrite a rearmed continuation. - Add embedded-Postgres regressions for permissions, policy, concurrency, restart, replacement accounts, both dispatch gates, and the delivery race. Keep genuine workspace repair holds blocking. ## Verification - Latest-head connection, wake-context, dispatch, and legacy recovery suites: 124 tests passed. The connection suite includes 44 real-Postgres cases. - Server typecheck passed. Full `pnpm -r typecheck` and `pnpm build` passed on the latest head. - `pnpm test:run` was run locally. Its general-server group passed 18,583 tests and failed 20. That long run started before the review fixes, then loaded new test cases against modules transformed before those fixes. The affected latest-head suites pass in a fresh process. - One full-run failure also reproduces on unchanged base commit de9ab8e7e06: native-session-resume's damaged-prior-epoch replacement case. It is outside this change. Full-run groups after general-server did not run because that command stopped on failure. - Apex reviewed commit 96b96297f41 at 5/5. All review threads are resolved. All latest-head CI gates passed. The branch has no merge conflicts. ## Risks A false readiness result could retire a request too early. The fix reuses the runtime access selection rules, makes no new grants, and rechecks access at delivery. Ordinary expired requests do not authorize continuation. A failed workspace restore still requires repair through the existing gate. No schema migration is needed. ## Model Used OpenAI GPT-6 through Codex, with repository inspection, code editing, shell execution, and test tools. The session identifies the model family as GPT-6. It does not expose a more specific runtime model ID or context window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass (focused suites; full local limitations are listed above) - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 23 ++ .../satisfied-connection-intents.test.ts | 388 ++++++++++++++++++ .../modules/run-dispatch/adapters/postgres.ts | 54 ++- .../src/modules/run-dispatch/domain/policy.ts | 8 + .../run-dispatch/domain/wake-context.test.ts | 7 + .../run-dispatch/domain/wake-context.ts | 5 +- .../services/connection-intent-delivery.ts | 76 +++- .../src/services/issue-thread-interactions.ts | 1 + .../native-interaction-bridge.ts | 16 +- .../src/services/pre-dispatch-review-wait.ts | 20 +- .../services/satisfied-connection-intents.ts | 133 ++++++ server/src/services/tool-gateway.ts | 6 + 12 files changed, 703 insertions(+), 34 deletions(-) create mode 100644 server/src/__tests__/satisfied-connection-intents.test.ts create mode 100644 server/src/services/satisfied-connection-intents.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 0be0461e71e..58e352d9a17 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1602,6 +1602,29 @@ question does not start setup. The real connection card keeps user identity, access grants, the decision, and continuation together. This guidance does not approve a connection or bypass its normal user decision. +An ordinary tool connection request becomes obsolete when an already authorized +connection is usable by its requesting agent with the addressed user's identity. +Startup and periodic connection delivery reconcile that condition, retire the +card as expired, and record a system decision without a fabricated human approval. +The saved continuation is delivered once after rechecking the same access. The +scheduled retry gate also verifies this condition before the next sweep, so the +obsolete card cannot cancel a valid retry. Missing, unhealthy, disabled, ambiguous, +or unauthorized identities remain blocked. Requests for additional tool access, +AI authentication, inbox setup, or an upstream aggregator service keep their normal +approval paths. Task ownership, real questions, approvals, budgets, pauses, and +workspace recovery checks still govern admission. + +New requests notify the connection worker so this reconciliation also starts after an idle startup scan. Readiness uses the gateway's runtime tool +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. 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 A confirmation, checkbox confirmation, or question answer is new conversation diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts new file mode 100644 index 00000000000..ed37e0b6337 --- /dev/null +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -0,0 +1,388 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +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"; +import { connectionIntentDeliveryService } from "../services/connection-intent-delivery.js"; +import { findSatisfiedToolConnection, satisfiedConnectionIntentService } from "../services/satisfied-connection-intents.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +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(); +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() }); + }, 20_000); + beforeEach(async () => { await db.execute(sql`truncate table companies cascade`); }); + afterAll(async () => { await db.$client.end({ timeout: 1 }); await temp?.cleanup(); }); + + async function seed() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Connections", issuePrefix: "CON", defaultResponsibleUserId: "connection-owner" }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "connection-owner", status: "active", membershipRole: "member" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Assignee", role: "engineer", adapterType: "claude_local", status: "idle", runtimeConfig: { heartbeat: { wakeOnDemand: true } } }); + const [task] = await db.insert(issues).values({ id: issueId, companyId, title: "Use Sentry", status: "in_review", assigneeAgentId: agentId, responsibleUserId: "connection-owner" }).returning(); + const [intent] = await db.insert(issueThreadInteractions).values({ companyId, issueId, kind: "connection_intent", status: "pending", addresseeUserId: "connection-owner", + requestedResolverPolicy: "human_only", effectiveResolverPolicy: "human_only", payload: { version: 1, serviceSlug: "sentry", serviceName: "Sentry", requestingAgentId: agentId, requestingAgentName: "Assignee", phase: "requested" } }).returning(); + const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: "sentry", name: "Sentry", type: "mcp_http", status: "active", metadata: { sourceTemplateKey: "sentry" } }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app!.id, name: "Existing Sentry", uid: "sentry", transport: "mcp_remote", authKind: "api_key", credentialPolicy: "per_user", status: "active", enabled: true, healthStatus: "ok", config: { sourceTemplateKey: "sentry" } }).returning(); + const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId: connection!.id, kind: "user", subjectUserId: "connection-owner", status: "active" }).returning(); + await db.insert(toolConnectionInstalls).values({ companyId, connectionId: connection!.id, targetType: "agent", targetId: agentId }); + const [tool] = await db.insert(toolCatalogEntries).values({ companyId, connectionId: connection!.id, toolName: "sentry-read", name: "sentry-read", versionHash: "v1", status: "active", entryKind: "tool" }).returning(); + const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: "reads", name: "Reads", defaultAction: "allow", status: "active" }).returning(); + await db.insert(toolProfileBindings).values({ companyId, profileId: profile!.id, targetType: "agent", targetId: agentId }); + return { companyId, agentId, issueId, task: task!, intent: intent!, connection: connection!, grant: grant!, tool: tool!, profile: profile! }; + } + async function retry(f: Awaited>) { + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + status: "scheduled_retry", scheduledRetryReason: "transient_failure", scheduledRetryAt: new Date(), runtimeMode: "legacy", + responsibleUserId: "connection-owner", contextSnapshot: { issueId: f.issueId }, resultJson: { conversationContinuation: "continue_conversation_v1" } }).returning(); + return run!; + } + + 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 }); + expect((await db.select().from(issueThreadInteractions))[0].status).toBe("pending"); + }); + + it("keeps a retry under another user's identity waiting for its response", async () => { + const f = await seed(); const run = await retry(f); + await db.update(heartbeatRuns).set({ responsibleUserId: "someone-else" }).where(eq(heartbeatRuns.id, run.id)); + expect(await createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ companyId: f.companyId, runId: run.id, now: new Date() })).toMatchObject({ allowed: false, errorCode: "issue_waiting_for_response" }); + }); + + it("notifies an idle worker when a new connection request commits", async () => { + const f = await seed(); const source = await retry(f); + await db.delete(issueThreadInteractions); + const wakeup = vi.fn().mockResolvedValue(null); + const worker = connectionIntentDeliveryService(db, { wakeup }); + expect(await worker.hasPending()).toBe(false); + const notified = vi.fn(); + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.connection, notified); + try { + await issueThreadInteractionService(db).createConnectionIntent({ id: f.issueId, companyId: f.companyId }, { + payload: f.intent.payload, sourceRunId: source.id, addresseeUserId: "connection-owner", idempotencyKey: "new-card", + }); + expect(notified).toHaveBeenCalledOnce(); + expect(await worker.hasPending()).toBe(true); + await worker.sweepPending(); + expect(wakeup).toHaveBeenCalledOnce(); + } finally { unsubscribe(); } + }); + + it.each(["wrong_identity", "revoked", "disabled", "removed_member", "denied_policy"] as const)("rechecks %s after the card retires", async (reason) => { + const f = await seed(); const run = await retry(f); + await satisfiedConnectionIntentService(db).sweepPending(); + if (reason === "wrong_identity") await db.update(heartbeatRuns).set({ responsibleUserId: "someone-else" }); + if (reason === "revoked") await db.update(connectionGrants).set({ status: "revoked" }); + if (reason === "disabled") await db.update(toolConnections).set({ enabled: false }); + if (reason === "removed_member") await db.delete(companyMemberships); + if (reason === "denied_policy") await db.insert(toolPolicies).values({ companyId: f.companyId, name: "Block this task", policyType: "block", selectors: { issueId: f.issueId } }); + expect(await createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ companyId: f.companyId, runId: run.id, now: new Date() })).toMatchObject({ allowed: false, errorCode: "issue_waiting_for_response" }); + }); + + it.each(["company", "issue", "project"] as const)("uses %s policy when deciding whether tools are ready", async (scope) => { + const f = await seed(); + const selectors = scope === "issue" ? { issueId: f.issueId } : scope === "project" ? { projectId: randomUUID() } : {}; + if (scope === "project") { + const projectId = (selectors as { projectId: string }).projectId; + await db.insert(projects).values({ id: projectId, companyId: f.companyId, name: "Tools" }); + await db.update(issues).set({ projectId }); + } + await db.insert(toolPolicies).values({ companyId: f.companyId, name: "Block tools", policyType: "block", selectors }); + expect(await satisfiedConnectionIntentService(db).sweepPending()).toMatchObject({ satisfied: 0 }); + expect((await db.select().from(issueThreadInteractions))[0].status).toBe("pending"); + }); + + it("retains invocation approval when the authorized connection's tools require it", async () => { + const f = await seed(); + await db.insert(toolPolicies).values({ companyId: f.companyId, name: "Ask before use", policyType: "require_approval", selectors: { issueId: f.issueId } }); + expect(await satisfiedConnectionIntentService(db).sweepPending()).toMatchObject({ satisfied: 1 }); + expect((await db.select().from(toolPolicies))[0].policyType).toBe("require_approval"); + expect(await db.select().from(approvals)).toHaveLength(0); + expect((await db.select().from(issueThreadInteractions))[0].resolvedByUserId).toBeNull(); + }); + + async function addWait(f: Awaited>, kind: "question" | "approval") { + if (kind === "question") { + const [question] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + kind: "ask_user_questions", status: "pending", payload: { version: 1, questions: [] } }).returning(); + return async () => { await db.update(issueThreadInteractions).set({ status: "answered" }).where(eq(issueThreadInteractions.id, question!.id)); }; + } + const [approval] = await db.insert(approvals).values({ companyId: f.companyId, type: "tool_action", status: "pending", payload: {} }).returning(); + await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issueId, approvalId: approval!.id }); + return async () => { await db.update(approvals).set({ status: "approved" }).where(eq(approvals.id, approval!.id)); }; + } + + it.each(["question", "approval"] as const)("defers the saved system continuation for another %s", async (kind) => { + const f = await seed(); const clearWait = await addWait(f, kind); + const wakeup = vi.fn().mockResolvedValue(null); + const worker = connectionIntentDeliveryService(db, { wakeup }); + await worker.sweepPending(); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, f.intent.id)))[0].status).toBe("expired"); + expect(wakeup).not.toHaveBeenCalled(); + expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).toBeNull(); + await clearWait(); + await db.update(connectionIntentDeliveries).set({ nextAttemptAt: new Date() }); + await worker.sweepPending(); + expect(wakeup).toHaveBeenCalledOnce(); + }); + + it.each(["question", "approval", "revoked"] as const)("rechecks a queued system continuation after %s changes", async (kind) => { + const f = await seed(); const run = await retry(f); + await satisfiedConnectionIntentService(db).sweepPending(); + await db.update(heartbeatRuns).set({ status: "queued", contextSnapshot: { issueId: f.issueId, + connectionIntentResolution: "existing_connection", interactionKind: "connection_intent", interactionStatus: "expired", + mutation: "interaction", wakeReason: "issue_commented", source: "connection_intent.resolved" } }).where(eq(heartbeatRuns.id, run.id)); + if (kind === "revoked") await db.update(connectionGrants).set({ status: "revoked" }); + else await addWait(f, kind); + expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id, + 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.each(["queued", "running"] as const)("rearms an unstarted %s wait without manufacturing a workspace repair hold", async (status) => { + 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, startedAt: status === "running" ? new Date() : null, + runtimeModeResolvedAt: status === "running" ? new Date() : null, + controllerBootId: status === "running" ? randomUUID() : null, + executionStage: status === "running" ? "preparing" : null, 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" }); + const dispatch = createPostgresRunDispatchAdapter(db); + if (status === "queued") { + expect(await dispatch.cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id, + expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" }); + } else { + const invoke = vi.fn(async () => null); + expect(await dispatch.dispatchResolvedInteractionIfCurrent({ companyId: f.companyId, runId: run.id, + expectedStatus: "running", now: new Date(), dispatch: invoke })).toMatchObject({ dispatched: false, + cancellation: { outcome: "cancelled", errorCode: "issue_waiting_for_response" } }); + expect(invoke).not.toHaveBeenCalled(); + } + 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("does not let an old delivery complete after cancellation rearms its continuation", 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(heartbeatRuns).set({ status: "queued", wakeupRequestId: wake!.id, + contextSnapshot: { issueId: f.issueId, interactionId: f.intent.id, connectionIntentResolution: "existing_connection", + interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", source: "connection_intent.resolved" } }).where(eq(heartbeatRuns.id, run.id)); + let cancelled = false; + // Pause precisely after durableWake() succeeds, before its completion write. + const racingDb = new Proxy(db, { get(target, property) { + if (property !== "update") return Reflect.get(target, property); + return (table: Parameters[0]) => { + const builder = db.update(table); + if (table !== connectionIntentDeliveries) return builder; + return { set(values: Parameters[0]) { + const update = builder.set(values); + if (!values.deliveredAt) return update; + return { async where(condition: Parameters[0]) { + 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" }); + cancelled = true; + return update.where(condition); + } }; + } }; + }; + } }); + await connectionIntentDeliveryService(racingDb, { wakeup: vi.fn() }).deliver(f.intent.id); + expect(cancelled).toBe(true); + expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).toBeNull(); + 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(); + }); + + 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 }); + try { + const outcomes = await Promise.all([satisfiedConnectionIntentService(single).sweepPending(), satisfiedConnectionIntentService(db).sweepPending()]); + expect(outcomes.reduce((sum, outcome) => sum + outcome.satisfied, 0)).toBe(1); + expect((await db.select().from(issueThreadInteractions))[0]).toMatchObject({ status: "expired", resolvedByUserId: null, result: { outcome: "expired", connectionId: f.connection.id } }); + expect(await db.select().from(connectionIntentDeliveries)).toHaveLength(1); + expect(await db.select().from(activityLog)).toEqual([expect.objectContaining({ actorType: "system", action: "issue.thread_interaction_resolved" })]); + expect(await db.select().from(connectionGrantDelegations)).toHaveLength(0); + expect(await db.select().from(toolPolicies)).toHaveLength(0); + expect(await db.select().from(connectionGrants)).toHaveLength(1); + expect(await db.select().from(toolConnectionInstalls)).toHaveLength(1); + expect(await satisfiedConnectionIntentService(single).sweepPending()).toMatchObject({ satisfied: 0 }); + } finally { await single.$client.end({ timeout: 1 }); } + }); + + it("discovers pending cards through the existing delivery worker", async () => { + const f = await seed(); const wakeup = vi.fn().mockResolvedValue(null); + const worker = connectionIntentDeliveryService(db, { wakeup }); + expect(await worker.hasPending()).toBe(true); + await worker.sweepPending(); + expect((await db.select().from(issueThreadInteractions))[0].status).toBe("expired"); + expect(wakeup).toHaveBeenCalledOnce(); + expect(wakeup).toHaveBeenCalledWith(f.agentId, expect.objectContaining({ requestedByActorType: "system", idempotencyKey: `connection-intent:${f.intent.id}:expired`, contextSnapshot: expect.objectContaining({ refreshTools: true, responsibleUserId: "connection-owner" }) })); + }); + + it("rechecks access after a restart before it delivers the saved continuation", async () => { + const f = await seed(); + await satisfiedConnectionIntentService(db).sweepPending(); + await db.update(connectionGrants).set({ status: "revoked" }); + const wakeup = vi.fn().mockResolvedValue(null); + const worker = connectionIntentDeliveryService(db, { wakeup }); + await worker.sweepPending(); + expect(wakeup).not.toHaveBeenCalled(); + expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).toBeNull(); + await db.update(connectionGrants).set({ status: "active" }); + await db.update(connectionIntentDeliveries).set({ nextAttemptAt: new Date() }); + await worker.sweepPending(); + expect(wakeup).toHaveBeenCalledOnce(); + }); + + it("delivers system retirement to the native runner only while the same connection remains usable", async () => { + const f = await seed(); const run = await retry(f); + await satisfiedConnectionIntentService(db).sweepPending(); + const input = { db, companyId: f.companyId, issueId: f.issueId, runId: run.id, agentId: f.agentId, interactionIds: [f.intent.id] }; + expect(await materializeNativeInteractionResponses(input)).toEqual([expect.objectContaining({ + kind: "connection_intent", response: expect.objectContaining({ status: "expired", result: expect.objectContaining({ connectionId: f.connection.id }) }) })]); + await db.update(connectionGrants).set({ status: "revoked" }); + await expect(materializeNativeInteractionResponses(input)).rejects.toMatchObject({ code: "native_interaction_unresolved" }); + }); + + it.each(["missing_install", "denied_tools", "disabled", "unhealthy", "revoked", "wrong_user", "viewer", "removed_member", "additional_access", "ai", "channel", "upstream", "reassigned", "human_owner", "closed", "other_company"] as const)("preserves a request for %s", async (reason) => { + const f = await seed(); + if (reason === "missing_install") await db.delete(toolConnectionInstalls); + if (reason === "denied_tools") await db.update(toolProfiles).set({ defaultAction: "deny" }); + if (reason === "disabled") await db.update(toolConnections).set({ enabled: false }); + if (reason === "unhealthy") await db.update(toolConnections).set({ healthStatus: "error" }); + if (reason === "revoked") await db.update(connectionGrants).set({ status: "revoked" }); + if (reason === "wrong_user") await db.update(connectionGrants).set({ subjectUserId: "someone-else" }); + if (reason === "viewer") await db.update(companyMemberships).set({ membershipRole: "viewer" }); + if (reason === "removed_member") await db.delete(companyMemberships); + if (["ai", "channel", "upstream", "additional_access"].includes(reason)) { + const extra = reason === "ai" || reason === "channel" ? { purpose: reason } + : reason === "upstream" ? { upstreamService: { slug: "sentry", name: "Sentry" } } + : { accessRequest: { connectionId: f.connection.id, connectionName: "Sentry", tools: [{ catalogEntryId: f.tool.id, toolName: f.tool.toolName, versionHash: "v1", permission: "allowed" }] } }; + await db.update(issueThreadInteractions).set({ payload: { ...f.intent.payload, ...extra } as typeof f.intent.payload }); + } + if (reason === "reassigned") await db.update(issues).set({ assigneeAgentId: null }); + if (reason === "human_owner") await db.update(issues).set({ assigneeUserId: "connection-owner" }); + if (reason === "closed") await db.update(issues).set({ status: "done" }); + if (reason === "other_company") { + const [other] = await db.insert(companies).values({ name: "Other", issuePrefix: "OTH" }).returning(); + await db.update(issueThreadInteractions).set({ companyId: other!.id }); + } + const task = (await db.select().from(issues))[0]; const intent = (await db.select().from(issueThreadInteractions))[0]; + expect(await findSatisfiedToolConnection(db, task, intent)).toBeNull(); + expect(await satisfiedConnectionIntentService(db).sweepPending()).toMatchObject({ satisfied: 0 }); + expect((await db.select().from(issueThreadInteractions))[0].status).toBe("pending"); + expect(await db.select().from(connectionIntentDeliveries)).toHaveLength(0); + }); + + it("keeps a real question blocking even when a sibling connection is available", async () => { + const f = await seed(); const run = await retry(f); + await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, kind: "ask_user_questions", status: "pending", payload: { version: 1, questions: [] } }); + expect(await createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ companyId: f.companyId, runId: run.id, now: new Date() })).toMatchObject({ allowed: false, errorCode: "issue_waiting_for_response" }); + }); + + it("does not bypass a retained workspace hold after connecting", async () => { + const f = await seed(); + const leaseId = randomUUID(); + const workspaceRestoreRecovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] }; + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, status: "interrupted", finishedAt: new Date(), runtimeMode: "legacy", + contextSnapshot: { issueId: f.issueId }, resultJson: { conversationContinuation: "continue_conversation_v1", workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery } }).returning(); + const identity = { id: leaseId, companyId: f.companyId, heartbeatRunId: source!.id, provider: "daytona", providerLeaseId: `fixture-${leaseId}` }; + await db.insert(environmentLeases).values({ ...identity, issueId: f.issueId, status: "released", leasePolicy: "retain_on_failure", releasedAt: new Date(), cleanupStatus: "success", + metadata: { remoteExecutionTermination: remoteTerminationReceipt(identity, { providerLeaseId: identity.providerLeaseId, state: "stopped" }) } }); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", ownerType: "board", + fingerprint: `legacy-execution:${source!.id}`, status: "resolved", outcome: "blocked", + nextAction: "Repair the retained workspace before continuing.", evidence: { runId: source!.id, + workspaceRestoreFailure: "restore_failed", automaticRecovery: { replay: "blocked" }, + workspaceRestoreRecovery } }); + const run = await retry(f); + await db.update(heartbeatRuns).set({ retryOfRunId: source!.id, contextSnapshot: { issueId: f.issueId, retryOfRunId: source!.id } }).where(eq(heartbeatRuns.id, run.id)); + await satisfiedConnectionIntentService(db).sweepPending(); + const dispatch = createPostgresRunDispatchAdapter(db); + await dispatch.promoteOrCancelDueRetry({ companyId: f.companyId, runId: run.id, now: new Date() }); + const outcome = await dispatch.cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id, expectedStatus: "queued", now: new Date() }); + expect(outcome).toMatchObject({ outcome: "cancelled", errorCode: "execution_reconciliation_required" }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("cancelled"); + }); +}); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 693a68a22b1..947e7d33863 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -6,9 +6,7 @@ import type { Db } from "@paperclipai/db"; import { agentWakeupRequests, agents, - approvals, - issueApprovals, - issueThreadInteractions, + connectionIntentDeliveries, heartbeatRuns, issueRecoveryActions, issueComments, @@ -22,6 +20,8 @@ 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"; import { appendHeartbeatRunEvent } from "../../../services/heartbeat-run-events.js"; @@ -72,6 +72,7 @@ import { RunDispatchApplicationError } from "../application/types.js"; type HeartbeatRun = typeof heartbeatRuns.$inferSelect; type LoadGateFactsInput = { conversationContinuation: boolean; + responsibleUserId: string | null; runId: string; companyId: string; agentId: string; @@ -84,6 +85,8 @@ type LoadGateFactsResult = | { agentFound: true; facts: ScheduledRetryFacts } | { agentFound: false; issueId: string | null }; type LoadStalenessFactsInput = { + responsibleUserId: string | null; + conversationContinuation: boolean; runId: string; companyId: string; agentId: string; @@ -337,6 +340,7 @@ export function createPostgresRunDispatchAdapter( .select({ id: issues.id, companyId: issues.companyId, + projectId: issues.projectId, status: issues.status, assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, @@ -364,19 +368,8 @@ export function createPostgresRunDispatchAdapter( facts.issueExecutionRunId = issue.executionRunId; facts.issueCheckoutRunId = issue.checkoutRunId; if (input.conversationContinuation) { - const [interactions, linkedApprovals] = await Promise.all([ - dbOrTx.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( - eq(issueThreadInteractions.companyId, input.companyId), - eq(issueThreadInteractions.issueId, issueId), eq(issueThreadInteractions.status, "pending"), - )).limit(1), - dbOrTx.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, and( - eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, issueApprovals.companyId), - )).where(and( - eq(issueApprovals.companyId, input.companyId), eq(issueApprovals.issueId, issueId), - inArray(approvals.status, ["pending", "revision_requested"]), - )).limit(1), - ]); - facts.pendingResponse = interactions.length > 0 ? "interaction" : linkedApprovals.length > 0 ? "approval" : null; + facts.pendingResponse = await connectionContinuationPendingResponse(dbOrTx as unknown as Db, + issue, input.responsibleUserId, input.runId); } facts.reviewParticipant = await readNativeReviewParticipantFacts(dbOrTx, { companyId: input.companyId, issueId, agentId: input.agentId, @@ -464,6 +457,7 @@ export function createPostgresRunDispatchAdapter( companyId: run.companyId, agentId: run.agentId, conversationContinuation: run.runtimeMode === "legacy" && hasConversationContinuationPolicy(run.resultJson), + responsibleUserId: run.responsibleUserId, contextSnapshot: parseObject(run.contextSnapshot), scheduledRetryReason: run.scheduledRetryReason, retryReasonOverride: input.retryReasonOverride, @@ -519,6 +513,9 @@ export function createPostgresRunDispatchAdapter( const issueQuery = dbOrTx .select({ id: issues.id, + companyId: issues.companyId, + projectId: issues.projectId, + assigneeUserId: issues.assigneeUserId, status: issues.status, assigneeAgentId: issues.assigneeAgentId, executionRunId: issues.executionRunId, @@ -600,6 +597,9 @@ export function createPostgresRunDispatchAdapter( unresolvedBlockerIssueIds: readiness.unresolvedBlockerIssueIds, unresolvedBlockerCount: readiness.unresolvedBlockerCount, } : null, + pendingResponse: issue && (input.conversationContinuation || context.connectionIntentResolution === "existing_connection") + ? await connectionContinuationPendingResponse(dbOrTx as unknown as Db, issue, input.responsibleUserId, input.runId) + : null, issueFound: issue !== null, issueStatus: issue?.status ?? null, issueAssigneeAgentId: issue?.assigneeAgentId ?? null, @@ -770,6 +770,7 @@ export function createPostgresRunDispatchAdapter( companyId: run.companyId, agentId: run.agentId, conversationContinuation: run.runtimeMode === "legacy" && hasConversationContinuationPolicy(run.resultJson), + responsibleUserId: run.responsibleUserId, contextSnapshot: parseObject(run.contextSnapshot), scheduledRetryReason: run.scheduledRetryReason, retryReasonOverride: run.scheduledRetryReason, @@ -881,6 +882,7 @@ export function createPostgresRunDispatchAdapter( decision: Extract, { stale: true }>, expectedStatus: "queued" | "running", now: Date, + providerDispatchNotStarted = false, ): Promise { const [row] = await tx .update(heartbeatRuns) @@ -892,6 +894,9 @@ export function createPostgresRunDispatchAdapter( resultJson: { ...parseObject(run.resultJson), stopReason: decision.errorCode, + ...(providerDispatchNotStarted && decision.errorCode === "issue_waiting_for_response" + ? { preDispatchResponseWait: { version: 1, providerWorkStarted: false } } + : {}), ...(decision.errorCode === "execution_reconciliation_required" ? { executionWait: decision.details } : {}), @@ -927,6 +932,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({ @@ -976,6 +995,8 @@ export function createPostgresRunDispatchAdapter( runId: run.id, companyId: run.companyId, agentId: run.agentId, + responsibleUserId: run.responsibleUserId, + conversationContinuation: run.runtimeMode === "legacy" && hasConversationContinuationPolicy(run.resultJson), issueId, contextSnapshot, scheduledRetryReason: run.scheduledRetryReason, @@ -1129,6 +1150,7 @@ export function createPostgresRunDispatchAdapter( decision, input.expectedStatus, input.now, + true, // This locked gate has not handed off to the provider callback. ); return { dispatched: false as const, cancellation }; } diff --git a/server/src/modules/run-dispatch/domain/policy.ts b/server/src/modules/run-dispatch/domain/policy.ts index ec4b729839f..9c26cba3eb1 100644 --- a/server/src/modules/run-dispatch/domain/policy.ts +++ b/server/src/modules/run-dispatch/domain/policy.ts @@ -108,6 +108,7 @@ export type ScheduledRetryFacts = { }; export type QueuedRunStalenessErrorCode = + | "issue_waiting_for_response" | "execution_reconciliation_required" | "issue_dependencies_blocked" | "issue_not_found" @@ -129,6 +130,7 @@ export type StalenessDecision = }; export type QueuedRunFacts = { + pendingResponse?: "interaction" | "approval" | null; /** Rechecked for automatic native replacements immediately before dispatch. */ dependenciesBlocked?: DependencyBlockFacts | null; runId: string; @@ -546,6 +548,12 @@ export function decideQueuedRunStaleness( } } + if (facts.pendingResponse) { + return { stale: true, errorCode: "issue_waiting_for_response", + reason: "Cancelled because the continuation is still waiting for an authorized response", + details: { issueId: facts.issueId, pendingResponse: facts.pendingResponse } }; + } + if (facts.continuationParkApplies && facts.continuationParksExecutor) { return { stale: true, diff --git a/server/src/modules/run-dispatch/domain/wake-context.test.ts b/server/src/modules/run-dispatch/domain/wake-context.test.ts index ac6b84eb68e..f8d196d21ce 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.test.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.test.ts @@ -9,6 +9,13 @@ import { } from "./wake-context.js"; describe("wake context", () => { + it("recognizes a retired available connection without accepting ordinary expiry", () => { + const context = { interactionId: "connection", interactionStatus: "expired", interactionKind: "connection_intent", + connectionIntentResolution: "existing_connection", mutation: "interaction", wakeReason: "issue_commented" }; + expect(isResolvedInteractionContinuationWakeContext(context)).toBe(true); + expect(isResolvedInteractionContinuationWakeContext({ ...context, interactionKind: "request_confirmation" })).toBe(false); + expect(isResolvedInteractionContinuationWakeContext({ ...context, connectionIntentResolution: undefined })).toBe(false); + }); it("requires the subscription-specific non-assignee receipt for a subscription wait", () => { expect(isNonAssigneeWorkspaceBusyRetry("ai_connection_busy", { aiConnectionBusyDeferredWhileAssignee: false, diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index 2100b357d9c..a25833758f0 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -95,7 +95,10 @@ export function isResolvedInteractionContinuationWakeContext(contextSnapshot: un const interactionId = readNonEmptyString(context.interactionId); const interactionStatus = readNonEmptyString(context.interactionStatus); if (!interactionId || !interactionStatus) return false; - if (!RESOLVED_INTERACTION_CONTINUATION_STATUSES.has(interactionStatus)) return false; + const retiredConnection = interactionStatus === "expired" + && context.interactionKind === "connection_intent" + && context.connectionIntentResolution === "existing_connection"; + if (!retiredConnection && !RESOLVED_INTERACTION_CONTINUATION_STATUSES.has(interactionStatus)) return false; const mutation = readNonEmptyString(context.mutation); const wakeReason = readNonEmptyString(context.wakeReason); diff --git a/server/src/services/connection-intent-delivery.ts b/server/src/services/connection-intent-delivery.ts index c128d309fcd..3a5c8dfba44 100644 --- a/server/src/services/connection-intent-delivery.ts +++ b/server/src/services/connection-intent-delivery.ts @@ -1,6 +1,7 @@ import { connectionIntentService } from "./connection-intents.js"; +import { connectionContinuationPendingResponse, findSatisfiedToolConnection, satisfiedConnectionIntentService } from "./satisfied-connection-intents.js"; import { isAiConnectionConfigurationFailure } from "./ai-auth-failure.js"; -import { and, eq, isNull, lte, asc, notInArray, desc, sql } from "drizzle-orm"; +import { and, eq, isNull, lte, asc, inArray, notInArray, desc, sql } from "drizzle-orm"; import { connectionIntentDeliveries, issueThreadInteractions, issues, agentWakeupRequests, companyMemberships, heartbeatRuns, chatConversations, chatEndpoints, type Db } from "@paperclipai/db"; import type { heartbeatService } from "./heartbeat.js"; import { issueService } from "./issues.js"; @@ -13,10 +14,11 @@ export async function wakeConnectionIntentAfterResolution( input: { loaded: { issue: { id: string; assigneeAgentId: string | null; status: string }; - interaction: { id: string; resolvedAt?: string | Date | null; payload?: unknown }; + interaction: { id: string; resolvedAt?: string | Date | null; payload?: unknown; addresseeUserId?: string | null }; }; status: string; actorId: string; + actorType?: "user" | "system"; }, ) { const agentId = input.loaded.issue.assigneeAgentId; @@ -37,7 +39,7 @@ export async function wakeConnectionIntentAfterResolution( mutation: "interaction", }, idempotencyKey: `connection-intent:${input.loaded.interaction.id}:${input.status}`, - requestedByActorType: "user", + requestedByActorType: input.actorType ?? "user", requestedByActorId: input.actorId, contextSnapshot: { issueId: input.loaded.issue.id, @@ -48,6 +50,9 @@ export async function wakeConnectionIntentAfterResolution( mutation: "interaction", wakeReason: "issue_commented", source: "connection_intent.resolved", + ...(input.actorType === "system" && input.loaded.interaction.addresseeUserId + ? { responsibleUserId: input.loaded.interaction.addresseeUserId, connectionIntentResolution: "existing_connection" } + : {}), ...(interactionResolvedAt ? { interactionResolvedAt } : {}), @@ -64,6 +69,7 @@ export async function wakeConnectionIntentAfterResolution( export function connectionIntentDeliveryService(db: Db, heartbeat: Pick) { + const satisfiedIntents = satisfiedConnectionIntentService(db); // Only a repaired AI-authentication failure may reopen a blocked task. An old // card must never resume a newer failure, a reassignment, or a manual hold. async function restoreAiBlockedTask(loaded: { @@ -120,11 +126,16 @@ 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); if (!restored) return; @@ -157,21 +200,32 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick 0; + if ((await db.select({ id: connectionIntentDeliveries.interactionId }).from(connectionIntentDeliveries) + .where(isNull(connectionIntentDeliveries.deliveredAt)).limit(1)).length > 0) return true; + return (await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions) + .innerJoin(issues, and(eq(issues.id, issueThreadInteractions.issueId), eq(issues.companyId, issueThreadInteractions.companyId))) + .where(and(eq(issueThreadInteractions.kind, "connection_intent"), eq(issueThreadInteractions.status, "pending"), + inArray(issues.status, ["in_progress", "in_review"]), isNull(issues.assigneeUserId))) + .limit(1)).length > 0; } return { deliver, hasPending, tryDeliver: async (id: string) => { try { await deliver(id); } catch { /* Persisted delivery remains due after its lease. */ } }, sweepPending: async () => { + await satisfiedIntents.sweepPending(); const rows = await db.select().from(connectionIntentDeliveries).where(and(isNull(connectionIntentDeliveries.deliveredAt), lte(connectionIntentDeliveries.nextAttemptAt, new Date()))) .orderBy(asc(connectionIntentDeliveries.nextAttemptAt)).limit(50); let failed = 0; diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 38287d45535..96049788e73 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -2739,6 +2739,7 @@ export function issueThreadInteractionService( } await touchIssue(tx, issue.id); inserted = true; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connection); return row; }); diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index e12f54c977b..42e4d9b06b9 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -19,6 +19,7 @@ import { commitNativeStatusDecision } from "./status-decision-committer.js"; import { nativeSha256 } from "./canonical.js"; import { recordNativeAttentionAssessment } from "./work-assessments.js"; import { formatDurableQuestionResponseSummary } from "../question-response-delivery.js"; +import { findSatisfiedToolConnection } from "../satisfied-connection-intents.js"; import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, @@ -163,7 +164,7 @@ export async function materializeNativeInteractionResponses(input: { if (requestedIds.size === 0) return []; const [interactions, issue] = await Promise.all([ issueThreadInteractionService(input.db).listForIssue(input.issueId), - input.db.select({ status: issues.status }).from(issues).where(and( + input.db.select().from(issues).where(and( eq(issues.id, input.issueId), eq(issues.companyId, input.companyId), )).limit(1).then((rows) => rows[0] ?? null), @@ -201,6 +202,19 @@ export async function materializeNativeInteractionResponses(input: { if (invocation.status !== expectedInvocationStatus || (request.status !== "rejected" && interaction.result?.toolAction?.status !== request.status)) throw new NativeInteractionBridgeError("native_interaction_governed_result_mismatch", "Tool review outcome does not match its invocation"); } const interactionResult = record(interaction.result); + if (interaction.kind === "connection_intent" && interaction.status === "expired" + && interactionResult.outcome === "expired" && typeof interactionResult.connectionId === "string") { + const [run] = await input.db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.agentId, input.agentId))); + const ready = issue.assigneeAgentId === input.agentId && run?.responsibleUserId === interaction.addresseeUserId + ? await findSatisfiedToolConnection(input.db, issue, { ...interaction, addresseeUserId: interaction.addresseeUserId ?? null, status: "pending" }, input.runId) : null; + if (!ready || ready.id !== interactionResult.connectionId) { + throw new NativeInteractionBridgeError("native_interaction_unresolved", "The existing connection is no longer available"); + } + responses.push({ interactionId: interaction.id, kind: interaction.kind, + response: { status: "expired", result: structuredClone(interaction.result) as unknown as Record } }); + continue; + } const supersessionOutcome = interaction.status === "expired" && ["superseded_by_newer_request", "superseded_by_comment", "stale_target"].includes(String(interactionResult.outcome)); if (supersessionOutcome) { diff --git a/server/src/services/pre-dispatch-review-wait.ts b/server/src/services/pre-dispatch-review-wait.ts index 08ea49b498e..7448d5f8e55 100644 --- a/server/src/services/pre-dispatch-review-wait.ts +++ b/server/src/services/pre-dispatch-review-wait.ts @@ -6,23 +6,33 @@ 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 { + const finalResponseWait = run.errorCode === "issue_waiting_for_response" && + (run.resultJson?.preDispatchResponseWait as { version?: number; providerWorkStarted?: boolean } | undefined)?.version === 1 && + (run.resultJson?.preDispatchResponseWait as { providerWorkStarted?: boolean } | undefined)?.providerWorkStarted === false; 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" && - run.startedAt === null && run.runtimeModeResolvedAt === null && + (finalResponseWait || (run.startedAt === null && run.runtimeModeResolvedAt === null)) && run.processPid === null && run.processGroupId === null && run.processStartedAt === null && run.nativeIssueId === null && run.nativeSessionId === null && run.sessionIdAfter === null && - run.controllerBootId === null && run.controllerLeaseExpiresAt === null && run.executionStage === null; + (finalResponseWait || (run.controllerBootId === null && run.controllerLeaseExpiresAt === null && run.executionStage === null)); } /** The row receipt is a candidate, not permission to discard conflicting diff --git a/server/src/services/satisfied-connection-intents.ts b/server/src/services/satisfied-connection-intents.ts new file mode 100644 index 00000000000..49d295ae97f --- /dev/null +++ b/server/src/services/satisfied-connection-intents.ts @@ -0,0 +1,133 @@ +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"; +import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; +import { connectionIntentService } from "./connection-intents.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { logger } from "../middleware/logger.js"; +import { createToolGatewayService } from "./tool-gateway.js"; + +type Task = Pick; +type Intent = Pick & Partial>; + +/** Verify existing access only. This never installs tools, grants access, or adopts an identity. */ +export async function findSatisfiedToolConnection(db: Db, task: Task, intent: Intent, policyRunId?: string | null) { + if (intent.kind !== "connection_intent" || intent.status !== "pending" + || intent.companyId !== task.companyId || intent.issueId !== task.id + || task.assigneeUserId || !["in_progress", "in_review"].includes(task.status)) return null; + const parsed = connectionIntentPayloadSchema.safeParse(intent.payload); + if (!parsed.success) return null; + const payload = parsed.data; + // Additional permissions, provider authentication and inbox setup still need + // their governed completion paths. Aggregator cards also name an upstream app. + if (payload.purpose || payload.accessRequest || payload.upstreamService + || payload.requestingAgentId !== task.assigneeAgentId || !intent.addresseeUserId) return null; + if (intent.addresseeUserId !== "local-board") { + const [membership] = await db.select().from(companyMemberships).where(and( + eq(companyMemberships.companyId, task.companyId), eq(companyMemberships.principalType, "user"), + eq(companyMemberships.principalId, intent.addresseeUserId), + )); + if (membership?.status !== "active" || !membership.membershipRole || membership.membershipRole === "viewer") return null; + } + const connection = await connectionIntentService(db).usableConnectionForAgent({ companyId: task.companyId, + agentId: payload.requestingAgentId, responsibleUserId: intent.addresseeUserId, serviceSlug: payload.serviceSlug }); + if (!connection) return null; + // Use the gateway's exact descriptors and listing policy, including task, + // project and routine context. Per-invocation approvals remain enforced. + const access = await createToolGatewayService(db).summarizeConnectionAccessForAgent({ + companyId: task.companyId, connectionId: connection.id, agentId: payload.requestingAgentId, + heartbeatRunId: policyRunId === undefined ? intent.sourceRunId ?? null : policyRunId, + issueId: task.id, projectId: task.projectId, + }); + return access.allowedCount + access.askFirstCount > 0 ? connection : null; +} + +/** Recheck both live requests and system-retired cards against the next run. */ +export async function connectionContinuationPendingResponse(db: Db, task: Task, responsibleUserId: string | null, + policyRunId?: string | null): Promise<"interaction" | "approval" | null> { + const interactions = await db.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, task.companyId), eq(issueThreadInteractions.issueId, task.id), + or(eq(issueThreadInteractions.status, "pending"), and(eq(issueThreadInteractions.kind, "connection_intent"), eq(issueThreadInteractions.status, "expired"))), + )); + for (const interaction of interactions) { + const retired = interaction.status === "expired" && interaction.result?.outcome === "expired" + && "connectionId" in interaction.result && typeof interaction.result.connectionId === "string"; + if (interaction.status !== "pending" && !retired) continue; + 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); + // 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), + )).where(and(eq(issueApprovals.companyId, task.companyId), eq(issueApprovals.issueId, task.id), + inArray(approvals.status, ["pending", "revision_requested"]))).limit(1); + return approval ? "approval" : null; +} + +export function satisfiedConnectionIntentService(db: Db) { + let afterId: string | null = null; + return { + async sweepPending() { + const candidates = await db.select({ id: issueThreadInteractions.id, issueId: issues.id, companyId: issues.companyId }) + .from(issueThreadInteractions).innerJoin(issues, and( + eq(issues.id, issueThreadInteractions.issueId), eq(issues.companyId, issueThreadInteractions.companyId), + )).where(and(eq(issueThreadInteractions.kind, "connection_intent"), eq(issueThreadInteractions.status, "pending"), + inArray(issues.status, ["in_progress", "in_review"]), isNull(issues.assigneeUserId), + ...(afterId ? [gt(issueThreadInteractions.id, afterId)] : []))) + .orderBy(asc(issueThreadInteractions.id)).limit(50); + // Rotate past unavailable cards so they cannot starve a later ready one. + afterId = candidates.length === 50 ? candidates[49]!.id : null; + let satisfied = 0; + let failed = 0; + for (const candidate of candidates) { + const publications: ActivityPublication[] = []; + try { + const updated = await db.transaction(async (tx) => { + const txDb = tx as unknown as Db; + const [task] = await tx.select().from(issues).where(and( + eq(issues.id, candidate.issueId), eq(issues.companyId, candidate.companyId), + )).for("update"); + const [intent] = await tx.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, candidate.id), eq(issueThreadInteractions.companyId, candidate.companyId), + eq(issueThreadInteractions.issueId, candidate.issueId), + )).for("update"); + if (!task || !intent) return false; + if (intent.addresseeUserId && intent.addresseeUserId !== "local-board") { + await tx.select({ id: companyMemberships.id }).from(companyMemberships).where(and( + eq(companyMemberships.companyId, candidate.companyId), eq(companyMemberships.principalType, "user"), + eq(companyMemberships.principalId, intent.addresseeUserId), + )).for("share"); + } + const connection = await findSatisfiedToolConnection(txDb, task, intent); + if (!connection) return false; + const now = new Date(); + // Retire the obsolete request without inventing a human approval. + await tx.update(issueThreadInteractions).set({ status: "expired", resolvedAt: now, updatedAt: now, + result: { version: 1, outcome: "expired", connectionId: connection.id, + reason: "An already authorized connection is available to the requesting agent." }, + }).where(and(eq(issueThreadInteractions.id, intent.id), eq(issueThreadInteractions.status, "pending"))); + await tx.update(issues).set({ updatedAt: now }).where(and(eq(issues.id, task.id), eq(issues.companyId, task.companyId))); + await tx.insert(connectionIntentDeliveries).values({ interactionId: intent.id, companyId: task.companyId }).onConflictDoNothing(); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connection); + await logActivity(txDb, { companyId: task.companyId, actorType: "system", actorId: "connection-reconciliation", + action: "issue.thread_interaction_resolved", entityType: "issue", entityId: task.id, + details: { interactionId: intent.id, interactionKind: "connection_intent", status: "expired", + connectionId: connection.id, resolutionSource: "existing_connection" }, + }, publications); + return true; + }); + for (const publication of publications) publishActivity(publication); + if (updated) satisfied += 1; + } catch (error) { + failed += 1; + logger.warn({ err: error, interactionId: candidate.id }, "Could not reconcile an already available connection"); + } + } + return { scanned: candidates.length, satisfied, failed }; + }, + }; +} diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 73b025d1e4c..550ec59a300 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -9119,6 +9119,9 @@ export function createToolGatewayService( companyId: string; connectionId: string; agentId: string; + heartbeatRunId?: string | null; + issueId?: string | null; + projectId?: string | null; }) { await assertAgentInCompany(input.companyId, input.agentId); const tools = await connectedMcpToolsForConnection( @@ -9130,6 +9133,9 @@ export function createToolGatewayService( policyInputForAgentTool({ companyId: input.companyId, agentId: input.agentId, + heartbeatRunId: input.heartbeatRunId, + issueId: input.issueId, + projectId: input.projectId, tool, }), )