diff --git a/doc/grok-bot-agent.md b/doc/grok-bot-agent.md index ffd37093d98..0da45937194 100644 --- a/doc/grok-bot-agent.md +++ b/doc/grok-bot-agent.md @@ -76,6 +76,13 @@ outcome needs inspection and explicit recovery; it is never sent again automatic Saved work remains on the task if execution fails. Reverify after clearing an inspected recovery hold before admitting another assignment. +New assignments remain in the durable queue while the Grok connection is +unverified, revoked, or held for recovery. Readiness is checked before claiming +a new run or allocating its Runner. Already-admitted session reconciliation +keeps its recorded authority; it does not gain permission to start new work. +A configured AI connection pool selects its own provider. Its native tasks use +the pool's normal admission checks and do not require an unused Grok connection. + **Stop Paperclip work** immediately revokes the session's Paperclip authority and stops future dispatch. Remote Grok execution may continue. Open the Grok Bot app to inspect it. Paperclip shows reported activity, not a complete remote transcript diff --git a/doc/plans/2026-10-13-grok-bot-production-readiness.md b/doc/plans/2026-10-13-grok-bot-production-readiness.md index 438cfca7607..dec52786c75 100644 --- a/doc/plans/2026-10-13-grok-bot-production-readiness.md +++ b/doc/plans/2026-10-13-grok-bot-production-readiness.md @@ -1,6 +1,6 @@ # Grok Bot production readiness -Status as of October 15: real assigned and Bot-initiated tasks reach canonical completion. The active-session wake correction is merged. The proactive admission echo correction is merged and source-verified; image publication and a real no-echo repeat remain pending. A real controller restart and active Stop diagnostic passed their authority checks; Stop also exposed a false workspace-repair hold, now under correction. Production acceptance is incomplete. The current release goal and matrix below supersede the earlier dated campaign entries. Historical successes and failures remain evidence for the revisions actually tested. +Status as of October 15: a real Bot-initiated repeat passed with one task, one run, seven completed operations, canonical Done, no assignment webhook and successful cleanup on packaged Core ea4aa185. The false file-recovery correction is merged and deployed as Core 5a9e6b63; historical inspection archived the incorrect obligation and fresh verification passed. Further assigned work is blocked by a provider-reported routine-start rate limit. Stop then exposed a separate queue-admission defect: pending work repeatedly allocated a Runner before discovering the connection hold. The qualification agent is paused and all affected leases are cleaned up. Production acceptance remains incomplete. ## Current production finish line — October 14 @@ -12,8 +12,8 @@ There is no testing call, case-count, token or spending cap. The former six-case | --- | --- | --- | | Ordinary onboarding works | Actual Grok UI, fixed agent consent, saved setup, wrong-key behavior, authenticated challenge, refresh, reconnect and revocation | Resumed real Grok authorization, saved setup Done, authenticated routine challenge and automatic refresh passed on the October 15 serving revision. Runtime reconnect copy fix is merged; verify its published UI and fresh production setup next. | | Assigned work reaches a usable result | One actual managed workflow: progress, question and answer, document, downloaded file/hash, canonical completion or review and cleanup | A preliminary comparison task on serving Core ff0 completed progress → question → answer → report/CSV → canonical Done, with both cleanup receipts. The cited CSV downloads with the exact publication hash. Two distinct CSV uploads accompanied an active-session result wake; correct that wake behavior and requalify the final candidate. | -| Bot initiates governed work | Real-client task creation/claim/completion with stable receipts and no outbound echo or duplicate effects | A real idle-conversation request created one task and one managed run, saved a 100-word report, reached accepted Done and cleaned up its lease. A delayed-admission wake and two acceptance attempts were also recorded. Functional completion passed; no-echo acceptance failed. Requalify the correction. | -| Recovery and authority are reliable | Healthy work past the prior five-minute failure window; actual controller restart preserving assignment/session/process and no duplicate wake; Stop, pause, reassignment, revocation, stale calls, queues and cleanup | A real managed diagnostic ran for 13 minutes 44 seconds, survived an actual same-image controller restart with the original assignment/session/process/lease, completed canonically and cleaned up without another wake. Active Stop fenced its work, rejected a real stale progress update, kept its checkpoint unchanged and cleaned up. Cancellation also created a false file-repair hold despite an admitted workspace-access-none contract; correct and requalify that recovery behavior. These diagnostics ran on Core 11d, not the final release image. | +| Bot initiates governed work | Real-client task creation/claim/completion with stable receipts and no outbound echo or duplicate effects | A real idle-conversation repeat on packaged Core ea4aa185 created one task and one run, read a public source marker with Grok's own tools, saved its report, reached canonical Done and cleaned up. One acceptance and seven operations completed, with no assignment webhook. The earlier duplicate-wake failure remains historical evidence; its correction passed this repeat. | +| Recovery and authority are reliable | Healthy work past the prior five-minute failure window; actual controller restart preserving assignment/session/process and no duplicate wake; Stop, pause, reassignment, revocation, stale calls, queues and cleanup | The Core 11d diagnostic survived an actual restart and 13 minutes 44 seconds, then completed and cleaned up. Active Stop denied a real stale mutation. Packaged Core 5a9e6b63 later archived the historical false file hold and stopped an unaccepted session without another file obligation. Queued preparation then churned behind the connection recovery hold; correct that admission gate and repeat the final workflow, restart and active Stop. The original October 14 exit cause remains unproven. | | One identifiable reviewed candidate passes source gates | Focused protocol/auth/Runner/UI tests, full CI/typecheck/tests/build/token/Storybook evidence and exact artifacts | Active-session wake correction #16394 merged with 55 passing checks, two optional skips and review 5/5. The proactive echo correction started from master 7fc2dc8b3236e1149ae32756b6153a503623404b and was rebased onto freshly fetched master 250429942e27d568f20d3bc7dddcfc84a666480a before submission. Two regression cases reproduce the real delayed-admission wake before the fix. All 214 focused protocol/auth/question checks pass, including 80 Grok and Dot checks. Repository typecheck/build, token gates, the 2,731-story structure and Storybook build pass. Fresh CI, review, publication and the real no-echo repeat remain required. | | Production opt-in is delivered and smoke-tested | Staging acceptance, correct served Core/distribution/Runner, normal editable opt-in controls/defaults, actual production Bot workflow and inspectable task/output links | Cloud opt-in policy and release-guide corrections are merged. Staging Cloud provenance and minimal feature values were verified, with Dot disabled and defaults unchanged. The corrected wake candidate, remaining real-client cases, ordinary production opt-in controls and production smoke are outstanding. | @@ -41,6 +41,26 @@ The active Stop diagnostic cancelled its run, fenced the assignment, kept the ch Keep native providers and person-purpose MCP connections unchanged. No browser automation belongs in normal operation. Defer localhost relays, internal Grok APIs, marketplace distribution and Paperclip-connected third-party tools. Completion of a Paperclip work session does not assert remote Grok termination. +## October 15 — queued admission during Grok recovery + +A real historical inspection on packaged Core `5a9e6b63` removed the false file +obligation while preserving the old assignment fence and no-replay evidence. +The routine returned an authenticated verification after one explicit retry of a +confirmed rejection. A new task message then produced a new offered assignment. +Grok rejected its webhook and its explicit retry with HTTP 400. A separate +revoked-binding diagnostic identified `Automation run rate limit exceeded`. +Included usage was still available; no documented cooldown or routine-start +allowance was visible. No billing setting or provider limit was changed. + +Stopping that unaccepted session created no false file obligation. Pending work +then made four failed preparation attempts with no new external assignment or +webhook. Pause stopped the churn, and every managed lease reports successful +cleanup. The admission correction checks the existing broker's readiness before +claiming fresh queued work, preserves the input for later recovery, and leaves +already-started reconciliation under its recorded execution contract. Current +runtime-selection and dispatch authority checks stay in force. The patch still +requires review, packaged deployment and a real recovery repeat. + ## Candidate and release contract The production candidate started at `origin/master` commit `833769ab557255b11dd8f650f43ed7333653f5e4` and incorporates master through `e0f30a4c71` in the managed `codex/grokbot-production` worktree. The complete earlier prototype is preserved in snapshot `e0a8150ca248c1b80c58a7e26c10ccda9951907b`; its October 2 test results are historical, not evidence for this candidate. Reuse its provider-specific work where appropriate. Do not import its old gateway baseline, migration numbers, independent session coordinator, or retired UI entry point wholesale. diff --git a/server/src/__tests__/ai-connection-router.test.ts b/server/src/__tests__/ai-connection-router.test.ts index 71186aba2cf..209bcb33df4 100644 --- a/server/src/__tests__/ai-connection-router.test.ts +++ b/server/src/__tests__/ai-connection-router.test.ts @@ -1,11 +1,11 @@ import { enableRunFailureReporting } from "./helpers/run-failure-reporting.js"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { randomUUID } from "node:crypto"; -import { mkdtemp, rm } from "node:fs/promises"; -import os from "node:os"; +import { mkdir } from "node:fs/promises"; +import { createIsolatedTestHome } from "./helpers/isolated-test-home.js"; import path from "node:path"; import { and, eq } from "drizzle-orm"; -import { createDb, closeRegisteredClients, companies, agents, companyMemberships, connectionGrants, plugins, aiConnectionRouterCursors, aiConnectionTaskPins, toolConnections, companySecrets, heartbeatRuns, issues } from "@paperclipai/db"; +import { createDb, closeRegisteredClients, companies, agents, companyMemberships, connectionGrants, plugins, aiConnectionRouterCursors, aiConnectionTaskPins, toolConnections, companySecrets, heartbeatRuns, issues, dotAgentBindings, grokBotRoutines } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "@paperclipai/db/test-embedded-postgres"; import type { AiConnectionPoolMember, AiConnectionRouterRequest, AiConnectionRouterResult, PaperclipPluginManifestV1 } from "@paperclipai/shared"; import { aiConnectionRouterService, AiConnectionPoolExhausted, poolMemberRuntimeConfig, applyAiConnectionRouterTaskSettings } from "../services/ai-connection-router.js"; @@ -26,6 +26,7 @@ import { connectionIntentService } from "../services/connection-intents.js"; import { legacyExecutionNeedsReconciliation } from "../services/legacy-execution-recovery.js"; import { getServerAdapter, registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js"; import { waitForPendingRunFailureReports } from "../services/run-failure-report.js"; +import { dotRunnerBroker } from "../services/dot-runner-broker.js"; const captureRunFailure = vi.hoisted(() => vi.fn()); vi.mock("../sentry.js", async (importOriginal) => ({ @@ -37,6 +38,7 @@ let database: Awaited>; const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim(); let db: ReturnType; let home: string; +let isolatedHome: ReturnType | undefined; const companyId = randomUUID(), otherCompanyId = randomUUID(), agentId = randomUUID(), secondAgentId = randomUUID(), pluginId = randomUUID(); const pluginKey = "fixture.ai-router"; const manifest: PaperclipPluginManifestV1 = { id: pluginKey, apiVersion: 1, version: "0.1.0", displayName: "Fixture router", description: "Fixture", author: "Tests", categories: ["connector"], capabilities: ["ai.connections.route"], aiConnectionRouter: { name: "AI connection pool", description: "Use saved connections" }, entrypoints: { worker: "worker.js" } }; @@ -53,7 +55,8 @@ const service = () => aiConnectionRouterService(db, worker); const resolve = (poolId: string, taskKey: string, extra = {}) => service().resolve({ companyId, poolId, agentId, userId: "alice", adapterType: "paperclip_runner", taskKey, ...extra }); const makePool = async (config = {}) => service().save(pluginKey, { companyId, config: { name: randomUUID(), enabled: true, mode: "round_robin", thresholdPercent: 90, members, ...config } }, "alice"); beforeAll(async () => { - home = await mkdtemp(path.join(os.tmpdir(), "paperclip-router-tests-")); + isolatedHome = createIsolatedTestHome(); + home = isolatedHome.home; vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_INSTANCE_ID", "router-fixture"); if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl); else { database = await startEmbeddedPostgresTestDatabase("paperclip-router-db-"); db = createDb(database.connectionString); } @@ -73,7 +76,7 @@ afterAll(async () => { await waitForPendingRunFailureReports(); if (externalTestDatabaseUrl) await closeRegisteredClients(externalTestDatabaseUrl); await db?.$client.end(); - await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); + await database?.cleanup(); vi.unstubAllEnvs(); isolatedHome?.cleanup(); }); describe("durable, authorized connection routing", () => { @@ -435,6 +438,51 @@ describe("durable, authorized connection routing", () => { expect(resolveTaskSessionConfigFreshness({ hasTaskSession: true, configuredModel: String(selected.runtimeConfig.model), taskSessionParams: { __paperclipConfiguredModel: selected.runtimeConfig.model, __paperclipConfigFingerprint: previous.fingerprint, __paperclipConfigFingerprintVersion: previous.version, __paperclipConfigCategoryFingerprints: previous.categoryFingerprints }, configMetadata: current, compatibleConfigMetadata }).reset).toBe(false); } finally { await before.cleanup(); await after.cleanup(); } }); + it("admits a healthy pool member when the saved Grok connection is held", async () => { + const original = await instanceSettingsService(db).getExperimental(); + await instanceSettingsService(db).updateExperimental({ enableAiConnectionRouters: true, + enableNativeRunner: true, enableGrokBot: true, enablePublicMcp: true, enableWorktreeRunExecution: true }); + const pool = await makePool({ members: [members[0]!] }); + const bindingId = randomUUID(); + const workspace = path.join(home, "routed-grok-work"); + await mkdir(workspace); + const [agent] = await db.insert(agents).values({ companyId, name: "Routed Grok fixture", status: "idle", + adapterType: "paperclip_runner", adapterConfig: { provider: "grok_bot", externalBindingId: bindingId, cwd: workspace }, + runtimeConfig: { aiConnection: { mode: "router", connectionId: pool.id }, heartbeat: { enabled: false, wakeOnDemand: true } } }).returning(); + await db.insert(dotAgentBindings).values({ id: bindingId, companyId, agentId: agent.id, + operatorId: "alice", providerKind: "grok_bot", status: "connected" }); + const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, members[0]!.binding.grantId)); + const ref = grant!.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!; + await db.insert(grokBotRoutines).values({ companyId, agentId: agent.id, bindingId, bindingGeneration: 1, + botName: "Unused held Bot", secretId: ref.secretId, secretVersion: 1, + recoveryRequiredAt: new Date(), recoveryReason: "work_session" }); + expect(await dotRunnerBroker(db).canStartNewWork(companyId, agent.id, bindingId)).toBe(false); + const readiness = vi.spyOn(dotRunnerBroker(db), "canStartNewWork"); + const beforeNative = vi.fn(async () => { throw new Error("fixture stopped before provider start"); }); + const heartbeat = heartbeatService(db, { pluginWorkerManager: worker, runtimeEnv: {}, + beforeNativeRuntimeSelection: beforeNative, + nativeSessionBackendFactory: () => { throw new Error("Unexpected provider start"); } }); + const [issue] = await db.insert(issues).values({ companyId, title: "Healthy routed task", status: "todo", + assigneeAgentId: agent.id, responsibleUserId: "alice", createdByUserId: "alice" }).returning(); + try { + const run = await heartbeat.wakeup(agent.id, { source: "assignment", reason: "issue_assigned", triggerDetail: "system", + payload: { issueId: issue.id }, contextSnapshot: { issueId: issue.id, responsibleUserId: "alice" } }); + expect(run).not.toBeNull(); + await heartbeat.drainActiveRunExecutions(); + const saved = await heartbeat.getRun(run!.id); + expect(saved?.contextSnapshot?.aiRouterSelection).toMatchObject({ poolId: pool.id, + memberId: members[0]!.id, runtimeConfig: { provider: "codex" } }); + expect(saved?.error).toContain("fixture stopped before provider start"); + expect(beforeNative).toHaveBeenCalledExactlyOnceWith(run!.id); + expect(readiness).not.toHaveBeenCalled(); + } finally { + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agent.id)); + await heartbeat.drainActiveRunExecutions(); + readiness.mockRestore(); + await instanceSettingsService(db).updateExperimental(original); + } + }, 30_000); + it("keeps session identity and fingerprints stable on authenticated refresh, but changes them on manual replacement", async () => { const input = { companyId, agentId, responsibleUserId: "alice", adapterType: "paperclip_runner", binding: members[0]!.binding, config: { provider: "codex", model: "gpt-5.6-sol" } }; const first = await prepareManagedAiRuntime(db, input); diff --git a/server/src/__tests__/grok-bot-routine.test.ts b/server/src/__tests__/grok-bot-routine.test.ts index c49b846a6d8..1bc93d8ff41 100644 --- a/server/src/__tests__/grok-bot-routine.test.ts +++ b/server/src/__tests__/grok-bot-routine.test.ts @@ -23,6 +23,8 @@ import { type GrokRoutineWake } from "../services/grok-bot/webhook.js"; import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { admitExplicitNativeContinuation } from "../services/explicit-native-continuation.js"; import { hasRequiredWorkspaceRecovery } from "../services/workspace-restore-recovery-state.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { environmentRuntimeService } from "../services/environment-runtime.js"; let temporary: Awaited>; let db: ReturnType; @@ -102,6 +104,63 @@ async function acceptance(f: Awaited>, work: typeof d } describe("durable Grok routine qualification", () => { + it("preflights current connection authority without treating database failures as readiness", async () => { + const f = await fixture(); + const broker = dotRunnerBroker(db); + const canStart = () => broker.canStartNewWork(f.company.id, f.agent.id, f.pairing.bindingId); + expect(await canStart()).toBe(false); + const row = await verification(f); + await f.service.send(f.company.id, f.agent.id, row.id); + await f.service.acknowledge(f.principal, acknowledgement(row)); + expect(await canStart()).toBe(true); + expect(await broker.canStartNewWork(randomUUID(), f.agent.id, f.pairing.bindingId)).toBe(false); + expect(await broker.canStartNewWork(f.company.id, f.agent.id, randomUUID())).toBe(false); + await db.update(grokBotRoutines).set({ recoveryRequiredAt: new Date(), recoveryReason: "work_session" }) + .where(eq(grokBotRoutines.bindingId, f.pairing.bindingId)); + expect(await canStart()).toBe(false); + await db.update(grokBotRoutines).set({ recoveryRequiredAt: null, recoveryReason: null }) + .where(eq(grokBotRoutines.bindingId, f.pairing.bindingId)); + await db.update(mcpOauthGrants).set({ revokedAt: new Date() }).where(eq(mcpOauthGrants.id, f.principal.grant.id)); + expect(await canStart()).toBe(false); + const unavailable = new Error("synthetic database unavailable"); + vi.spyOn(broker, "snapshot").mockRejectedValueOnce(unavailable); + await expect(canStart()).rejects.toBe(unavailable); + }); + + it("holds public heartbeat input before claiming a Runner while Grok recovery is required", async () => { + const f = await fixture(); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, f.agent.id)); + const row = await verification(f); + await f.service.send(f.company.id, f.agent.id, row.id); + await f.service.acknowledge(f.principal, acknowledgement(row)); + await db.update(grokBotRoutines).set({ recoveryRequiredAt: new Date(), recoveryReason: "work_session" }) + .where(eq(grokBotRoutines.bindingId, f.pairing.bindingId)); + const [task] = await db.insert(issues).values({ companyId: f.company.id, title: "Wait for Grok inspection", + status: "todo", assigneeAgentId: f.agent.id, responsibleUserId: f.principal.grant.userId }).returning(); + const runtime = environmentRuntimeService(db); + const acquire = vi.spyOn(runtime, "acquireRunLease").mockRejectedValue(new Error("Unexpected provider allocation")); + const heartbeat = heartbeatService(db, { environmentRuntime: runtime }); + try { + await heartbeat.wakeup(f.agent.id, { source: "assignment", reason: "issue_assigned", + requestedByActorType: "user", requestedByActorId: f.principal.grant.userId, + payload: { issueId: task.id }, contextSnapshot: { issueId: task.id } }); + await heartbeat.drainActiveRunExecutions(); + for (let poll = 0; poll < 3; poll++) await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id)); + expect(runs).toEqual([expect.objectContaining({ status: "queued", startedAt: null })]); + expect(await db.select().from(environmentLeases).where(eq(environmentLeases.companyId, f.company.id))).toEqual([]); + expect(await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.agentId, f.agent.id))).toEqual([]); + expect(await db.select().from(grokBotDispatches).where(eq(grokBotDispatches.bindingId, f.pairing.bindingId))) + .toEqual([expect.objectContaining({ id: row.id, status: "acknowledged" })]); + expect(acquire).not.toHaveBeenCalled(); + expect((await db.select().from(issues).where(eq(issues.id, task.id)))[0].status).toBe("todo"); + } finally { + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agent.id)); + await heartbeat.drainActiveRunExecutions(); + } + }); + it("receives an authenticated Grok acknowledgement through the actual MCP transport without exposing personal or Dot event tools", async () => { const f = await fixture(); const row = await verification(f); await f.service.send(f.company.id, f.agent.id, row.id); diff --git a/server/src/services/dot-runner-broker.ts b/server/src/services/dot-runner-broker.ts index c602ee37a3e..cc2e594f15d 100644 --- a/server/src/services/dot-runner-broker.ts +++ b/server/src/services/dot-runner-broker.ts @@ -204,6 +204,15 @@ function createBroker(db: Db) { } }); }, + async canStartNewWork(companyId: string, agentId: string, bindingId: string): Promise { + try { + await this.snapshot(companyId, agentId, bindingId); + return true; + } catch (error) { + if (error instanceof McpOAuthError) return false; + throw error; + } + }, async snapshot(companyId: string, agentId: string, bindingId: string): Promise { const state = await this.bindingForAgent(companyId, agentId); if (!state || !await enabled(state.providerKind)) throw fail("External provider is disabled for new work."); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 2525be9a261..9cd1ff8635e 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { dotRunnerBroker } from "./dot-runner-broker.js"; import { withRunExecutionTiming, measureRunPhase, measureWorkspaceSyncPhase } from "./execution-preparation-timing.js"; import { createHeartbeatPreparationTiming } from "./heartbeat/preparation-timing.js"; import { createHeartbeatInstructionCopyRelease, createHeartbeatInstructionWorkingCopy } from "./heartbeat/instruction-working-copy.js"; @@ -1203,6 +1204,13 @@ export function heartbeatService( }); const heartbeatQueue = createHeartbeatQueue(db, { + canStartNewExternalBotWork: async agent => { + if (agent.adapterType !== "paperclip_runner" || agent.adapterConfig.provider !== "grok_bot") return true; + const aiConnection = aiRuntimeConnectionBindingSchema.safeParse(agent.runtimeConfig?.aiConnection); + if (aiConnection.success && aiConnection.data.mode === "router") return true; + return dotRunnerBroker(db).canStartNewWork(agent.companyId, agent.id, + readNonEmptyString(agent.adapterConfig.externalBindingId) ?? ""); + }, applyWakeQueuePostCommitEffects, setRunStatus, setWakeupStatus, diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 3572e68daf2..3e676e5af0a 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -48,6 +48,7 @@ function callbacks(db: Db) { runTaskKey: (run: Run) => typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null, filterZombieCoalesceTarget: (run: T | null) => run, getSchedulingSuppression: vi.fn(async () => ({ suppressed: false, reason: null } as Awaited>)), + canStartNewExternalBotWork: vi.fn(async () => true), executeRun: vi.fn(async () => undefined), applyWakeQueuePostCommitEffects: vi.fn(async () => undefined), releaseIssueExecutionAndPromote: vi.fn(async () => undefined), @@ -199,6 +200,56 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].executionRunId).toBe(waiter.id); }); + it.each(["direct", "batch"] as const)("keeps fresh work queued across a held external connection on the %s path", async path => { + const { run, agent, issue, wake } = await fixture("queued"); + await db.update(heartbeatRuns).set({ responsibleUserId: "queue-test-user" }).where(eq(heartbeatRuns.id, run.id)); + const deps = { ...callbacks(db), canStartNewExternalBotWork: vi.fn(async () => false) }; + deps.getAgent.mockResolvedValue({ ...agent, adapterType: "paperclip_runner", adapterConfig: { provider: "grok_bot" } }); + const queue = createHeartbeatQueue(db, deps); + for (let attempt = 0; attempt < 3; attempt++) { + if (path === "direct") expect(await queue.claimQueuedRun({ ...run, responsibleUserId: "queue-test-user" })).toBeNull(); + else expect(await queue.startNextQueuedRunForAgent(agent.id)).toEqual([]); + } + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toMatchObject({ status: "queued", startedAt: null }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0].status).toBe(wake.status); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].status).toBe("todo"); + expect(deps.executeRun).not.toHaveBeenCalled(); + expect(deps.setRunStatus).not.toHaveBeenCalled(); + }); + + it("admits the same queued input after the external connection is repaired", async () => { + const { run, agent } = await fixture("queued"); + await db.update(heartbeatRuns).set({ responsibleUserId: "queue-test-user" }).where(eq(heartbeatRuns.id, run.id)); + const deps = { ...callbacks(db), canStartNewExternalBotWork: vi.fn(async () => false) }; + deps.getAgent.mockResolvedValue(agent); + const queue = createHeartbeatQueue(db, deps); + expect(await queue.startNextQueuedRunForAgent(agent.id)).toEqual([]); + deps.canStartNewExternalBotWork.mockResolvedValue(true); + expect(await queue.startNextQueuedRunForAgent(agent.id)).toEqual([expect.objectContaining({ id: run.id, status: "running" })]); + expect(deps.executeRun).toHaveBeenCalledExactlyOnceWith(run.id); + }); + + it("rechecks the connection hold when Stop races the queue scan", async () => { + const { run, agent } = await fixture("queued"); + const deps = { ...callbacks(db), canStartNewExternalBotWork: vi.fn(async () => false) }; + deps.getAgent.mockResolvedValue(agent); + deps.canStartNewExternalBotWork.mockResolvedValueOnce(true); + expect(await createHeartbeatQueue(db, deps).startNextQueuedRunForAgent(agent.id)).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toMatchObject({ status: "queued", startedAt: null }); + expect(deps.canStartNewExternalBotWork).toHaveBeenCalledTimes(2); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + + it("keeps admitted resumptions outside the fresh external connection gate", async () => { + const { run, agent } = await fixture("queued"); + await db.update(heartbeatRuns).set({ startedAt: new Date(), responsibleUserId: "queue-test-user" }).where(eq(heartbeatRuns.id, run.id)); + const [admitted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + const deps = { ...callbacks(db), canStartNewExternalBotWork: vi.fn(async () => false) }; + deps.getAgent.mockResolvedValue(agent); + expect(await createHeartbeatQueue(db, deps).claimQueuedRun(admitted)).toMatchObject({ id: run.id, status: "running" }); + expect(deps.canStartNewExternalBotWork).not.toHaveBeenCalled(); + }); + it("cancels at the daily run cap before claiming or dispatching", async () => { const { run, agent } = await fixture("queued"); const cappedAgent = { ...agent, runtimeConfig: { heartbeat: { maxDailyRuns: 0 } } }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index b81906283a5..6a61871516b 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -196,6 +196,7 @@ export interface HeartbeatQueueDependencies extends Pick string | null; filterZombieCoalesceTarget: (run: T | null, live: { has(id: string): boolean }) => T | null; getSchedulingSuppression: () => Promise<{ suppressed: boolean; reason: "worktree_instance" | "database_restore_in_progress" | "task_drain" | null }>; + canStartNewExternalBotWork: (agent: typeof agents.$inferSelect) => Promise; executeRun: (runId: string) => Promise; releaseIssueExecutionAndPromote: (run: Pick, options?: { suppressImmediateRecovery?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise; applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise; @@ -371,6 +372,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende cancelActiveForAgentInternal, listCompanyAgentOrgRows, executeRun, + canStartNewExternalBotWork, activeRunExecutionPromises, activeWakeupPromises, instanceSettings, @@ -984,6 +986,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ); return null; } + if (!run.startedAt && !await canStartNewExternalBotWork(agent)) return null; const context = parseObject(run.contextSnapshot); const budgetBlock = await budgets.getInvocationBlock( @@ -1935,6 +1938,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ) .orderBy(asc(heartbeatRuns.createdAt)); if (queuedRuns.length === 0) return []; + if (queuedRuns.every(run => !run.startedAt) && !await canStartNewExternalBotWork(agent)) return []; const dependencyReadiness = await listQueuedRunDependencyReadiness( agent.companyId, diff --git a/tests/e2e/composer-stop.spec.ts b/tests/e2e/composer-stop.spec.ts index 46cc882fe14..a9c747f9bf5 100644 --- a/tests/e2e/composer-stop.spec.ts +++ b/tests/e2e/composer-stop.spec.ts @@ -278,12 +278,12 @@ for (const adapter of ["process", "paperclip_runner"] as const) { await request.get(`/api/issues/${parent.id}/comments`), ); expect(JSON.stringify(comments)).toContain("Please check mobile too."); - const queue = await json( - await request.get(`/api/issues/${parent.id}/queued-comments`), - ); - expect(JSON.stringify(queue.entries)).toContain( - "Please check mobile too.", - ); + await expect.poll(async () => { + const queue = await json( + await request.get(`/api/issues/${parent.id}/queued-comments`), + ); + return JSON.stringify(queue.entries); + }).toContain("Please check mobile too."); let dispatchedAt = 0; page.on("request", (req) => {