mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-22 16:51:28 +02:00
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 commitde9ab8e7e0: 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 commit96b96297f4at 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 <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
e733bddc45
commit
5847f047c7
12 files changed
+703
-34
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<typeof createDb>;
|
||||
let temp!: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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<ReturnType<typeof seed>>) {
|
||||
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<ReturnType<typeof seed>>, 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<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.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<typeof db.update>[0]) => {
|
||||
const builder = db.update(table);
|
||||
if (table !== connectionIntentDeliveries) return builder;
|
||||
return { set(values: Parameters<typeof builder.set>[0]) {
|
||||
const update = builder.set(values);
|
||||
if (!values.deliveredAt) return update;
|
||||
return { async where(condition: Parameters<typeof update.where>[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");
|
||||
});
|
||||
});
|
||||
@@ -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<ReturnType<typeof decideQueuedRunStaleness>, { stale: true }>,
|
||||
expectedStatus: "queued" | "running",
|
||||
now: Date,
|
||||
providerDispatchNotStarted = false,
|
||||
): Promise<CancelStaleQueuedRunOutcome> {
|
||||
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 };
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<Heartbeat, "wakeup">) {
|
||||
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<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";
|
||||
const payload = interaction?.payload as { requestingAgentId?: string; serviceSlug?: string; purpose?: "ai" } | undefined;
|
||||
if (!loaded || !interaction || !["accepted", "rejected"].includes(interaction.status)
|
||||
if (!loaded || !interaction || (!retiredForAvailableConnection && !["accepted", "rejected"].includes(interaction.status))
|
||||
|| ["done", "cancelled"].includes(loaded.issue.status) || loaded.issue.assigneeAgentId !== payload?.requestingAgentId) {
|
||||
await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(eq(connectionIntentDeliveries.interactionId, interactionId));
|
||||
await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(and(eq(connectionIntentDeliveries.interactionId, interactionId),
|
||||
eq(connectionIntentDeliveries.companyId, claimed.companyId),
|
||||
eq(connectionIntentDeliveries.nextAttemptAt, claimed.nextAttemptAt), isNull(connectionIntentDeliveries.deliveredAt)));
|
||||
return;
|
||||
}
|
||||
const userId = interaction.addresseeUserId;
|
||||
@@ -134,7 +145,9 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick<Heartbea
|
||||
eq(companyMemberships.principalId, userId ?? ""), eq(companyMemberships.status, "active"),
|
||||
)).limit(1);
|
||||
if (!membership?.membershipRole || membership.membershipRole === "viewer") {
|
||||
await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(eq(connectionIntentDeliveries.interactionId, interactionId));
|
||||
await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(and(eq(connectionIntentDeliveries.interactionId, interactionId),
|
||||
eq(connectionIntentDeliveries.companyId, claimed.companyId),
|
||||
eq(connectionIntentDeliveries.nextAttemptAt, claimed.nextAttemptAt), isNull(connectionIntentDeliveries.deliveredAt)));
|
||||
return;
|
||||
}
|
||||
}
|
||||
@@ -143,6 +156,36 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick<Heartbea
|
||||
agentId: payload!.requestingAgentId!, responsibleUserId: userId!, serviceSlug: payload!.serviceSlug!, purpose: payload!.purpose });
|
||||
if (!ready) return;
|
||||
}
|
||||
if (retiredForAvailableConnection) {
|
||||
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);
|
||||
if (!restored) return;
|
||||
@@ -157,21 +200,32 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick<Heartbea
|
||||
// Check before dispatch: the previous worker may have crashed after enqueueing.
|
||||
if (!(await durableWake()).length) {
|
||||
try {
|
||||
await wakeConnectionIntentAfterResolution(heartbeat, { loaded, status: interaction.status, actorId: interaction.resolvedByUserId ?? interaction.addresseeUserId! });
|
||||
await wakeConnectionIntentAfterResolution(heartbeat, { loaded, status: interaction.status,
|
||||
actorId: retiredForAvailableConnection ? "connection-reconciliation" : interaction.resolvedByUserId ?? interaction.addresseeUserId!,
|
||||
...(retiredForAvailableConnection ? { actorType: "system" as const } : {}),
|
||||
});
|
||||
} catch (error) {
|
||||
// The unique wake key also protects overlapping leases. Other failures retry.
|
||||
if (!(await durableWake()).length) throw error;
|
||||
}
|
||||
}
|
||||
if ((await durableWake()).length) {
|
||||
await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(eq(connectionIntentDeliveries.interactionId, interactionId));
|
||||
await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(and(eq(connectionIntentDeliveries.interactionId, interactionId),
|
||||
eq(connectionIntentDeliveries.companyId, claimed.companyId),
|
||||
eq(connectionIntentDeliveries.nextAttemptAt, claimed.nextAttemptAt), isNull(connectionIntentDeliveries.deliveredAt)));
|
||||
}
|
||||
}
|
||||
async function hasPending() {
|
||||
return (await db.select({ id: connectionIntentDeliveries.interactionId }).from(connectionIntentDeliveries)
|
||||
.where(isNull(connectionIntentDeliveries.deliveredAt)).limit(1)).length > 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;
|
||||
|
||||
@@ -2739,6 +2739,7 @@ export function issueThreadInteractionService(
|
||||
}
|
||||
await touchIssue(tx, issue.id);
|
||||
inserted = true;
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.connection);
|
||||
return row;
|
||||
});
|
||||
|
||||
|
||||
@@ -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<string, unknown> } });
|
||||
continue;
|
||||
}
|
||||
const supersessionOutcome = interaction.status === "expired"
|
||||
&& ["superseded_by_newer_request", "superseded_by_comment", "stale_target"].includes(String(interactionResult.outcome));
|
||||
if (supersessionOutcome) {
|
||||
|
||||
@@ -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<Run, "runtimeMode" | "status" | "errorCode" | "resultJson"> & Partial<Pick<Run, DispatchFields>>,
|
||||
): 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
|
||||
|
||||
@@ -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<typeof issues.$inferSelect, "id" | "companyId" | "projectId" | "status" | "assigneeAgentId" | "assigneeUserId">;
|
||||
type Intent = Pick<typeof issueThreadInteractions.$inferSelect, "id" | "companyId" | "issueId" | "kind" | "status" | "payload" | "addresseeUserId"> & Partial<Pick<typeof issueThreadInteractions.$inferSelect, "sourceRunId">>;
|
||||
|
||||
/** 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 };
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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,
|
||||
}),
|
||||
)
|
||||
|
||||
Reference in new issue
Block a user