mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 15:35:12 +01:00
fix(grok): hold new runs while connection needs recovery (#16445)
## Thinking Path > - Paperclip manages agents and their work. > - Grok Bot assignments use a managed Runner and a verified connection. > - Stop can leave the connection held for remote-work inspection. > - The queue started new runs before it checked that connection. > - Those runs allocated Runners, failed, and caused more preparation attempts. > - This pull request checks readiness at queue admission. > - Pending input waits for recovery without using managed compute. ## Linked Issues or Issue Description Refs #16427 and #16422. The open Grok changes cover the coding adapter and do not duplicate this fix. **What happened?** A real Stop test held the Grok connection for inspection. A second task was pending. Four new preparation attempts acquired managed Runners, then failed because the connection was held. No new external assignment or webhook was sent. Pause stopped the loop, and all leases completed cleanup. **Expected behavior** Keep fresh input in the durable queue until its connection is ready. Do not allocate a Runner to discover a known connection hold. Keep existing session reconciliation under its recorded authority. **Steps to reproduce** 1. Connect Grok Bot and start an assignment. 2. Queue another task for that agent. 3. Stop the first assignment and leave the inspection hold in place. 4. Run queue recovery. The pending task repeatedly starts preparation and fails. **Paperclip version or commit** Observed on packaged Core5a9e6b6315in Cloud staging. This branch includes current masterafe5b3418d. ## What Changed - Reuse the broker's current connection and authority checks before fresh Grok run claims. - Let valid connection-pool routing select a native provider before applying any unused Grok connection hold. - Leave pending input queued while readiness is denied. Recheck at the claim boundary if Stop races the queue scan. - Preserve already-started reconciliation and all later runtime-selection and dispatch checks. - Propagate database failures. Treat only explicit connection-authority errors as a hold. - Add queue, OAuth, broker and assembled heartbeat regressions. - Update the Grok guide and existing release evidence plan. ## Verification - Two queue admission assertions fail before the correction. - Six focused suites pass all 813 checks, including current master's native ownership coverage and a real healthy Codex pool member with a held Grok connection. The routing test reaches native runtime selection and then injects a controlled stop before provider start. - The assembled test uses real OAuth and broker state, repeats public queue recovery, and checks zero leases, assignments and extra webhooks. - Repository typecheck and build pass on final head83664a91ce. Its full local suite is still running. The earlier full attempt was stopped as superseded after the valid router review. - The full suite uses a clean tool PATH. On the earlier baseline, mocked Gemini tests reached the installed host CLI and exceeded their five-second deadline. Both tests pass with that host CLI absent. Test source and deadlines are unchanged. - The existing native cleanup and queue claim-isolation tests pass. The process-agent Stop browser journey passes locally. Its first CI attempt queried the queue 67 ms after comment creation, before asynchronous publication. The test now polls for the same durable entry with its existing deadline. - One server CI worker shut down after 1,047 tests passed. The failed attempt remains recorded. No routing flags or gates were changed. No UI component or schema changes. - The final head passes 55 CI checks, with two optional skips. Greptile reviewed that exact head at 5/5, with the earlier thread resolved and no new actionable finding. - Actual packaged recovery admission still needs a live repeat. The provider currently rejects routine starts with an external rate-limit error. This patch does not remove that provider limit. ## Risks Readiness can change after preflight. Runtime selection and dispatch retain their current authority checks. Already-started runs bypass this fresh-work check so they can reconcile their original session. Other providers keep their current queue behavior. Expected connection denials hold input; unexpected database errors still propagate. No migration is needed. ## Model Used OpenAI GPT-6 in Codex, with reasoning, repository editing, shell tools and computer use. The exact serving model ID, snapshot and context window are not exposed by this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
e4ce989099
commit
60e03c002d
9 files changed
+220
-14
No files matched your search
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim();
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let home: string;
|
||||
let isolatedHome: ReturnType<typeof createIsolatedTestHome> | 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);
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
@@ -102,6 +104,63 @@ async function acceptance(f: Awaited<ReturnType<typeof fixture>>, 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);
|
||||
|
||||
@@ -204,6 +204,15 @@ function createBroker(db: Db) {
|
||||
}
|
||||
});
|
||||
},
|
||||
async canStartNewWork(companyId: string, agentId: string, bindingId: string): Promise<boolean> {
|
||||
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<DotBindingSnapshot> {
|
||||
const state = await this.bindingForAgent(companyId, agentId);
|
||||
if (!state || !await enabled(state.providerKind)) throw fail("External provider is disabled for new work.");
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -48,6 +48,7 @@ function callbacks(db: Db) {
|
||||
runTaskKey: (run: Run) => typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null,
|
||||
filterZombieCoalesceTarget: <T extends { id: string; status: string }>(run: T | null) => run,
|
||||
getSchedulingSuppression: vi.fn(async () => ({ suppressed: false, reason: null } as Awaited<ReturnType<HeartbeatQueueDependencies["getSchedulingSuppression"]>>)),
|
||||
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 } } };
|
||||
|
||||
@@ -196,6 +196,7 @@ export interface HeartbeatQueueDependencies extends Pick<HeartbeatRetryDependenc
|
||||
runTaskKey: (run: HeartbeatRun) => string | null;
|
||||
filterZombieCoalesceTarget: <T extends { id: string; status: string }>(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<boolean>;
|
||||
executeRun: (runId: string) => Promise<void>;
|
||||
releaseIssueExecutionAndPromote: (run: Pick<HeartbeatRun, "id" | "companyId">, options?: { suppressImmediateRecovery?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise<unknown>;
|
||||
applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise<void>;
|
||||
@@ -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,
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
Reference in new issue
Block a user