diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index a1eeab5167..37e8d31525 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -494,6 +494,29 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { }, ); + it("reports an unwired manager separately from a stopped sandbox worker", async () => { + const { companyId, environment, runId } = await seedReusablePluginSandboxLease(); + const input = { + companyId, + environment, + issueId: null, + heartbeatRunId: runId, + persistedExecutionWorkspace: null, + }; + const runtimeWithoutManager = environmentRuntimeService(db); + await expect(runtimeWithoutManager.acquireRunLease(input)) + .rejects.toThrow("sandbox plugin workers are unavailable in this server process"); + + const offlineManager = { isRunning: () => false, call: vi.fn() } as unknown as PluginWorkerManager; + const runtimeWithStoppedWorker = environmentRuntimeService(db, { + pluginWorkerManager: offlineManager, + pluginWorkerReadyTimeoutMs: 0, + }); + await expect(runtimeWithStoppedWorker.acquireRunLease(input)) + .rejects.toThrow("its worker is not running"); + expect(offlineManager.call).not.toHaveBeenCalled(); + }); + it("retains a successful reusable sandbox lease without stopping the provider resource", async () => { const { pluginId, runId, reusableLease } = await seedReusablePluginSandboxLease(); const workerManager = { diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 19acf5fd4f..212f19d00b 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -212,6 +212,7 @@ import { redactDetectedSuccessfulRunProgressSummaryForBoard, redactSuccessfulRunHandoffEvidence, } from "../services/heartbeat.ts"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.ts"; import { claimNativeRestartRecoveries, currentNativeControllerIdentity, @@ -4221,8 +4222,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); it("schedules an infra retry for a setup failure caused by a transient sandbox provider worker restart", async () => { - // Reproduces the production incident: the "Kubernetes Sandbox" plugin - // worker was mid-restart when a run tried to acquire a lease. The lease + // Model a configured plugin manager whose worker is mid-restart when + // a run tries to acquire a lease. A missing manager is a wiring error, + // not a worker restart. The lease // acquisition fails BEFORE the adapter is ever dispatched (no call to // mockAdapterExecute), so this hits the setup-failure catch (errorCode // "setup_failed") rather than the adapter-failure catch. The condition is @@ -4342,16 +4344,23 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { .set({ status: "in_progress" }) .where(eq(issues.id, issueId)); - const heartbeat = heartbeatService(db); + const heartbeat = heartbeatService(db, { + pluginWorkerManager: { + isRunning: () => false, + call: vi.fn(), + } as unknown as PluginWorkerManager, + }); await heartbeat.resumeQueuedRuns(); + // The real manager waits up to five seconds for worker readiness before + // lease acquisition fails and the heartbeat can schedule its retry. const runs = await waitForValue(async () => { const rows = await db .select() .from(heartbeatRuns) .where(eq(heartbeatRuns.agentId, agentId)); return rows.length >= 2 ? rows : null; - }); + }, 10_000); expect(runs).toHaveLength(2); const failedRun = runs?.find((row) => row.id === runId); diff --git a/server/src/__tests__/issue-tree-control-routes.test.ts b/server/src/__tests__/issue-tree-control-routes.test.ts index babfff8dc3..503e872a59 100644 --- a/server/src/__tests__/issue-tree-control-routes.test.ts +++ b/server/src/__tests__/issue-tree-control-routes.test.ts @@ -1,4 +1,5 @@ import express from "express"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { PgDialect } from "drizzle-orm/pg-core"; import request from "supertest"; import { beforeEach, describe, expect, it, vi } from "vitest"; @@ -28,14 +29,19 @@ const mockHeartbeatService = vi.hoisted(() => ({ wakeup: vi.fn(), })); +const mockHeartbeatFactory = vi.hoisted(() => vi.fn()); + vi.mock("../services/index.js", () => ({ - heartbeatService: () => mockHeartbeatService, + heartbeatService: mockHeartbeatFactory, issueService: () => mockIssueService, issueTreeControlService: () => mockTreeControlService, logActivity: mockLogActivity, })); -async function createApp(actor: Record) { +async function createApp( + actor: Record, + pluginWorkerManager?: PluginWorkerManager, +) { const [{ errorHandler }, { issueTreeControlRoutes }] = await Promise.all([ import("../middleware/index.js"), import("../routes/issue-tree-control.js"), @@ -52,7 +58,7 @@ async function createApp(actor: Record) { where: (predicate: unknown) => { mockReplayWhere(predicate); return mockReplayBlocks(); }, limit: mockReplayBlocks, }; - app.use("/api", issueTreeControlRoutes({ select: () => query } as any)); + app.use("/api", issueTreeControlRoutes({ select: () => query } as any, { pluginWorkerManager })); app.use(errorHandler); return app; } @@ -60,6 +66,7 @@ async function createApp(actor: Record) { describe("issue tree control routes", () => { beforeEach(() => { vi.clearAllMocks(); + mockHeartbeatFactory.mockReturnValue(mockHeartbeatService); mockReplayBlocks.mockResolvedValue([]); mockExecutionBlocker.mockResolvedValue(null); mockTreeControlService.getHold.mockResolvedValue(null); @@ -185,6 +192,39 @@ describe("issue tree control routes", () => { }, ); + it("resumes sandbox-backed work through the shared plugin worker manager", async () => { + const rootId = "11111111-1111-4111-8111-111111111111"; + const holdId = "33333333-3333-4333-8333-333333333333"; + const pluginWorkerManager = { isRunning: () => true } as unknown as PluginWorkerManager; + mockHeartbeatFactory.mockImplementation((_db, options) => ({ + ...mockHeartbeatService, + wakeup: async (...args: unknown[]) => { + // Lease acquisition needs the process's manager, even when its worker + // is already running. A service without it cannot dispatch this run. + expect(options?.pluginWorkerManager).toBe(pluginWorkerManager); + return mockHeartbeatService.wakeup(...args); + }, + })); + mockIssueService.getById.mockResolvedValue({ + id: rootId, companyId: "company-2", status: "todo", assigneeAgentId: "agent-parent", + }); + mockTreeControlService.releaseHold.mockResolvedValue({ + id: holdId, mode: "pause", status: "released", members: [{ issueId: rootId }], + }); + const app = await createApp({ + type: "board", userId: "user-1", companyIds: ["company-2"], source: "session", + }, pluginWorkerManager); + const response = await request(app) + .post(`/api/issues/${rootId}/tree-holds/${holdId}/release`) + .send({ metadata: { wakeAgents: true } }); + expect(response.status).toBe(200); + expect(response.body.wakeFailures).toBeUndefined(); + expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith("agent-parent", expect.objectContaining({ + reason: "issue_tree_resumed", + contextSnapshot: expect.objectContaining({ issueId: rootId, source: "issue.tree_resume" }), + })); + }); + it("reports wake failures without undoing release or skipping other assignees", async () => { const rootId = "11111111-1111-4111-8111-111111111111"; const holdId = "33333333-3333-4333-8333-333333333333"; diff --git a/server/src/app.ts b/server/src/app.ts index a1d603c5b3..a8345da723 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -740,7 +740,7 @@ export async function createApp( api.use(projectToolRoutes(db)); api.use(projectRoutes(db)); api.use(caseRoutes(db, opts.storageService)); - api.use(issueTreeControlRoutes(db)); + api.use(issueTreeControlRoutes(db, { pluginWorkerManager: workerManager })); api.use(fileResourceRoutes(db)); api.use(routineRoutes(db, { pluginWorkerManager: workerManager })); api.use(pipelineRoutes(db)); diff --git a/server/src/routes/issue-tree-control.ts b/server/src/routes/issue-tree-control.ts index c8c4327a4b..2a75da4be2 100644 --- a/server/src/routes/issue-tree-control.ts +++ b/server/src/routes/issue-tree-control.ts @@ -22,6 +22,8 @@ import { } from "../services/index.js"; import { assertBoard, getAccessibleResource, getActorInfo } from "./authz.js"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; + const TREE_RUN_CANCELLATION_RESPONSE_WAIT_MS = 1_000; const RESUME_EXECUTABLE_STATUSES = ["todo", "in_progress", "in_review"]; @@ -43,11 +45,16 @@ async function waitForRunCancellationTasks(tasks: Promise[]) { } } -export function issueTreeControlRoutes(db: Db) { +export function issueTreeControlRoutes( + db: Db, + options: { pluginWorkerManager?: PluginWorkerManager } = {}, +) { const router = Router(); const issuesSvc = issueService(db); const treeControlSvc = issueTreeControlService(db); - const heartbeat = heartbeatService(db); + const heartbeat = heartbeatService(db, { + pluginWorkerManager: options.pluginWorkerManager, + }); async function resolveRootIssue(req: Request) { const rootIssueId = req.params.id as string; diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 323a3d9843..b86ea142c8 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1785,16 +1785,18 @@ function createSandboxEnvironmentDriver( `Sandbox provider "${parsed.config.provider}" is installed via plugin "${pluginProvider.resolved.plugin.pluginKey}", but that plugin is currently ${pluginProvider.resolved.plugin.status}.`, ); } - if (pluginProvider.state === "worker_unavailable") { - throw new Error( - `Sandbox provider "${parsed.config.provider}" is installed via plugin "${pluginProvider.resolved.plugin.pluginKey}", but its worker is not running.`, - ); - } + // A missing manager is a wiring failure, even when the plugin worker + // is healthy elsewhere in this process. Check it before worker state. if (!pluginWorkerManager) { throw new Error( `Sandbox provider "${parsed.config.provider}" is installed, but sandbox plugin workers are unavailable in this server process.`, ); } + if (pluginProvider.state === "worker_unavailable") { + throw new Error( + `Sandbox provider "${parsed.config.provider}" is installed via plugin "${pluginProvider.resolved.plugin.pluginKey}", but its worker is not running.`, + ); + } const workerConfig = stripSandboxProviderEnvelope(parsed.config); const storedConfig = storedParsed.config;