diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 5727800977e..abf1b6ea455 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1068,8 +1068,10 @@ npx paperclipai allowed-hostname dotta-macbook-pro ## Agent status recovery Lifecycle verification stops automatic attempts after missing authentication, -an unavailable model or command, an invalid working directory, or unexpected -probe output. A generic failed test gets up to three failures before automatic +an unavailable model or command, an invalid working directory, unexpected +probe output, or an owned AI account selection refusal. Account selection guidance +names missing defaults, unavailable accounts, incompatible bindings, missing +responsible users, or credential sharing requirements without provider error text. A generic failed test gets up to three failures before automatic checks stop. The saved `verificationFailures` count only advances when the host verification test fails; claims deferred before the test do not spend that budget. An explicit retry or execution configuration change resets that count. diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 5ce2a939bf9..5ce43c1ac39 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -649,6 +649,34 @@ stall, including resource-wait retries of its wake, are retried after ten minutes, then twenty, and not after the third failed run. The stall stays listed on the waiting task, and a new stall wakes it again. +An unstarted queued turn names an agent setup or verification wait. A failed +setup that requires explicit Retry shows **Agent setup needs attention**, while +its saved turn remains queued. Temporary errors with automatic retries still show +**Waiting for agent setup**. +Queued setup failures also create the existing inline AI connection card on the +task, addressed to the person whose account host verification checks (the saved +lifecycle responsible user or company default). A queued turn can have a different +responsible user; neither its authority nor its saved identity is rewritten. +Runtime tokens remain unavailable for +queued runs. Completing the card rechecks verification through the lifecycle +worker and retains the original queued turn; it does not enqueue another model +turn. Decline, Stop, reassignment and other lifecycle holds remain effective. +Setup cards stay bound to their queued source turn. A replacement turn supersedes +the prior setup card instead of reusing a stopped turn's delivery. +New cards also name the lifecycle operation, so a stale completion cannot recheck +a newer configuration. Owned AI selection failures park during resume as well as +initial verification. For a routed agent, repair uses the account already pinned +to that lifecycle operation; it never allocates a task route, advances the pool +cursor, or replaces the agent's pool. Legacy direct-account setup cards remain +readable without an operation ID. +AI account selection failures use the selector's owned configuration evidence: +missing defaults, unavailable or incompatible connections, missing responsible +users, and credential sharing refusals park automatic host verification until +configuration repair or explicit Retry. They retain fixed, redacted guidance. +Provider and infrastructure failures keep their existing retry policy. Readiness +must commit before normal queue dispatch; this display grants no execution or +credential authority and does not change the task's lifecycle status. + An instance admission drain keeps accepted new turns queued and displays **Waiting for maintenance**. Ending the audited drain requests admission for these existing runs immediately. The shared delivery coordinator owns the drain's diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index a892973b51f..4598caecc50 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -1565,6 +1565,10 @@ export type ConnectionIntentPhase = "requested" | "authorizing" | "needs_retry"; * authorization URLs are returned solely from addressed board endpoints. */ export interface ConnectionIntentPayload { + /** Controller-created repair for a queued turn awaiting agent verification. */ + agentSetup?: true; + /** Server-owned lifecycle operation whose selected account this card repairs. */ + agentSetupOperationId?: string; /** Human-selected audience for this OAuth draft; applied only at completion. */ setupAccess?: { connectionId: string; access: "all_agents" | { agentIds: string[] } }; /** Existing chat setup; the server derives the destination from these IDs. */ diff --git a/packages/shared/src/validators/connection-intent.test.ts b/packages/shared/src/validators/connection-intent.test.ts index 3804b42f490..a65d6b659b6 100644 --- a/packages/shared/src/validators/connection-intent.test.ts +++ b/packages/shared/src/validators/connection-intent.test.ts @@ -42,6 +42,17 @@ describe("connection intent contracts", () => { expect(connectionIntentPayloadSchema.safeParse({ ...base, phase: "connected" }).success).toBe(false); }); + it("keeps queued agent setup markers server-owned and limited to AI cards", () => { + const payload = { version: 1, serviceSlug: "openrouter", serviceName: "OpenRouter", requestingAgentId: agentId, + requestingAgentName: "Reviewer", phase: "requested", agentSetup: true }; + expect(connectionIntentPayloadSchema.safeParse({ ...payload, purpose: "ai" }).success).toBe(true); + expect(connectionIntentPayloadSchema.safeParse({ ...payload, purpose: "ai", agentSetupOperationId: agentId }).success).toBe(true); + expect(connectionIntentPayloadSchema.safeParse({ ...payload, agentSetup: undefined, purpose: "ai", agentSetupOperationId: agentId }).success).toBe(false); + expect(connectionIntentPayloadSchema.safeParse(payload).success).toBe(false); + expect(connectionRequestInputSchema.safeParse({ service: "openrouter", agentSetup: true }).success).toBe(false); + expect(connectionRequestInputSchema.safeParse({ service: "openrouter", agentSetupOperationId: agentId }).success).toBe(false); + }); + it("validates terminal outcomes independently from payload state", () => { expect(connectionIntentResultSchema.parse({ version: 1, diff --git a/packages/shared/src/validators/issue.ts b/packages/shared/src/validators/issue.ts index caa8b843185..307eb382514 100644 --- a/packages/shared/src/validators/issue.ts +++ b/packages/shared/src/validators/issue.ts @@ -1261,6 +1261,8 @@ const connectionIntentBrandAssetSchema = z export const connectionIntentPayloadSchema = z .object({ + agentSetup: z.literal(true).optional(), + agentSetupOperationId: z.string().guid().optional(), setupAccess: z.object({ connectionId: z.string().guid(), access: z.union([z.literal("all_agents"), z.object({ agentIds: z.array(z.string().guid()).max(250) }).strict()]), @@ -1291,7 +1293,9 @@ export const connectionIntentPayloadSchema = z requestingAgentName: z.string().trim().min(1).max(160), phase: connectionIntentPhaseSchema, }) - .strict(); + .strict() + .refine(value => !value.agentSetup || value.purpose === "ai", "Agent setup cards require AI authentication") + .refine(value => !value.agentSetupOperationId || value.agentSetup === true, "A setup operation requires an agent setup card"); export const connectionIntentResultSchema = z .object({ diff --git a/server/src/__tests__/agent-setup-connection-intents.test.ts b/server/src/__tests__/agent-setup-connection-intents.test.ts new file mode 100644 index 00000000000..1dff91b6032 --- /dev/null +++ b/server/src/__tests__/agent-setup-connection-intents.test.ts @@ -0,0 +1,277 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { activityLog, agents, companies, companyMemberships, heartbeatRuns, issues, issueThreadInteractions, + connectionIntentDeliveries, toolApplications, toolConnections, connectionGrants, aiProviderDefaults, plugins, aiConnectionTaskPins, aiConnectionRouterCursors, createDb } from "@paperclipai/db"; +import type { AiConnectionRouterRequest } from "@paperclipai/shared"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { aiConnectionRouterService } from "../services/ai-connection-router.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { readAiConnectionConfigurationFailure } from "../services/ai-connection-configuration-failure.js"; +import { aiConnectionService } from "../services/ai-connections.js"; +import { connectionIntentService } from "../services/connection-intents.js"; +import { connectionIntentDeliveryService, wakeConnectionIntentAfterResolution } from "../services/connection-intent-delivery.js"; +import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js"; +import { evaluateAgentInvokabilityFromDb } from "../services/agent-invokability.js"; +import { configureAgentLifecycle } from "../services/agent-lifecycle.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("queued agent setup uses task connection cards", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("agent-setup-card-"); db = createDb(database.connectionString); }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + async function fixture(separateSetupUser = false) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), userId = randomUUID(); + const setupUserId = separateSetupUser ? randomUUID() : userId; + await db.insert(companies).values({ id: companyId, name: "Setup card", issuePrefix: `S${companyId.slice(0, 7)}` }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" }); + if (setupUserId !== userId) await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: setupUserId, status: "active", membershipRole: "member" }); + const binding = { mode: "responsible_user", provider: "openrouter", method: "api_key" } as const; + await db.insert(agents).values({ id: agentId, companyId, name: "Reviewer", status: "paused", lifecycleState: "verifying", lifecycleHolds: [], + lifecycleRequiredPluginIds: [], lifecycleOperation: { responsibleUserId: setupUserId, id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 1, retryRequired: true }, + adapterType: "paperclip_runner", adapterConfig: { provider: "opencode", model: "openrouter/deepseek/deepseek-v4.1-flash-1017" }, runtimeConfig: { aiConnection: binding } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Review after setup", status: "todo", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "queued", responsibleUserId: userId, contextSnapshot: { issueId } }); + const service = connectionIntentService(db); + const claims = { sub: agentId, company_id: companyId, run_id: runId, responsible_user_id: userId }; + return { companyId, agentId, issueId, runId, userId, setupUserId, binding, service, claims }; + } + async function repair(f: Awaited>, userId = f.setupUserId) { + const [app] = await db.insert(toolApplications).values({ companyId: f.companyId, applicationKey: `setup-openrouter-${userId}`, name: `OpenRouter ${userId}`, type: "mcp_http", metadata: { sourceTemplateKey: "openrouter" } }).returning(); + const [account] = await db.insert(toolConnections).values({ companyId: f.companyId, applicationId: app!.id, name: "Review account", uid: randomUUID(), connectionPurpose: "ai", transport: "runtime_auth", authKind: "api_key", credentialPolicy: "per_user", healthStatus: "ok", status: "active", enabled: true, config: { sourceTemplateKey: "openrouter", ai: { provider: "openrouter", method: "api_key" } } }).returning(); + const [grant] = await db.insert(connectionGrants).values({ companyId: f.companyId, connectionId: account!.id, kind: "user", subjectUserId: userId, createdByUserId: userId }).returning(); + await db.insert(aiProviderDefaults).values({ companyId: f.companyId, userId: userId, provider: "openrouter", grantId: grant!.id }); + return account!; + } + const cards = (issueId: string) => db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, issueId)); + + it.each(["verifying", "resuming"] as const)("repairs %s through one inline card and the same queued turn", async phase => { + const f = await fixture(); + if (phase === "resuming") await db.update(agents).set({ lifecycleState: phase, + lifecycleOperation: sql`${agents.lifecycleOperation} || '{"resumeState":"ready"}'::jsonb` }).where(eq(agents.id, f.agentId)); + await expect(f.service.validate(f.claims)).rejects.toThrow("no longer active"); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const pending = await cards(f.issueId); + expect(pending).toHaveLength(1); + expect((await db.select().from(activityLog).where(eq(activityLog.entityId, f.issueId))) + .filter(row => row.action === "issue.thread_interaction_created")).toHaveLength(1); + expect(pending[0]).toMatchObject({ kind: "connection_intent", status: "pending", sourceRunId: f.runId, addresseeUserId: f.userId, + payload: { purpose: "ai", serviceSlug: "openrouter", agentSetup: true } }); + expect((await f.service.setupOptions(pending[0]!.id)).aiConnection).toEqual(f.binding); + const account = await repair(f); + await f.service.complete(pending[0]!.id, account.id, f.userId); + const wakeup = vi.fn(); + const delivery = connectionIntentDeliveryService(db, { wakeup } as never); + await delivery.deliver(pending[0]!.id); + await delivery.deliver(pending[0]!.id); + expect(wakeup).not.toHaveBeenCalled(); + const [held] = await db.select().from(agents).where(eq(agents.id, f.agentId)); + expect(held!.lifecycleState).toBe(phase); + expect(held!.lifecycleOperation).not.toHaveProperty("retryRequired"); + const onReady = vi.fn(); + const host = vi.fn(async () => "complete" as const); + const worker = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runHost: host, runPlugin: async () => "complete" as const, onReady }); + try { await worker.process(f.agentId); await worker.process(f.agentId); } + finally { await worker.stop(); } + expect(host).toHaveBeenCalledOnce(); + expect(onReady).toHaveBeenCalledOnce(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agentId)))).toMatchObject([{ id: f.runId, status: "queued", startedAt: null }]); + expect((await db.select().from(connectionIntentDeliveries).where(eq(connectionIntentDeliveries.interactionId, pending[0]!.id)))[0]!.deliveredAt).not.toBeNull(); + }); + + it.each([{ taskAccountReady: false, companyDefault: false }, { taskAccountReady: true, companyDefault: false }, + { taskAccountReady: false, companyDefault: true }])("repairs the verifier's user without changing task authority: %j", async ({ taskAccountReady, companyDefault }) => { + const f = await fixture(true); + if (companyDefault) { + await db.update(companies).set({ defaultResponsibleUserId: f.setupUserId }).where(eq(companies.id, f.companyId)); + await db.update(agents).set({ lifecycleOperation: sql`${agents.lifecycleOperation} - 'responsibleUserId'` }).where(eq(agents.id, f.agentId)); + } + if (taskAccountReady) await repair(f, f.userId); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const [card] = await cards(f.issueId); + expect(card).toMatchObject({ addresseeUserId: f.setupUserId, sourceRunId: f.runId, payload: { agentSetup: true } }); + const account = await repair(f); + await f.service.complete(card!.id, account.id, f.setupUserId); + const wakeup = vi.fn(); + await connectionIntentDeliveryService(db, { wakeup } as never).deliver(card!.id); + const onReady = vi.fn(); + const worker = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runPlugin: async () => "complete" as const, + runHost: async agent => { + const [company] = await db.select().from(companies).where(eq(companies.id, f.companyId)); + const verificationUserId = agent.lifecycleOperation?.responsibleUserId ?? company!.defaultResponsibleUserId; + expect(verificationUserId).toBe(f.setupUserId); + await aiConnectionService(db).select({ companyId: f.companyId, agentId: f.agentId, + userId: verificationUserId, adapterType: "paperclip_runner", runnerProvider: "opencode", + model: "openrouter/deepseek/deepseek-v4.1-flash-1017", binding: f.binding }); + return "complete"; + }, onReady }); + try { await worker.process(f.agentId); } finally { await worker.stop(); } + expect(onReady).toHaveBeenCalledOnce(); + expect(wakeup).not.toHaveBeenCalled(); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(source).toMatchObject({ status: "queued", startedAt: null, responsibleUserId: f.userId }); + expect(source!.contextSnapshot).toEqual({ issueId: f.issueId }); + }); + + it("repairs a router's pinned lifecycle account without rerouting the task or changing its pool", async () => { + const f = await fixture(); + const account = await aiConnectionService(db).save(f.companyId, f.setupUserId, { provider: "openai", method: "api_key", + ownership: "personal", name: "Router setup account", apiKey: "fixture", agentIds: [], allAgents: true }, "fixture-key"); + const pluginKey = `fixture.setup-router-${f.companyId}`; + await db.insert(plugins).values({ pluginKey, packageName: pluginKey, version: "0.1.0", status: "ready", manifestJson: { + id: pluginKey, apiVersion: 1, version: "0.1.0", displayName: "Setup router", description: "Fixture", author: "Tests", + categories: ["connector"], capabilities: ["ai.connections.route"], aiConnectionRouter: { name: "Setup pool", description: "Saved accounts" }, entrypoints: { worker: "worker.js" } } }); + await instanceSettingsService(db).updateExperimental({ enableAiConnectionRouters: true }); + const route = vi.fn(async (_id: string, _method: string, input: AiConnectionRouterRequest) => + ({ kind: "selected", memberId: input.candidates[0]!.member.id })); + const router = aiConnectionRouterService(db, { isRunning: () => true, call: route } as unknown as PluginWorkerManager); + const binding = { mode: "delegated", provider: "openai", method: "api_key", ...account } as const; + const pool = await router.save(pluginKey, { companyId: f.companyId, config: { name: "Lifecycle setup pool", enabled: true, + mode: "round_robin", thresholdPercent: 90, members: [{ id: randomUUID(), binding, + profile: { provider: "codex", model: "gpt-5.6-sol", effort: "low" } }] } }, f.setupUserId); + const [agent] = await db.select().from(agents).where(eq(agents.id, f.agentId)); + const routerBinding = { mode: "router", connectionId: pool.id }; + await db.update(agents).set({ runtimeConfig: { aiConnection: routerBinding } }).where(eq(agents.id, f.agentId)); + const input = { companyId: f.companyId, poolId: pool.id, agentId: f.agentId, userId: f.setupUserId, + adapterType: "paperclip_runner", taskKey: `lifecycle:${agent!.lifecycleOperation!.id}` }; + const selected = await router.resolve(input); + await db.update(toolConnections).set({ healthStatus: "error" }).where(eq(toolConnections.id, account.connectionId)); + try { await router.resolve(input); throw new Error("Expected owned account refusal"); } + catch (error) { expect(readAiConnectionConfigurationFailure(error)).toBe("ai_connection_unavailable"); } + route.mockClear(); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const [card] = await cards(f.issueId); + expect(card).toMatchObject({ status: "pending", sourceRunId: f.runId, payload: { purpose: "ai", serviceSlug: "openai", agentSetup: true } }); + expect((await f.service.setupOptions(card!.id)).aiConnection).toEqual(binding); + await db.update(toolConnections).set({ healthStatus: "ok" }).where(eq(toolConnections.id, account.connectionId)); + const alternate = await aiConnectionService(db).save(f.companyId, f.setupUserId, { provider: "openai", method: "api_key", + ownership: "personal", name: "Unselected router account", apiKey: "fixture", agentIds: [], allAgents: true }, "fixture-alternate-key"); + await expect(f.service.complete(card!.id, alternate.connectionId, f.setupUserId)).rejects.toMatchObject({ status: 409 }); + await f.service.complete(card!.id, account.connectionId, f.setupUserId); + const wakeup = vi.fn(); + await connectionIntentDeliveryService(db, { wakeup } as never).deliver(card!.id); + const [repaired] = await db.select().from(agents).where(eq(agents.id, f.agentId)); + expect(repaired!.lifecycleOperation).not.toHaveProperty("retryRequired"); + expect(repaired!.runtimeConfig.aiConnection).toEqual(routerBinding); + expect((await db.select().from(aiConnectionTaskPins).where(eq(aiConnectionTaskPins.agentId, f.agentId)))) + .toMatchObject([{ taskKey: input.taskKey, selection: selected }]); + expect((await db.select().from(aiConnectionRouterCursors).where(eq(aiConnectionRouterCursors.poolId, pool.id)))[0]!.version).toBe(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]!.contextSnapshot).toEqual({ issueId: f.issueId }); + expect(route).not.toHaveBeenCalled(); + expect(wakeup).not.toHaveBeenCalled(); + }); + + it("supersedes a setup card when the lifecycle operation changes on the same queued turn", async () => { + const f = await fixture(); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const [old] = await cards(f.issueId); + const operationId = randomUUID(); + await db.update(agents).set({ lifecycleOperation: sql`${agents.lifecycleOperation} || ${JSON.stringify({ id: operationId })}::jsonb` }).where(eq(agents.id, f.agentId)); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const all = await cards(f.issueId); + expect(all).toHaveLength(2); + expect(all.find(card => card.id === old!.id)).toMatchObject({ status: "expired", result: { outcome: "superseded" } }); + expect(all.find(card => card.status === "pending")).toMatchObject({ sourceRunId: f.runId, + payload: { agentSetupOperationId: operationId } }); + await expect(f.service.setupOptions(old!.id)).rejects.toMatchObject({ status: 409 }); + }); + + it("replaces a stopped turn's setup card and repairs the current queued turn", async () => { + const f = await fixture(); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const [stoppedCard] = await cards(f.issueId); + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + const [next] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "queued", responsibleUserId: f.userId, contextSnapshot: { issueId: f.issueId } }).returning(); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const current = await cards(f.issueId); + expect(current).toHaveLength(2); + const replacement = current.find(card => card.sourceRunId === next!.id)!; + expect(replacement).toMatchObject({ status: "pending", payload: { agentSetup: true } }); + expect(current.find(card => card.id === stoppedCard!.id)).toMatchObject({ status: "expired", + result: { outcome: "superseded", supersededByInteractionId: replacement.id } }); + const account = await repair(f); + await f.service.complete(replacement.id, account.id, f.userId); + const wakeup = vi.fn(); + await connectionIntentDeliveryService(db, { wakeup } as never).deliver(replacement.id); + expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0]!.lifecycleOperation) + .not.toHaveProperty("retryRequired"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "cancelled", startedAt: null }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, next!.id)))[0]) + .toMatchObject({ status: "queued", startedAt: null, responsibleUserId: f.userId }); + expect(wakeup).not.toHaveBeenCalled(); + }); + + it.each(["retry", "continuation"] as const)("restores an inline setup card through full recovery for a retained %s", async kind => { + const f = await fixture(); + if (kind === "retry") { + const [prior] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "failed", finishedAt: new Date(), contextSnapshot: { issueId: f.issueId } }).returning(); + await db.update(heartbeatRuns).set({ retryOfRunId: prior!.id }).where(eq(heartbeatRuns.id, f.runId)); + } else await db.update(heartbeatRuns).set({ continuationAttempt: 1 }).where(eq(heartbeatRuns.id, f.runId)); + const deps = createHeartbeatQueueTestDependencies(db); + deps.getAgent.mockImplementation(async id => (await db.select().from(agents).where(eq(agents.id, id)))[0] ?? null); + deps.getAgentInvokability.mockImplementation(agent => evaluateAgentInvokabilityFromDb(db, agent)); + const queue = createHeartbeatQueue(db, deps); + await queue.startNextQueuedRunForAgent(f.agentId); + expect(await cards(f.issueId)).toHaveLength(0); + await queue.resumeFreshQueuedRunsAtStartup(); + expect(await cards(f.issueId)).toHaveLength(0); + expect(deps.executeRun).not.toHaveBeenCalledWith(f.runId); + // The global startup pass may dispatch unrelated ready fixtures. The full + // recovery below is scoped to this retained turn’s company. + deps.executeRun.mockClear(); + await queue.resumeQueuedRunsAfterDrain({ companyId: f.companyId }); + await queue.resumeQueuedRunsAfterDrain({ companyId: f.companyId }); + expect(await cards(f.issueId)).toMatchObject([{ status: "pending", sourceRunId: f.runId, payload: { agentSetup: true } }]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "queued", startedAt: null, responsibleUserId: f.userId }); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + + it.each(["decline", "stop", "reassign", "setup_owner_changed", "setup_operation_changed"] as const)("does not dispatch or lift verification after %s", async action => { + const f = await fixture(); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + const [card] = await cards(f.issueId); + if (action === "decline") await f.service.decline(card!.id, f.userId); + else { + const account = await repair(f); + await f.service.complete(card!.id, account.id, f.userId); + if (action === "stop") await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + else if (action === "reassign") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId)); + else if (action === "setup_owner_changed") await db.update(agents).set({ lifecycleOperation: sql`${agents.lifecycleOperation} || ${JSON.stringify({ responsibleUserId: randomUUID() })}::jsonb` }).where(eq(agents.id, f.agentId)); + else await db.update(agents).set({ lifecycleOperation: sql`${agents.lifecycleOperation} || ${JSON.stringify({ id: randomUUID() })}::jsonb` }).where(eq(agents.id, f.agentId)); + } + const wakeup = vi.fn(); + await connectionIntentDeliveryService(db, { wakeup } as never).deliver(card!.id); + expect(wakeup).not.toHaveBeenCalled(); + expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0]!.lifecycleOperation?.retryRequired).toBe(true); + expect((await db.select().from(connectionIntentDeliveries).where(eq(connectionIntentDeliveries.interactionId, card!.id)))[0]!.deliveredAt).not.toBeNull(); + }); + + it.each(["manual_pause", "started", "foreign_company"] as const)("does not create a setup card for %s", async state => { + const f = await fixture(); + if (state === "manual_pause") await db.update(agents).set({ lifecycleState: "paused", lifecycleHolds: ["manual"] }).where(eq(agents.id, f.agentId)); + if (state === "started") await db.update(heartbeatRuns).set({ startedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + await f.service.requestForQueuedAgentSetup(state === "foreign_company" ? randomUUID() : f.companyId, f.agentId); + expect(await cards(f.issueId)).toHaveLength(0); + }); + + it("does not disclose a private task to a different setup account owner", async () => { + const f = await fixture(true); + await db.update(issues).set({ visibility: "private", responsibleUserId: f.userId }).where(eq(issues.id, f.issueId)); + await f.service.requestForQueuedAgentSetup(f.companyId, f.agentId); + expect(await cards(f.issueId)).toHaveLength(0); + }); + + it("keeps setup-card outcomes out of the generic provider wake path", async () => { + const wakeup = vi.fn(); + await wakeConnectionIntentAfterResolution({ wakeup } as never, { loaded: { issue: { id: "task", status: "in_progress", assigneeAgentId: "agent" }, interaction: { id: "card", payload: { purpose: "ai", agentSetup: true } } }, status: "accepted", actorId: "user" }); + expect(wakeup).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/lib/ai-connection-configuration-failure.ts b/server/src/lib/ai-connection-configuration-failure.ts new file mode 100644 index 00000000000..5acb51c6d62 --- /dev/null +++ b/server/src/lib/ai-connection-configuration-failure.ts @@ -0,0 +1,24 @@ +const reasons = [ + "ai_connection_responsible_user_missing", + "ai_connection_default_missing", + "ai_connection_missing", + "ai_connection_incompatible", + "ai_connection_unavailable", + "ai_connection_credential_not_shared", +] as const; +export type AiConnectionConfigurationReason = typeof reasons[number]; +const failures = new WeakMap(); + +/** Owned selection evidence shared with lifecycle policy, without HTTP dependencies. */ +export function recordAiConnectionConfigurationFailure(error: Error, reason: AiConnectionConfigurationReason) { + failures.set(error, reason); +} + +/** Only owned producers supply this evidence; matching names or HTTP codes do not. */ +export function readAiConnectionConfigurationFailure(error: unknown): AiConnectionConfigurationReason | null { + return error instanceof Error ? failures.get(error) ?? null : null; +} + +export function isAiConnectionConfigurationReason(value: unknown): value is AiConnectionConfigurationReason { + return typeof value === "string" && reasons.some((reason) => reason === value); +} diff --git a/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts b/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts index 0d295e2b37e..b78dcab5de3 100644 --- a/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts @@ -1,3 +1,5 @@ +import { aiConnectionConfigurationFailure, aiConnectionCredentialNotSharedFailure } from "../../../services/ai-connection-configuration-failure.js"; +import { unprocessable } from "../../../errors.js"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { Db } from "@paperclipai/db"; import { createLifecycleWorker } from "../application/worker.js"; @@ -85,6 +87,23 @@ describe("lifecycle failure diagnostics", () => { expect(lifecycleVerificationNeedsRetry({ stage: "host", phase: "cleaning_up" }, error)).toBe(false); }); + it.each([ + "ai_connection_responsible_user_missing", "ai_connection_default_missing", "ai_connection_missing", + "ai_connection_incompatible", "ai_connection_unavailable", "ai_connection_credential_not_shared", + ] as const)("holds owned AI selection failures with safe guidance: %s", reason => { + const error = reason === "ai_connection_credential_not_shared" ? aiConnectionCredentialNotSharedFailure() + : aiConnectionConfigurationFailure(reason, "private provider detail", { connectionId: "private-id" }); + const context = { stage: "host" as const, phase: "verifying" as const }; + expect(lifecycleVerificationNeedsRetry(context, error)).toBe(true); + const message = lifecycleFailureMessage(context, error); + expect(message).toContain("Automatic verification is paused"); + expect(message).not.toMatch(/private|retry automatically/); + expect(lifecycleVerificationNeedsRetry({ stage: "plugin", phase: "verifying" }, error)).toBe(false); + // Only the selector can establish this evidence; unrelated HTTP errors still retry. + expect(lifecycleVerificationNeedsRetry(context, unprocessable("same text", { code: reason }))).toBe(false); + expect(lifecycleVerificationNeedsRetry(context, { code: reason })).toBe(false); + }); + it("allows two automatic retries of a generic failed test before waiting for repair", () => { for (const verificationFailures of [1, 2, 3, 8_000]) { const context = { stage: "host" as const, phase: "verifying" as const, attempts: 8_000, verificationFailures }; diff --git a/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts b/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts index 3976bf71e34..23dfeb303bd 100644 --- a/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { agents, companies, createDb, applyPendingMigrations, closeRegisteredClients, type Db } from "@paperclipai/db"; +import { agents, companies, companyMemberships, createDb, applyPendingMigrations, closeRegisteredClients, type Db } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; @@ -12,6 +12,8 @@ import type { LifecycleAgent } from "../application/ports.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; +import { aiConnectionService } from "../../../services/ai-connections.js"; + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_LIFECYCLE_VERIFICATION_RETRY_TEST_DATABASE_URL"); const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("failed lifecycle verification retries", () => { @@ -53,6 +55,80 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu return { agent: await current(agent.id), work, driver, runHost }; } + it("parks a real missing-default selection across restart and permits repair through the normal ready handoff", async () => { + const userId = randomUUID(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active" }); + let repaired = false; + const runHost = vi.fn(async (agent: LifecycleAgent) => { + if (agent.lifecycleState === "verifying" && !repaired) await aiConnectionService(db).select({ + companyId, userId, agentId: agent.id, adapterType: "paperclip_runner", runnerProvider: "opencode", + model: "openrouter/deepseek/deepseek-v4.1-flash-1017", + binding: { mode: "responsible_user", provider: "openrouter", method: "api_key" }, + }); + return "complete" as const; + }); + const onReady = vi.fn(); + const driver = { requiredPluginIds: async () => [], runHost, runPlugin: async () => "complete" as const, onReady }; + const work = configureAgentLifecycle(db, driver); workers.push(work); + const hired = await createAgentLifecycle(db).requestHire(companyId, { name: "Missing default", adapterType: "process" }); + await work.process(hired.id); + const held = await current(hired.id); + expect(held).toMatchObject({ lifecycleState: "verifying", status: "paused", lifecycleOperation: { retryRequired: true } }); + expect(held.lifecycleError).toContain("personal default"); + expect(held.lifecycleError).not.toContain("retry automatically"); + expect(held.lifecycleOperation).not.toHaveProperty("retryAt"); + expect(onReady).not.toHaveBeenCalled(); + await work.stop(); + const restarted = startAgentLifecycle(db, createAgentLifecycleEffects(), driver, () => true); workers.push(restarted); + const calls = runHost.mock.calls.length; + expect(await restarted.hasPending()).toBe(false); + expect(restarted.nextRunAt()).toBeNull(); + await restarted.sweep(); + expect(runHost).toHaveBeenCalledTimes(calls); + // Repair and Retry must still pass verification before the normal ready handoff. + repaired = true; + await createAgentLifecycle(db).retry(hired.id); + await restarted.sweep(); + expect(await current(hired.id)).toMatchObject({ lifecycleState: "ready", status: "idle", lifecycleError: null }); + expect(onReady).toHaveBeenCalledOnce(); + }); + + it("parks owned account failures after pause/resume and accepts a durable connection repair recheck", async () => { + const userId = randomUUID(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active" }); + let repaired = false; + const runHost = vi.fn(async (agent: LifecycleAgent) => { + if (agent.lifecycleState === "resuming" && !repaired) await aiConnectionService(db).select({ companyId, userId, + agentId: agent.id, adapterType: "paperclip_runner", runnerProvider: "opencode", + model: "openrouter/deepseek/deepseek-v4.1-flash-1017", binding: { mode: "responsible_user", provider: "openrouter", method: "api_key" } }); + return "complete" as const; + }); + const onReady = vi.fn(); + const driver = { requiredPluginIds: async () => [], runHost, runPlugin: async () => "complete" as const, onReady }; + const lifecycle = createAgentLifecycle(db), work = configureAgentLifecycle(db, driver); workers.push(work); + const hired = await lifecycle.requestHire(companyId, { name: "Resume account repair", adapterType: "process" }); + await work.process(hired.id); + await lifecycle.pauseAgent(hired.id); await work.process(hired.id); + expect((await current(hired.id)).lifecycleState).toBe("paused"); + onReady.mockClear(); + await lifecycle.resumeAgent(hired.id); await work.process(hired.id); + const held = await current(hired.id); + expect(held).toMatchObject({ lifecycleState: "resuming", lifecycleOperation: { retryRequired: true } }); + expect(held.lifecycleOperation).not.toHaveProperty("retryAt"); + await work.stop(); + const restarted = startAgentLifecycle(db, createAgentLifecycleEffects(), driver, () => true); workers.push(restarted); + const calls = runHost.mock.calls.length; + expect(await restarted.hasPending()).toBe(false); + expect(restarted.nextRunAt()).toBeNull(); + await restarted.sweep(); expect(runHost).toHaveBeenCalledTimes(calls); + repaired = true; + expect(await db.transaction(tx => requestAgentLifecycleRecheckInTransaction(tx, { companyId, agentId: hired.id, + configurationRepaired: true }, () => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle)))).toBe(true); + await restarted.sweep(); + expect((await current(hired.id)).lifecycleState).toBe("ready"); + expect(onReady).toHaveBeenCalledOnce(); + }); + it("persists a failed test across worker restarts without scanning, claiming, or probing it again", async () => { const { agent, work, driver, runHost } = await failed(); expect(agent).toMatchObject({ lifecycleState: "verifying", status: "paused", lifecycleOperation: { retryRequired: true } }); @@ -75,6 +151,28 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect((await current(agent.id)).lifecycleOperation).toEqual(agent.lifecycleOperation); }); + it("does not let an in-flight failed verifier overwrite a completed connection repair", async () => { + const [agent] = await db.insert(agents).values({ companyId, name: "Repair while checking", status: "paused", + lifecycleState: "verifying", lifecycleHolds: [], lifecycleRequiredPluginIds: [], + lifecycleOperation: { id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 0, verificationFailures: 2 } }).returning(); + const store = createLifecycleStore(db, createAgentLifecycleEffects()); + const owner = "old-account-check"; + const claimed = await store.claim(agent!.id, owner, new Date()); + expect(claimed).not.toBeNull(); + await db.transaction(async tx => { + expect(await requestAgentLifecycleRecheckInTransaction(tx, { companyId, agentId: agent!.id, + expectedVersion: agent!.lifecycleVersion, configurationRepaired: true }, + () => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle))).toBe(true); + }); + await store.defer(claimed!, owner, "Old missing account", new Date(), { retryRequired: true }); + const repaired = await current(agent!.id); + expect(repaired.lifecycleError).toBeNull(); + expect(repaired.lifecycleOperation).not.toHaveProperty("retryRequired"); + expect(repaired.lifecycleOperation).not.toHaveProperty("retryAt"); + expect(repaired.lifecycleOperation).not.toHaveProperty("verificationFailures"); + expect(await store.claim(agent!.id, "new-check", new Date())).not.toBeNull(); + }); + it("permits an explicit retry of the same operation and completes verification before readiness", async () => { const { agent, work, runHost } = await failed(); let release!: () => void; diff --git a/server/src/modules/agent-lifecycle/adapters/postgres.ts b/server/src/modules/agent-lifecycle/adapters/postgres.ts index 9c113b14569..6378e51dd05 100644 --- a/server/src/modules/agent-lifecycle/adapters/postgres.ts +++ b/server/src/modules/agent-lifecycle/adapters/postgres.ts @@ -17,7 +17,7 @@ type ChangeOptions = { version?: number; owner?: string; reason?: string }; export const pendingLifecycleWork = and( inArray(agents.lifecycleState, ["preparing", "verifying", "pausing", "resuming", "terminating", "cleaning_up"]), sql`not (${agents.lifecycleState} = 'verifying' and ${agents.lifecycleHolds} @> '["setup"]'::jsonb)`, - sql`not (${agents.lifecycleState} = 'verifying' and coalesce(${agents.lifecycleOperation}->>'retryRequired', 'false') = 'true')`, + sql`not (${agents.lifecycleState} in ('verifying', 'resuming') and coalesce(${agents.lifecycleOperation}->>'retryRequired', 'false') = 'true')`, sql`not (${agents.lifecycleState} = 'verifying' and coalesce(${agents.lifecycleOperation}->>'waitingForSignal', 'false') = 'true')`, ); @@ -58,15 +58,15 @@ export async function releaseAgentLifecycleLeaseOfHolderInTransaction( export async function requestAgentLifecycleRecheckInTransaction( tx: Pick, - input: { companyId: string; agentId: string; expectedVersion?: number; onlyIfWaiting?: boolean }, + input: { companyId: string; agentId: string; expectedVersion?: number; onlyIfWaiting?: boolean; configurationRepaired?: boolean }, notify: () => Promise, ): Promise { if (!("nestedIndex" in tx)) throw new AgentLifecycleConflict("Lifecycle rechecks require the producer transaction"); const [agent] = await tx.select().from(agents).where(and( eq(agents.id, input.agentId), eq(agents.companyId, input.companyId), )).for("update"); - if (!agent || !["preparing", "verifying"].includes(agent.lifecycleState) - || agent.lifecycleOperation?.retryRequired + if (!agent || !["preparing", "verifying", "resuming"].includes(agent.lifecycleState) + || agent.lifecycleOperation?.retryRequired && !input.configurationRepaired || input.expectedVersion !== undefined && agent.lifecycleVersion !== input.expectedVersion || input.onlyIfWaiting && !agent.lifecycleOperation?.waitingForSignal) return false; await notify(); @@ -75,7 +75,9 @@ export async function requestAgentLifecycleRecheckInTransaction( await tx.update(agents).set({ lifecycleError: null, lifecycleOperation: owned ? sql`${agents.lifecycleOperation} || '{"recheckRequested":true}'::jsonb` - : sql`${agents.lifecycleOperation} - 'waitingForSignal' - 'recheckRequested' - 'retryAt'`, + : input.configurationRepaired + ? sql`${agents.lifecycleOperation} - 'waitingForSignal' - 'recheckRequested' - 'retryAt' - 'retryRequired' - 'verificationFailures'` + : sql`${agents.lifecycleOperation} - 'waitingForSignal' - 'recheckRequested' - 'retryAt'`, updatedAt: new Date(), }).where(and(eq(agents.id, agent.id), eq(agents.companyId, input.companyId), eq(agents.lifecycleVersion, agent.lifecycleVersion))); return true; @@ -266,16 +268,19 @@ export function createLifecycleStore(db: Db, effects: LifecycleEffects): Lifecyc sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id }); return rows.length > 0; } - const retry: Partial = options?.retryRequired && agent.lifecycleState === "verifying" + const retry: Partial = options?.retryRequired && ["verifying", "resuming"].includes(agent.lifecycleState) ? { retryRequired: true } : { retryAt: new Date(now.getTime() + (error ? 60_000 : 2_000)).toISOString() }; if (options?.verificationFailures !== undefined && agent.lifecycleState === "verifying") { retry.verificationFailures = options.verificationFailures; } const rows = await db.update(agents).set({ - lifecycleError: error, - lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'leaseHolder' - 'retryAt' - 'retryRequired' - 'waitingForSignal' - 'recheckRequested') - || ${JSON.stringify(retry)}::jsonb`, + lifecycleError: sql`case when coalesce(${agents.lifecycleOperation}->>'recheckRequested', 'false') = 'true' then null else ${error}::text end`, + lifecycleOperation: sql`((case when coalesce(${agents.lifecycleOperation}->>'recheckRequested', 'false') = 'true' + then ${agents.lifecycleOperation} - 'verificationFailures' else ${agents.lifecycleOperation} end) + - 'leaseOwner' - 'leaseUntil' - 'leaseHolder' - 'retryAt' - 'retryRequired' - 'waitingForSignal' - 'recheckRequested') + || case when coalesce(${agents.lifecycleOperation}->>'recheckRequested', 'false') = 'true' + then '{}'::jsonb else ${JSON.stringify(retry)}::jsonb end`, updatedAt: now, }).where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion), sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id }); diff --git a/server/src/modules/agent-lifecycle/application/failure.ts b/server/src/modules/agent-lifecycle/application/failure.ts index 905f7fd203a..648073b356e 100644 --- a/server/src/modules/agent-lifecycle/application/failure.ts +++ b/server/src/modules/agent-lifecycle/application/failure.ts @@ -1,6 +1,15 @@ +import { readAiConnectionConfigurationFailure } from "../../../lib/ai-connection-configuration-failure.js"; import type { LifecycleFailure } from "./ports.js"; // Only fixed text may leave this boundary. Provider errors can contain secrets. +const connectionReasons = { + ai_connection_responsible_user_missing: "Set the responsible user before selecting this agent's AI connection.", + ai_connection_default_missing: "Connect an AI account and choose the responsible user's personal default for this provider.", + ai_connection_missing: "The selected AI connection is unavailable. Choose an available connection in Configuration.", + ai_connection_incompatible: "Choose an AI connection compatible with this agent's harness and model in Configuration.", + ai_connection_unavailable: "Reconnect or validate the selected AI account in Configuration.", + ai_connection_credential_not_shared: "The credential is not shared with the responsible user. Ask its owner for access or choose an authorized connection.", +} as const; const reasons = { harness_test_failed: "The execution test did not pass. Open Configuration and run the agent test to inspect the failed checks.", ai_connection_required: "This agent's environment requires a managed AI connection and an isolated Runner home. Check both in Configuration, then retry.", @@ -24,11 +33,14 @@ const configurationFailures = new Set([ export function lifecycleVerificationNeedsRetry(context: LifecycleFailure, error: unknown): boolean { const code = lifecycleFailureCode(error) ?? ""; - return context.phase === "verifying" && context.stage === "host" - && (configurationFailures.has(code) || (code === "harness_test_failed" && (context.verificationFailures ?? 0) >= 3)); + return ["verifying", "resuming"].includes(context.phase ?? "") && context.stage === "host" + && (readAiConnectionConfigurationFailure(error) !== null || context.phase === "verifying" + && (configurationFailures.has(code) || (code === "harness_test_failed" && (context.verificationFailures ?? 0) >= 3))); } export function lifecycleFailureCode(error: unknown): string | undefined { + const selectionFailure = readAiConnectionConfigurationFailure(error); + if (selectionFailure) return selectionFailure; const candidate = error as { code?: unknown; cause?: { code?: unknown } } | null; return [candidate?.code, candidate?.cause?.code].find((code): code is string => typeof code === "string" && (Object.hasOwn(reasons, code) || transientCodes.has(code))); @@ -46,7 +58,8 @@ export function lifecycleFailureMessage(context: LifecycleFailure, error: unknow : ["save_plugins", "complete_host", "complete_plugin", "transition"].includes(context.stage) ? `${phase}: saving progress` : phase; const code = lifecycleFailureCode(error); - const reason = code && Object.hasOwn(reasons, code) ? reasons[code as keyof typeof reasons] + const reason = code && Object.hasOwn(connectionReasons, code) ? connectionReasons[code as keyof typeof connectionReasons] + : code && Object.hasOwn(reasons, code) ? reasons[code as keyof typeof reasons] : "Paperclip could not complete this step. It will retry automatically. If this continues, contact your instance administrator."; return `${step} failed. ${reason}${lifecycleVerificationNeedsRetry(context, error) ? " Automatic verification is paused. Update Configuration or choose Retry now." : ""}`; diff --git a/server/src/services/agent-harness-verification.ts b/server/src/services/agent-harness-verification.ts index 05f8bc8abf5..199d9fc6fd5 100644 --- a/server/src/services/agent-harness-verification.ts +++ b/server/src/services/agent-harness-verification.ts @@ -1,3 +1,6 @@ +import { connectionIntentService } from "./connection-intents.js"; +import { readAiConnectionConfigurationFailure } from "./ai-connection-configuration-failure.js"; +import { logger } from "../middleware/logger.js"; import { createHmac } from "node:crypto"; import { and, eq, inArray } from "drizzle-orm"; import { agents, companies, companySecrets, type Db } from "@paperclipai/db"; @@ -195,5 +198,13 @@ export function agentHarnessVerificationService(db: Db, manager: PluginWorkerMan return "complete"; } finally { await target.release(status); } } - return { verify }; + return { async verify(snapshot: LifecycleAgent) { + try { return await verify(snapshot); } + catch (error) { + if (readAiConnectionConfigurationFailure(error)) await connectionIntentService(db) + .requestForQueuedAgentSetup(snapshot.companyId, snapshot.id) + .catch(() => logger.warn({ agentId: snapshot.id }, "Could not attach queued agent setup connection request")); + throw error; + } + } }; } diff --git a/server/src/services/ai-connection-configuration-failure.ts b/server/src/services/ai-connection-configuration-failure.ts index 06c76a17f3f..a31275b3185 100644 --- a/server/src/services/ai-connection-configuration-failure.ts +++ b/server/src/services/ai-connection-configuration-failure.ts @@ -1,15 +1,7 @@ import { forbidden, unprocessable } from "../errors.js"; +import { recordAiConnectionConfigurationFailure, type AiConnectionConfigurationReason } from "../lib/ai-connection-configuration-failure.js"; +export { readAiConnectionConfigurationFailure, isAiConnectionConfigurationReason } from "../lib/ai-connection-configuration-failure.js"; -const reasons = [ - "ai_connection_responsible_user_missing", - "ai_connection_default_missing", - "ai_connection_missing", - "ai_connection_incompatible", - "ai_connection_unavailable", - "ai_connection_credential_not_shared", -] as const; -type AiConnectionConfigurationReason = typeof reasons[number]; -const failures = new WeakMap(); type CredentialAccessFailure = { connectionName: string; grantId?: string }; const credentialAccessFailures = new WeakMap(); @@ -20,14 +12,14 @@ export function aiConnectionConfigurationFailure( details: Record = {}, ) { const error = unprocessable(message, { ...details, code: reason }); - failures.set(error, reason); + recordAiConnectionConfigurationFailure(error, reason); return error; } /** A verified active user lacks this credential's human sharing permission. */ export function aiConnectionCredentialNotSharedFailure(context?: CredentialAccessFailure) { const error = forbidden("This credential is not shared with the responsible user"); - failures.set(error, "ai_connection_credential_not_shared"); + recordAiConnectionConfigurationFailure(error, "ai_connection_credential_not_shared"); if (context) credentialAccessFailures.set(error, { ...context }); return error; } @@ -36,12 +28,3 @@ export function aiConnectionCredentialNotSharedFailure(context?: CredentialAcces export function readAiCredentialAccessFailure(error: unknown): CredentialAccessFailure | null { return error instanceof Error ? credentialAccessFailures.get(error) ?? null : null; } - -/** Only owned producers supply this evidence; matching names or HTTP codes do not. */ -export function readAiConnectionConfigurationFailure(error: unknown): AiConnectionConfigurationReason | null { - return error instanceof Error ? failures.get(error) ?? null : null; -} - -export function isAiConnectionConfigurationReason(value: unknown): value is AiConnectionConfigurationReason { - return typeof value === "string" && reasons.some((reason) => reason === value); -} diff --git a/server/src/services/connection-intent-delivery.ts b/server/src/services/connection-intent-delivery.ts index 561cd6e7ba0..02ba53c3bef 100644 --- a/server/src/services/connection-intent-delivery.ts +++ b/server/src/services/connection-intent-delivery.ts @@ -1,3 +1,6 @@ +import { isAgentAwaitingSetup, requestAgentLifecycleRecheckInTransaction } from "../modules/agent-lifecycle/index.js"; +import { agents, companies } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { connectionIntentService } from "./connection-intents.js"; import { connectionContinuationPendingResponse, findSatisfiedToolConnection, satisfiedConnectionIntentService } from "./satisfied-connection-intents.js"; import { isAiConnectionConfigurationFailure } from "./ai-auth-failure.js"; @@ -27,6 +30,9 @@ export async function wakeConnectionIntentAfterResolution( const resolvedAt = input.loaded.interaction.resolvedAt; const interactionResolvedAt = resolvedAt instanceof Date ? resolvedAt.toISOString() : resolvedAt; const payload = input.loaded.interaction.payload; + // Setup repair resumes the accepted queued turn through lifecycle readiness. + // It never creates an additional provider turn through the generic wake route. + if (payload !== null && typeof payload === "object" && "agentSetup" in payload && payload.agentSetup === true) return; const repairsProviderAuthentication = payload !== null && typeof payload === "object" && "purpose" in payload && payload.purpose === "ai"; await heartbeat.wakeup(agentId, { source: "automation", @@ -131,7 +137,7 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick { + const [task] = await tx.select().from(issues).where(and(eq(issues.id, loaded.issue.id), eq(issues.companyId, claimed.companyId))).for("update"); + const [source] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, interaction!.sourceRunId!), eq(heartbeatRuns.companyId, claimed.companyId), + eq(heartbeatRuns.agentId, payload.requestingAgentId!), + )).for("no key update"); + const [agent] = await tx.select().from(agents).where(and(eq(agents.id, payload.requestingAgentId!), eq(agents.companyId, claimed.companyId))).for("update"); + const [company] = await tx.select({ defaultResponsibleUserId: companies.defaultResponsibleUserId }) + .from(companies).where(eq(companies.id, claimed.companyId)); + const verificationUserId = agent?.lifecycleOperation?.responsibleUserId ?? company?.defaultResponsibleUserId; + if (interaction!.status === "accepted" && task && !["done", "cancelled"].includes(task.status) + && task.assigneeAgentId === payload.requestingAgentId && source?.status === "queued" && !source.startedAt + && source.runtimeMode === "legacy" && agent && isAgentAwaitingSetup(agent) + && verificationUserId === interaction!.addresseeUserId + && (!payload.agentSetupOperationId || agent.lifecycleOperation?.id === payload.agentSetupOperationId)) { + const ready = await connectionIntentService(db).usableConnectionForAgent({ companyId: claimed.companyId, + agentId: payload.requestingAgentId!, responsibleUserId: userId!, serviceSlug: payload.serviceSlug!, purpose: "ai", + sourceRunId: interaction!.sourceRunId, agentSetup: true, agentSetupOperationId: payload.agentSetupOperationId }); + if (!ready) return; + await requestAgentLifecycleRecheckInTransaction(tx, { companyId: claimed.companyId, agentId: agent.id, + configurationRepaired: true }, () => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle)); + } + // Decline, Stop and stale ownership also settle this delivery without a wake. + await tx.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; + } if (retiredForAvailableConnection) { const publications: ActivityPublication[] = []; const current = await db.transaction(async tx => { diff --git a/server/src/services/connection-intents.ts b/server/src/services/connection-intents.ts index 683de9b6ce5..7bb35f1c511 100644 --- a/server/src/services/connection-intents.ts +++ b/server/src/services/connection-intents.ts @@ -1,3 +1,6 @@ +import { canActorReadIssuePrivacy } from "./authorization.js"; +import { isAgentAwaitingSetup } from "../modules/agent-lifecycle/index.js"; +import { readAiConnectionConfigurationFailure } from "./ai-connection-configuration-failure.js"; import { assertAgentConnectionInstallAccess, connectionManagementService } from "./connection-management.js"; import { CHAT_HUMAN_CREDENTIAL_FIELDS, chatCredentialEntryGenerations } from "./chat-channels/credentials.js"; import { updateAgentConfigurationInTransaction } from "./agent-configuration-transaction.js"; @@ -6,11 +9,12 @@ import { emailConnectionService } from "./email-connections.js"; import { grantConnectionAgentTools } from "./connection-agent-access.js"; import { agentService } from "./agents.js"; import { createHash } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; import { logActivity } from "./activity-log.js"; import { aiConnectionService } from "./ai-connections.js"; import { aiConnectionBindingSchema, aiConnectionRouterBindingSchema } from "@paperclipai/shared"; import { aiBindingForAuthRecovery, isAiAuthenticationFailure, isAiAuthenticationRepairable } from "./ai-auth-failure.js"; -import { and, eq, desc, isNull, sql } from "drizzle-orm"; +import { and, eq, desc, isNull, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agents, @@ -56,7 +60,7 @@ import { } from "@paperclipai/shared"; import { instanceSettingsService } from "./instance-settings.js"; import { createNativeConnectorYield } from "./connection-native-yield.js"; -import { conflict, forbidden, notFound, unprocessable } from "../errors.js"; +import { HttpError, conflict, forbidden, notFound, unprocessable } from "../errors.js"; import type { RuntimeToolsTokenClaims } from "../runtime-tools-token.js"; import { issueThreadInteractionService, type InteractionResolutionMutationOptions } from "./issue-thread-interactions.js"; import { lockToolConnectionRuntimeInputs, toolAccessService } from "./tool-access.js"; @@ -199,13 +203,15 @@ export function connectionIntentService(db: Db) { } } - async function loadRunContext(claims: ConnectionRunClaims, failedAuthRun = false) { + async function loadRunContext(claims: ConnectionRunClaims, mode: "active" | "failed_auth" | "queued_setup" = "active") { let run = await db .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, + startedAt: heartbeatRuns.startedAt, + runtimeMode: heartbeatRuns.runtimeMode, errorCode: heartbeatRuns.errorCode, resultJson: heartbeatRuns.resultJson, responsibleUserId: heartbeatRuns.responsibleUserId, @@ -222,10 +228,12 @@ export function connectionIntentService(db: Db) { || run.agentId !== claims.sub || (!run.activeIdentityContextId && run.responsibleUserId !== claims.responsible_user_id) ) throw forbidden("Runtime tool token does not match its heartbeat run"); - if (failedAuthRun ? run.status !== "failed" || !isAiAuthenticationRepairable(run) : run.status !== "running") { + if (mode === "failed_auth" ? run.status !== "failed" || !isAiAuthenticationRepairable(run) + : mode === "queued_setup" ? run.status !== "queued" || run.startedAt !== null || run.runtimeMode !== "legacy" || run.activeIdentityContextId !== null + : run.status !== "running") { throw forbidden("Runtime tool token is no longer active"); } - if (run.activeIdentityContextId && !failedAuthRun) { + if (run.activeIdentityContextId && mode === "active") { const current = await captureRunIdentity(db, { companyId: run.companyId, agentId: run.agentId, runId: run.id }, { readScope: "identity" }); run = { ...run, responsibleUserId: current.run.responsibleUserId }; } @@ -281,9 +289,38 @@ export function connectionIntentService(db: Db) { }; } - async function managedAgent(companyId: string, agentId: string, serviceSlug: string, options: { fallback?: AiConnectionBinding; sourceRunId?: string | null } = {}, client = db) { + // Reuse host verification's durable lifecycle allocation. This read must not + // allocate a task route, advance a pool cursor, or replace the agent's pool. + async function queuedSetupManagedAgent(agent: typeof agents.$inferSelect, client = db) { + const saved = aiConnectionBindingSchema.safeParse(agent.runtimeConfig.aiConnection).data; + if (saved) return { agent, binding: saved, adapterConfig: agent.adapterConfig, requiresAdoption: false }; + const router = aiConnectionRouterBindingSchema.safeParse(agent.runtimeConfig.aiConnection).data; + const operationId = text(agent.lifecycleOperation?.id); + if (!router || !operationId) return null; + const [pin] = await client.select().from(aiConnectionTaskPins).where(and( + eq(aiConnectionTaskPins.companyId, agent.companyId), eq(aiConnectionTaskPins.agentId, agent.id), + eq(aiConnectionTaskPins.poolId, router.connectionId), eq(aiConnectionTaskPins.taskKey, `lifecycle:${operationId}`), + )); + const binding = aiConnectionBindingSchema.safeParse(pin?.selection.binding).data; + if (!pin || !binding || pin.selection.poolId !== router.connectionId || !isDeepStrictEqual(binding, pin.member.binding)) return null; + return { agent, binding, adapterConfig: { ...agent.adapterConfig, ...pin.selection.runtimeConfig }, requiresAdoption: false }; + } + + async function managedAgent(companyId: string, agentId: string, serviceSlug: string, options: { fallback?: AiConnectionBinding; sourceRunId?: string | null; agentSetup?: true; agentSetupOperationId?: string } = {}, client = db) { const [agent] = await client.select().from(agents).where(and(eq(agents.companyId, companyId), eq(agents.id, agentId))); if (!agent) return null; + if (options.agentSetup) { + if (options.agentSetupOperationId && options.agentSetupOperationId !== agent.lifecycleOperation?.id) return null; + const [source] = options.sourceRunId ? await client.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, options.sourceRunId), eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), + )) : []; + if (!source || source.status !== "queued" || source.startedAt || source.runtimeMode !== "legacy" || source.activeIdentityContextId) return null; + // Routed setup cards must name the lifecycle operation that selected the + // account. Legacy direct-account cards remain readable across upgrades. + if (!options.agentSetupOperationId && aiConnectionRouterBindingSchema.safeParse(agent.runtimeConfig.aiConnection).success) return null; + const managed = await queuedSetupManagedAgent(agent, client); + return managed?.binding.provider === serviceSlug ? managed : null; + } const router = aiConnectionRouterBindingSchema.safeParse(agent.runtimeConfig.aiConnection).data; if (router) { // Repair the durable task allocation, without adopting a personal default @@ -310,9 +347,11 @@ export function connectionIntentService(db: Db) { serviceSlug: string; purpose?: "ai" | "channel"; sourceRunId?: string | null; + agentSetup?: true; + agentSetupOperationId?: string; inventory?: Awaited>; }) { - const managed = input.purpose === "ai" ? await managedAgent(input.companyId, input.agentId, input.serviceSlug, { sourceRunId: input.sourceRunId }) : null; + const managed = input.purpose === "ai" ? await managedAgent(input.companyId, input.agentId, input.serviceSlug, { sourceRunId: input.sourceRunId, agentSetup: input.agentSetup, agentSetupOperationId: input.agentSetupOperationId }) : null; if (managed) { try { const selected = await aiConnectionService(db).select({ companyId: input.companyId, agentId: input.agentId, userId: input.responsibleUserId, adapterType: managed.agent.adapterType, model: managed.adapterConfig.model, runnerProvider: managed.adapterConfig.provider, acpxAgent: managed.adapterConfig.acpxAgent, binding: managed.binding }); @@ -796,7 +835,7 @@ export function connectionIntentService(db: Db) { async function requestWithContext( context: Awaited>, serviceSlug: string, - options: { purpose?: "ai" | "channel"; selectionInteractionId?: string; targetService?: string; connectionId?: string; toolNames?: string[] } = {}, + options: { agentSetup?: true; agentSetupOperationId?: string; purpose?: "ai" | "channel"; selectionInteractionId?: string; targetService?: string; connectionId?: string; toolNames?: string[] } = {}, ): Promise { const claims = { sub: context.agent.id, company_id: context.run.companyId, run_id: context.run.id, responsible_user_id: context.run.responsibleUserId! }; const route = parseAggregatorRoute(serviceSlug); @@ -885,6 +924,7 @@ export function connectionIntentService(db: Db) { serviceSlug: app.slug, purpose: options.purpose, sourceRunId: context.run.id, + ...(options.agentSetup ? { agentSetup: true as const, agentSetupOperationId: options.agentSetupOperationId } : {}), }); if (ready && !accessRequest && (!options.connectionId || ready.id === options.connectionId)) { return { @@ -913,6 +953,7 @@ export function connectionIntentService(db: Db) { { payload: { version: 1, + ...(options.agentSetup ? { agentSetup: true as const, agentSetupOperationId: options.agentSetupOperationId } : {}), ...(accessRequest ? { accessRequest } : {}), serviceSlug: app.slug, ...(options.purpose ? { purpose: options.purpose } : {}), @@ -927,7 +968,7 @@ export function connectionIntentService(db: Db) { sourceRunId: context.run.id, sourceIdentityContextId: context.run.activeIdentityContextId, addresseeUserId: context.run.responsibleUserId!, - idempotencyKey: `connection-intent:${context.run.id}:${context.run.responsibleUserId}:${app.slug}${upstreamService ? `:${upstreamService.slug}` : ""}${options.purpose ? `:${options.purpose}` : ""}${accessRequest ? `:access:${createHash("sha256").update(JSON.stringify(accessRequest)).digest("hex")}` : ""}`, + idempotencyKey: `connection-intent:${context.run.id}:${context.run.responsibleUserId}:${app.slug}${upstreamService ? `:${upstreamService.slug}` : ""}${options.purpose ? `:${options.purpose}` : ""}${options.agentSetup ? `:setup:${options.agentSetupOperationId ?? "legacy"}` : ""}${accessRequest ? `:access:${createHash("sha256").update(JSON.stringify(accessRequest)).digest("hex")}` : ""}`, }, ); if (interaction.status !== "pending") throw conflict("This connection request has already been resolved. Follow its recorded outcome."); @@ -1025,7 +1066,7 @@ export function connectionIntentService(db: Db) { readyConnectionId: inboxes.find(inbox => inbox.id === interactionId)?.connectionId ?? null }, }; } - let managed = payload.purpose === "ai" ? await managedAgent(loaded.issue.companyId, payload.requestingAgentId, app.slug, { sourceRunId: loaded.interaction.sourceRunId }) : null; + let managed = payload.purpose === "ai" ? await managedAgent(loaded.issue.companyId, payload.requestingAgentId, app.slug, { sourceRunId: loaded.interaction.sourceRunId, agentSetup: payload.agentSetup, agentSetupOperationId: payload.agentSetupOperationId }) : null; if (payload.purpose === "ai" && !managed) { const [source] = await db.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.id, loaded.interaction.sourceRunId!), eq(heartbeatRuns.companyId, loaded.issue.companyId), @@ -1040,7 +1081,7 @@ export function connectionIntentService(db: Db) { const inventory = await connectionInventory(loaded.issue.companyId); let usableAiConnection = managed ? await usableConnectionForAgent({ companyId: loaded.issue.companyId, agentId: payload.requestingAgentId, - responsibleUserId: loaded.interaction.addresseeUserId!, serviceSlug: app.slug, purpose: "ai", sourceRunId: loaded.interaction.sourceRunId, + responsibleUserId: loaded.interaction.addresseeUserId!, serviceSlug: app.slug, purpose: "ai", sourceRunId: loaded.interaction.sourceRunId, agentSetup: payload.agentSetup, agentSetupOperationId: payload.agentSetupOperationId, }) : null; if (managed?.requiresAdoption) { try { @@ -1227,7 +1268,7 @@ export function connectionIntentService(db: Db) { }); managed = { agent: updated, binding, adapterConfig: updated.adapterConfig, requiresAdoption: false }; } else { - managed = await managedAgent(loaded.issue.companyId, payload.requestingAgentId, payload.serviceSlug, { sourceRunId: loaded.interaction.sourceRunId }, txDb); + managed = await managedAgent(loaded.issue.companyId, payload.requestingAgentId, payload.serviceSlug, { sourceRunId: loaded.interaction.sourceRunId, agentSetup: payload.agentSetup, agentSetupOperationId: payload.agentSetupOperationId }, txDb); } if (!managed) throw conflict("Configure the agent’s AI connection before using this account"); const service = aiConnectionService(txDb); @@ -1357,8 +1398,71 @@ export function connectionIntentService(db: Db) { ); } + // Controller entry point for accepted work that cannot yet reach run preparation. + // Runtime tokens still require a running run. No credential is read or granted. + async function requestForQueuedAgentSetup(companyId: string, agentId: string) { + const [agent] = await db.select().from(agents).where(and(eq(agents.companyId, companyId), eq(agents.id, agentId))); + if (!agent || !isAgentAwaitingSetup(agent)) return; + const [company] = await db.select({ defaultResponsibleUserId: companies.defaultResponsibleUserId }) + .from(companies).where(eq(companies.id, companyId)); + // Match host verification's account owner. The queued task can have a + // different responsible user; its execution identity must stay unchanged. + const verificationUserId = agent.lifecycleOperation?.responsibleUserId ?? company?.defaultResponsibleUserId; + if (!verificationUserId) return; + await assertCurrentUserWriteAccess(companyId, verificationUserId); + const managed = await queuedSetupManagedAgent(agent); + if (!managed) return; + const { binding, adapterConfig } = managed; + const operationId = text(agent.lifecycleOperation?.id); + const candidates = await db.select({ run: heartbeatRuns, issue: issues }).from(heartbeatRuns).innerJoin(issues, and( + eq(issues.companyId, heartbeatRuns.companyId), + sql`${issues.id}::text = coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`, + )).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), + eq(heartbeatRuns.status, "queued"), isNull(heartbeatRuns.startedAt), eq(heartbeatRuns.runtimeMode, "legacy"), + isNull(heartbeatRuns.activeIdentityContextId), eq(issues.assigneeAgentId, agentId), + notInArray(issues.status, ["done", "cancelled"]), + )).orderBy(desc(heartbeatRuns.createdAt)).limit(100); + const seen = new Set(); + for (const { run, issue } of candidates) { + const issueId = text(run.contextSnapshot?.issueId) ?? text(run.contextSnapshot?.taskId); + if (!issueId || !run.responsibleUserId || seen.has(issueId)) continue; + seen.add(issueId); + const [pending] = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId), + eq(issueThreadInteractions.createdByAgentId, agentId), eq(issueThreadInteractions.addresseeUserId, verificationUserId), + eq(issueThreadInteractions.kind, "connection_intent"), eq(issueThreadInteractions.status, "pending"), + eq(issueThreadInteractions.sourceRunId, run.id), + operationId ? sql`${issueThreadInteractions.payload}->>'agentSetupOperationId' = ${operationId}` : undefined, + sql`${issueThreadInteractions.payload}->>'agentSetup' = 'true'`, + sql`${issueThreadInteractions.payload}->>'serviceSlug' = ${binding.provider}`, + )).limit(1); + if (pending) continue; + if (!await canActorReadIssuePrivacy(db, { type: "board", userId: verificationUserId, ignoreInstanceAdmin: true }, issue)) continue; + try { + const context = await loadRunContext({ sub: agentId, company_id: companyId, run_id: run.id, + responsible_user_id: run.responsibleUserId }, "queued_setup"); + try { + await aiConnectionService(db).select({ companyId, agentId, userId: verificationUserId, + adapterType: agent.adapterType, runnerProvider: adapterConfig.provider, + acpxAgent: adapterConfig.acpxAgent, model: adapterConfig.model, binding }); + } catch (error) { + if (!readAiConnectionConfigurationFailure(error)) throw error; + // This context selects the card recipient only. Do not rewrite the + // queued run, its task principal, or its saved execution input. + const cardContext = { ...context, run: { ...context.run, responsibleUserId: verificationUserId } }; + await requestWithContext(cardContext, binding.provider, { purpose: "ai", agentSetup: true, agentSetupOperationId: operationId ?? undefined }); + } + } catch (error) { + // A task may stop or change owners during this bounded active-work pass. + if (!(error instanceof HttpError) || ![403, 404, 409].includes(error.status)) throw error; + } + } + } + return { - validate: loadRunContext, + validate: (claims: ConnectionRunClaims) => loadRunContext(claims), + requestForQueuedAgentSetup, usableConnectionForAgent, search, request, @@ -1367,7 +1471,7 @@ export function connectionIntentService(db: Db) { requestForRunAuthFailure: async (runId: string) => { const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); if (!run || run.status !== "failed" || !isAiAuthenticationRepairable(run) || !run.responsibleUserId) return null; - const context = await loadRunContext({ sub: run.agentId, company_id: run.companyId, run_id: run.id, responsible_user_id: run.responsibleUserId }, true); + const context = await loadRunContext({ sub: run.agentId, company_id: run.companyId, run_id: run.id, responsible_user_id: run.responsibleUserId }, "failed_auth"); const [latest] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, run.companyId), sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId', ${heartbeatRuns.nativeIssueId}::text) = ${context.issue.id}`, diff --git a/server/src/services/execution-projection-recovery.test.ts b/server/src/services/execution-projection-recovery.test.ts index 603d59bdc0c..d03c165b996 100644 --- a/server/src/services/execution-projection-recovery.test.ts +++ b/server/src/services/execution-projection-recovery.test.ts @@ -15,6 +15,37 @@ describe.skipIf(!support.supported)("persisted stopped-workspace projection", () }, 60_000); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + it("names failed setup for an unstarted queued run and removes the label after readiness", async () => { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Setup wait", issuePrefix: "SW" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Reviewer", adapterType: "paperclip_runner", + status: "paused", lifecycleState: "verifying", lifecycleHolds: [], + lifecycleOperation: { id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 1, retryRequired: true }, + lifecycleError: "Execution configuration check failed. Connect an account and choose your personal default." }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "queued" }); + expect((await executionProjectionsForRuns(db, companyId, [runId])).get(runId)).toMatchObject({ + phase: "queued", label: "Agent setup needs attention", cause: "agent_setup_failed", canStop: true, + nextAction: "Open the agent's Harness / Runtime settings to repair its configuration, then retry verification.", + }); + await db.update(agents).set({ + lifecycleError: "Execution configuration check failed. The remote Runner is being updated.", + lifecycleOperation: { id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 1, + retryAt: new Date(Date.now() + 60_000).toISOString() }, + }).where(eq(agents.id, agentId)); + expect((await executionProjectionsForRuns(db, companyId, [runId])).get(runId)).toMatchObject({ + phase: "queued", label: "Waiting for agent setup", cause: "agent_setup_pending", + nextAction: "This turn starts automatically after the agent finishes setup and verification.", + }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]).toMatchObject({ status: "queued", startedAt: null }); + expect((await executionProjectionsForRuns(db, randomUUID(), [runId])).size).toBe(0); + await db.update(agents).set({ lifecycleState: "ready", status: "idle", lifecycleError: null }).where(eq(agents.id, agentId)); + expect((await executionProjectionsForRuns(db, companyId, [runId])).get(runId)).toMatchObject({ label: "Queued", cause: null }); + // Historical or started work must not inherit a later setup failure. + await db.update(agents).set({ lifecycleState: "verifying", status: "paused", lifecycleError: "safe failure" }).where(eq(agents.id, agentId)); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + expect((await executionProjectionsForRuns(db, companyId, [runId])).get(runId)?.label).toBe("Completed"); + }); + it("reads recovered files on a cancelled run through the real batched query", async () => { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Stopped workspace", issuePrefix: "SWP" }); diff --git a/server/src/services/execution-projection.test.ts b/server/src/services/execution-projection.test.ts index e8b4a049e7d..c9265157946 100644 --- a/server/src/services/execution-projection.test.ts +++ b/server/src/services/execution-projection.test.ts @@ -67,6 +67,18 @@ describe("execution truth projection", () => { expect(project(run({ status }), undefined).canStop).toBe(false); } }); + it("shows setup progress without overriding real holds or started work", () => { + const queued = run({ status: "queued", startedAt: null }); + const setup = { lifecycleState: "verifying", lifecycleHolds: [], lifecycleError: null }; + const projectSetup = (r = queued, agent = setup) => projectExecution(r, undefined, [], undefined, now, undefined, false, agent); + expect(projectSetup()).toMatchObject({ label: "Waiting for agent setup", cause: "agent_setup_pending" }); + expect(projectSetup(queued, { ...setup, lifecycleState: "paused" })).toMatchObject({ label: "Queued", cause: null }); + expect(projectExecution(queued, undefined, [], undefined, now, undefined, false, + { ...setup, lifecycleHolds: ["budget"] })).toMatchObject({ label: "Queued", cause: null }); + expect(projectSetup(run({ status: "queued", startedAt: now }))).toMatchObject({ label: "Queued", cause: null }); + expect(projectSetup(run({ status: "running" })).label).toBe("Confirming execution"); + }); + it("shows maintenance only for admission that has not started", () => { startTaskDrain(); try { diff --git a/server/src/services/execution-projection.ts b/server/src/services/execution-projection.ts index 95a0db338ef..cd51b3c73b1 100644 --- a/server/src/services/execution-projection.ts +++ b/server/src/services/execution-projection.ts @@ -1,5 +1,6 @@ import { and, desc, eq, inArray, isNull, ne, sql } from "drizzle-orm"; import { + agents, heartbeatRuns, issues, dotAgentBindings, @@ -19,6 +20,7 @@ import { workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; import { readTaskDrain } from "./task-admission.js"; import { getStartupRecoveryState } from "../startup-recovery-state.js"; import { hasPendingRunCleanup } from "./execution-blocker.js"; +import { isAgentAwaitingSetup } from "../modules/agent-lifecycle/index.js"; const MAX_PREDECESSOR_CLEANUP_CHECKS = 16; const text = (v: unknown) => (typeof v === "string" ? v : null); const executionRunColumns = { @@ -189,6 +191,13 @@ export async function executionProjectionsForRuns( // executor the claim also waits for. Pending uses admission's own predicate. const settlingPredecessorKeys = new Set(); const queuedRuns = runs.filter(run => run.status === "queued"); + const setupAgentIds = [...new Set(queuedRuns.filter(run => !run.startedAt).map(run => run.agentId))]; + const setupAgents = setupAgentIds.length ? await read(async () => db + .select({ id: agents.id, lifecycleState: agents.lifecycleState, + lifecycleHolds: agents.lifecycleHolds, + retryRequired: sql`coalesce(${agents.lifecycleOperation}->>'retryRequired', 'false') = 'true'` }) + .from(agents).where(and(eq(agents.companyId, companyId), inArray(agents.id, setupAgentIds)))) : []; + const setupByAgent = new Map(setupAgents.map(agent => [agent.id, agent])); const issueIdOf = (run: Run) => run.nativeIssueId ?? text(run.contextSnapshot?.issueId); const queuedIssueIds = [...new Set(queuedRuns.map(issueIdOf).filter((id): id is string => !!id))]; if (queuedIssueIds.length) { @@ -250,6 +259,7 @@ export async function executionProjectionsForRuns( now, waitingByRun.get(run.id), settlingPredecessorKeys.has(run.id), + setupByAgent.get(run.agentId), ), ); } @@ -275,6 +285,7 @@ export function projectExecution( now = new Date(), waitingForProvider?: WaitingExternalProvider, waitingForPredecessor = false, + setupAgent?: { lifecycleState: string; lifecycleHolds: readonly string[]; retryRequired?: boolean }, ): ExecutionProjection { const detail = coordinator?.failureDetail ?? {}; const successorRunId = text(detail.successorRunId); @@ -476,6 +487,15 @@ export function projectExecution( ? set("working", "Working") : set("reconnecting", "Confirming execution"); } + if (run.status === "queued" && !run.startedAt && setupAgent + && isAgentAwaitingSetup(setupAgent)) { + const needsRepair = setupAgent.retryRequired === true; + projection.cause = needsRepair ? "agent_setup_failed" : "agent_setup_pending"; + projection.nextAction = needsRepair + ? "Open the agent's Harness / Runtime settings to repair its configuration, then retry verification." + : "This turn starts automatically after the agent finishes setup and verification."; + return set("queued", needsRepair ? "Agent setup needs attention" : "Waiting for agent setup"); + } if (run.status === "queued" && !run.startedAt && readTaskDrain(now)) { projection.cause = "task_drain"; projection.nextAction = "This turn will start automatically when maintenance finishes."; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 4881298f030..8f9f8395991 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { connectionIntentService } from "../connection-intents.js"; import { verifyBlockerStallWake } from "../recovery/blocker-stall-wakeups.js"; import { CONTEXT_COMPACTION_WAKE_REASON, ContextCompactionError, isContextCompactionRun, readContextCompactionRequest } from "../context-compaction.js"; import { isProcessRunAdmissionClosed } from "./run-admission.js"; @@ -2058,7 +2059,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const agentIds = [...new Set(queuedRuns.map((r) => r.agentId))]; for (const agentId of agentIds) { - await startNextQueuedRunForAgent(agentId); + await startNextQueuedRunForAgent(agentId, true); } } @@ -2090,7 +2091,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ), ); for (const { agentId } of rows) { - await startNextQueuedRunForAgent(agentId); + await startNextQueuedRunForAgent(agentId, true); } return { agents: rows.length }; } @@ -2134,7 +2135,13 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } } - async function startNextQueuedRunForAgent(agentId: string) { + async function attachQueuedSetupRequest(agent: typeof agents.$inferSelect) { + if (isAgentAwaitingSetup(agent)) await connectionIntentService(db) + .requestForQueuedAgentSetup(agent.companyId, agent.id) + .catch(() => logger.warn({ agentId: agent.id }, "Could not attach queued agent setup connection request")); + } + + async function startNextQueuedRunForAgent(agentId: string, attachSetupRequest = false) { if (isRunAdmissionClosed() || (await getSchedulingSuppression()).suppressed) return []; const cutoff = await getWorktreeExecutionCutoff(); // Cancelled after the start lock is released: cancelRunInternal promotes the @@ -2149,6 +2156,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (!agent) return []; const invokability = await getAgentInvokability(agent); if (!invokability.invokable) { + if (attachSetupRequest) await attachQueuedSetupRequest(agent); if (shouldCancelRunsForNonInvokableAgent(invokability)) { cancellationReason = `Cancelled because the agent is not invokable: ${invokability.reason}`; } @@ -4750,12 +4758,15 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende return null; } if (outcome.kind === "coalesced") { + await attachQueuedSetupRequest(agent); await startNextQueuedRunForAgent(agent.id); return outcome.run; } if (outcome.kind === "replayed") { - if (outcome.run.status === "queued") + if (outcome.run.status === "queued") { + await attachQueuedSetupRequest(agent); await startNextQueuedRunForAgent(agent.id); + } return outcome.run; } @@ -4772,6 +4783,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }, }); + await attachQueuedSetupRequest(agent); await startNextQueuedRunForAgent(agent.id); return newRun; } @@ -4983,6 +4995,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }, }); + await attachQueuedSetupRequest(agent); await startNextQueuedRunForAgent(agent.id); return newRun; diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 9599b760f4c..c75d546bb95 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -2908,7 +2908,9 @@ export function issueThreadInteractionService( ) => { const payload = connectionIntentPayloadSchema.parse(input.payload); const sameSavedSetup = (candidate: typeof payload) => - candidate.savedSetup?.endpointId === payload.savedSetup?.endpointId + candidate.agentSetup === payload.agentSetup + && candidate.agentSetupOperationId === payload.agentSetupOperationId + && candidate.savedSetup?.endpointId === payload.savedSetup?.endpointId && candidate.savedSetup?.connectionId === payload.savedSetup?.connectionId && candidate.savedSetup?.action === payload.savedSetup?.action && Boolean(candidate.savedSetup?.credentialRotation) === Boolean(payload.savedSetup?.credentialRotation) @@ -2962,7 +2964,8 @@ export function issueThreadInteractionService( eq(issueThreadInteractionIdempotencyKeys.idempotencyKey, input.idempotencyKey))).limit(1); if (qualityAlias) throw conflict("Interaction idempotency key already exists for a different request", { idempotencyKey: input.idempotencyKey }); - // Serialize on the task so retries and later runs share the same live card. + // Generic requests share live cards across runs. Setup cards stay bound + // to the queued turn that their delivery can repair. const pending = await tx.select().from(issueThreadInteractions).where(and( eq(issueThreadInteractions.companyId, issue.companyId), eq(issueThreadInteractions.issueId, issue.id), @@ -2972,14 +2975,21 @@ export function issueThreadInteractionService( eq(issueThreadInteractions.addresseeUserId, input.addresseeUserId), )); const reusable = pending.find((candidate) => - connectionIntentPayloadSchema.parse(candidate.payload).serviceSlug === payload.serviceSlug && connectionIntentPayloadSchema.parse(candidate.payload).purpose === payload.purpose && sameSavedSetup(connectionIntentPayloadSchema.parse(candidate.payload)) && isDeepStrictEqual(connectionIntentPayloadSchema.parse(candidate.payload).accessRequest, payload.accessRequest)); + (!payload.agentSetup || candidate.sourceRunId === input.sourceRunId) + && connectionIntentPayloadSchema.parse(candidate.payload).serviceSlug === payload.serviceSlug && connectionIntentPayloadSchema.parse(candidate.payload).purpose === payload.purpose && sameSavedSetup(connectionIntentPayloadSchema.parse(candidate.payload)) && isDeepStrictEqual(connectionIntentPayloadSchema.parse(candidate.payload).accessRequest, payload.accessRequest)); if (reusable) { await enqueueIssueInteractionChatPublications(tx as unknown as Db, hydrateInteraction(reusable)); return reusable; } - const [sourceRun] = await tx.select({ context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns) + const sourceQuery = tx.select({ context: heartbeatRuns.contextSnapshot, status: heartbeatRuns.status, + agentId: heartbeatRuns.agentId, startedAt: heartbeatRuns.startedAt, runtimeMode: heartbeatRuns.runtimeMode }).from(heartbeatRuns) .where(and(eq(heartbeatRuns.id, input.sourceRunId), eq(heartbeatRuns.companyId, issue.companyId))); + const [sourceRun] = await (payload.agentSetup ? sourceQuery.for("no key update") : sourceQuery); + if (payload.agentSetup && (payload.purpose !== "ai" || sourceRun?.agentId !== payload.requestingAgentId + || sourceRun.status !== "queued" || sourceRun.startedAt !== null || sourceRun.runtimeMode !== "legacy")) { + throw conflict("Agent setup requests require an unstarted queued turn"); + } if (!sourceRun) throw unprocessable("Interaction source run is unavailable"); const originCommentIds = await currentContinuationOrigins(tx as unknown as Db, issue.companyId, issue.id, sourceRun.context); await notifyIssueReviewState(tx, { companyId: issue.companyId, issueId: issue.id }); diff --git a/ui/src/components/AgentLifecycleStatus.test.tsx b/ui/src/components/AgentLifecycleStatus.test.tsx index 0162f4c62ff..498ec7f354e 100644 --- a/ui/src/components/AgentLifecycleStatus.test.tsx +++ b/ui/src/components/AgentLifecycleStatus.test.tsx @@ -26,8 +26,8 @@ it.each([undefined, "ready", "paused", "pending_approval", "terminated", "reject expect(agentLifecycleRefetchInterval(lifecycleState)).toBe(30_000); }); -it("does not promise automatic retry of a failed configuration check", () => { - const html = renderToStaticMarkup( {}} retryPending={false} />); +it.each(["verifying", "resuming"] as const)("does not promise automatic retry of a failed configuration check while %s", lifecycleState => { + const html = renderToStaticMarkup( {}} retryPending={false} />); expect(html).toContain("Retry now"); expect(html).not.toContain("Paperclip retries automatically"); }); diff --git a/ui/src/components/AgentLifecycleStatus.tsx b/ui/src/components/AgentLifecycleStatus.tsx index 622281a8ca7..51c87c63e4e 100644 --- a/ui/src/components/AgentLifecycleStatus.tsx +++ b/ui/src/components/AgentLifecycleStatus.tsx @@ -64,7 +64,7 @@ export function AgentLifecycleStatus({ agent, onRetry, retryPending, refreshErro {retryPending ? "Retrying…" : "Retry now"} } - {phase && agent.lifecycleError &&

{agent.lifecycleState === "verifying" + {phase && agent.lifecycleError &&

{agent.lifecycleState === "verifying" || agent.lifecycleState === "resuming" ? "Retry now checks the saved configuration again immediately." : "Paperclip retries automatically; Retry now checks again immediately."}

} ; diff --git a/ui/src/components/IssueChatThread.test.tsx b/ui/src/components/IssueChatThread.test.tsx index 9ffc0f4ba4c..fccea4c9000 100644 --- a/ui/src/components/IssueChatThread.test.tsx +++ b/ui/src/components/IssueChatThread.test.tsx @@ -4473,6 +4473,15 @@ describe("IssueAssigneePausedNotice", () => { act(() => root.unmount()); }); + it.each(["preparing", "verifying", "resuming"])("leaves %s recovery to the task setup status and inline connection card", lifecycleState => { + const root = createRoot(container); + const agent = { ...pausedAgent(""), pauseReason: null, lifecycleState, lifecycleError: "Connect an account" } as Agent; + act(() => root.render()); + expect(container.querySelector('[data-testid="issue-assignee-paused-resume"]')).toBeNull(); + expect(container.textContent).not.toContain("is paused"); + act(() => root.unmount()); + }); + it.each([ ["preparing", "computer is being prepared"], ["verifying", "setup isn't finished"], diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index 21a5bfef6cd..138d2000e9e 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -827,6 +827,10 @@ export function IssueAssigneePausedNotice({ }) { if (!agent || agent.status !== "paused" || issueStatusIsTerminalDisposition(issueStatus)) return null; + // Lifecycle setup is shown by the queued turn and its inline connection card. + // Resume changes operator holds; it cannot complete verification. + if (!agent.pauseReason && ["preparing", "verifying", "resuming"].includes(agent.lifecycleState ?? "")) return null; + // A freshly created agent is stored as paused with pauseReason "setup" until // its setup finishes. That is not an operator pause, so say what is actually // happening and point at the real recovery path instead of "Resume agent".