From 02cd9bf648937013e069320f90570aceb4e9835b Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 10 Jul 2026 22:56:17 -0700 Subject: [PATCH] fix: suspension and restart wait for background execs (#104172) * fix: block restart and suspension on background execs * test: cover background exec suspension inspector --- docs/gateway/background-process.md | 2 + docs/gateway/external-apps.md | 6 +- .../src/gateway-suspend.test.ts | 30 ++++ .../src/schema/gateway-suspend.ts | 1 + src/agents/bash-process-registry.test.ts | 87 ++++++++++ src/agents/bash-process-registry.ts | 19 ++- .../bash-tools.exec-host-gateway.test.ts | 154 +++++++++++++++++- src/agents/bash-tools.exec-host-gateway.ts | 99 +++++++---- src/gateway/server-reload-handlers.test.ts | 115 +++++++++++++ src/gateway/server-reload-handlers.ts | 24 ++- src/gateway/server.impl.ts | 4 + .../plugins-http.suspension-admission.test.ts | 1 + src/infra/gateway-active-work.ts | 11 ++ src/infra/gateway-suspend-coordinator.test.ts | 57 +++++++ src/infra/restart-coordinator.test.ts | 83 +++++++++- src/infra/restart-coordinator.ts | 25 ++- src/infra/restart-suspension.test.ts | 1 + 17 files changed, 675 insertions(+), 44 deletions(-) diff --git a/docs/gateway/background-process.md b/docs/gateway/background-process.md index bbb62d4a848a..19615df7ca7c 100644 --- a/docs/gateway/background-process.md +++ b/docs/gateway/background-process.md @@ -81,6 +81,8 @@ Actions: Notes: - Only backgrounded sessions are listed/persisted — in memory only, not on disk. Sessions are lost on process restart. +- A live background session blocks cooperative host suspension and safe Gateway restart until the process owner confirms its actual exit. +- `process remove` can hide a running session immediately after requesting termination; suspension and restart remain blocked until exit confirmation. - Session logs are only saved to chat history if you run `process poll`/`log` and the tool result is recorded. - `process` is scoped per agent; it only sees sessions started by that agent. - Use `poll`/`log` for status, logs, or completion confirmation when automatic completion wake is unavailable. diff --git a/docs/gateway/external-apps.md b/docs/gateway/external-apps.md index 5dbc2c767d9c..4cd539fb20cc 100644 --- a/docs/gateway/external-apps.md +++ b/docs/gateway/external-apps.md @@ -134,8 +134,10 @@ transports, or control the hosting platform. The host must fence its ingress before preparation and remains responsible for wake, snapshot/freeze, and stop. `activeCount` is the aggregate tracked-work count, while `blockers` contains the non-zero category counts and bounded task details. This is not a -general process-quiescence barrier. Channel health, maintenance, cache refresh, -established plugin WebSocket sessions, and plugin-owned background work can +general process-quiescence barrier. A `background-exec` blocker is aggregate +only: command text, process IDs, output, and session or scope identifiers never +cross the protocol. Channel health, maintenance, cache refresh, established +plugin WebSocket sessions, and unregistered plugin-owned background work can remain active. The hosting platform must freeze or snapshot the full process tree and its filesystem consistently; unregistered work cannot be proven idle by this first diff --git a/packages/gateway-protocol/src/gateway-suspend.test.ts b/packages/gateway-protocol/src/gateway-suspend.test.ts index 9e6b07ce62bf..e0e33ca4da31 100644 --- a/packages/gateway-protocol/src/gateway-suspend.test.ts +++ b/packages/gateway-protocol/src/gateway-suspend.test.ts @@ -44,6 +44,36 @@ describe("gateway suspension protocol", () => { ).toBe(true); }); + it("keeps background exec blockers count-only", () => { + const result = { + status: "busy", + reason: "active-work", + retryAfterMs: 20_000, + activeCount: 1, + blockers: [ + { + kind: "background-exec", + count: 1, + message: "1 active background exec session(s)", + }, + ], + }; + + expect(validateGatewaySuspendPrepareResult(result)).toBe(true); + expect( + validateGatewaySuspendPrepareResult({ + ...result, + blockers: [{ ...result.blockers[0], sessionIds: ["private-session-id"] }], + }), + ).toBe(false); + expect( + validateGatewaySuspendPrepareResult({ + ...result, + blockers: [{ ...result.blockers[0], command: "private command" }], + }), + ).toBe(false); + }); + it("validates status and resume results", () => { expect(validateGatewaySuspendStatusResult({ status: "running" })).toBe(true); expect(validateGatewaySuspendStatusResult({ status: "ready", expiresAtMs: 123 })).toBe(true); diff --git a/packages/gateway-protocol/src/schema/gateway-suspend.ts b/packages/gateway-protocol/src/schema/gateway-suspend.ts index b8b37569391b..9de821a53aa2 100644 --- a/packages/gateway-protocol/src/schema/gateway-suspend.ts +++ b/packages/gateway-protocol/src/schema/gateway-suspend.ts @@ -27,6 +27,7 @@ export const GatewaySuspendBlockerSchema = Type.Object( Type.Literal("queue"), Type.Literal("reply"), Type.Literal("embedded-run"), + Type.Literal("background-exec"), Type.Literal("cron-run"), Type.Literal("task"), Type.Literal("root-request"), diff --git a/src/agents/bash-process-registry.test.ts b/src/agents/bash-process-registry.test.ts index 2f2352a16b59..e0f7b085288c 100644 --- a/src/agents/bash-process-registry.test.ts +++ b/src/agents/bash-process-registry.test.ts @@ -9,8 +9,12 @@ import type { ProcessSession } from "./bash-process-registry.js"; import { addSession, appendOutput, + createSessionSlug, + deleteSession, drainSession, + getActiveBackgroundExecSessionCount, listFinishedSessions, + listRunningSessions, markBackgrounded, markExited, resetProcessRegistryForTests, @@ -19,6 +23,14 @@ import { } from "./bash-process-registry.js"; import { createProcessSessionFixture } from "./bash-process-registry.test-helpers.js"; +const randomMocks = vi.hoisted(() => ({ + generateSecureInt: vi.fn(() => 0), +})); + +vi.mock("../infra/secure-random.js", () => ({ + generateSecureInt: randomMocks.generateSecureInt, +})); + describe("bash process registry", () => { function createRegistrySession(params: { id?: string; @@ -37,6 +49,8 @@ describe("bash process registry", () => { } beforeEach(() => { + randomMocks.generateSecureInt.mockReset(); + randomMocks.generateSecureInt.mockReturnValue(0); resetProcessRegistryForTests(); }); @@ -173,6 +187,79 @@ describe("bash process registry", () => { ]); }); + it("tracks only live backgrounded sessions", () => { + const session = createRegistrySession({ + maxOutputChars: 100, + pendingMaxOutputChars: 30_000, + backgrounded: false, + }); + + addSession(session); + expect(getActiveBackgroundExecSessionCount()).toBe(0); + + markBackgrounded(session); + markBackgrounded(session); + expect(getActiveBackgroundExecSessionCount()).toBe(1); + + markExited(session, 0, null, "completed"); + expect(getActiveBackgroundExecSessionCount()).toBe(0); + + markBackgrounded(session); + expect(getActiveBackgroundExecSessionCount()).toBe(0); + }); + + it("keeps a hidden background session active until its process exits", () => { + const session = createRegistrySession({ + id: "hidden-until-exit", + maxOutputChars: 100, + pendingMaxOutputChars: 30_000, + backgrounded: false, + }); + + addSession(session); + markBackgrounded(session); + deleteSession(session.id); + + expect(listRunningSessions()).toHaveLength(0); + expect(getActiveBackgroundExecSessionCount()).toBe(1); + + markExited(session, null, "SIGTERM", "killed"); + expect(getActiveBackgroundExecSessionCount()).toBe(0); + }); + + it("keeps a hidden active session id reserved until exit", () => { + const session = createRegistrySession({ + id: "amber-atlas", + maxOutputChars: 100, + pendingMaxOutputChars: 30_000, + backgrounded: false, + }); + + addSession(session); + markBackgrounded(session); + deleteSession(session.id); + expect(createSessionSlug()).toBe("amber-atlas-2"); + + session.backgrounded = false; + markExited(session, 0, null, "completed"); + expect(createSessionSlug()).toBe("amber-atlas"); + }); + + it("clears background activity in the test reset", () => { + const session = createRegistrySession({ + maxOutputChars: 100, + pendingMaxOutputChars: 30_000, + backgrounded: false, + }); + + addSession(session); + markBackgrounded(session); + expect(getActiveBackgroundExecSessionCount()).toBe(1); + + resetProcessRegistryForTests(); + expect(getActiveBackgroundExecSessionCount()).toBe(0); + }); + it("clamps a zero retention TTL to one minute", () => { vi.useFakeTimers(); try { diff --git a/src/agents/bash-process-registry.ts b/src/agents/bash-process-registry.ts index 3ed078d3138a..bf95aa3b6b35 100644 --- a/src/agents/bash-process-registry.ts +++ b/src/agents/bash-process-registry.ts @@ -109,11 +109,14 @@ interface FinishedSession { const runningSessions = new Map(); const finishedSessions = new Map(); +const activeBackgroundExecSessionIds = new Set(); let sweeper: NodeJS.Timeout | null = null; function isSessionIdTaken(id: string) { - return runningSessions.has(id) || finishedSessions.has(id); + return ( + runningSessions.has(id) || finishedSessions.has(id) || activeBackgroundExecSessionIds.has(id) + ); } /** Creates a unique short session id that avoids running and retained sessions. */ @@ -137,7 +140,7 @@ export function getFinishedSession(id: string) { return finishedSessions.get(id); } -/** Removes a session from both running and finished registries. */ +/** Removes visible session records without changing live-process activity. */ export function deleteSession(id: string) { runningSessions.delete(id); finishedSessions.delete(id); @@ -194,6 +197,9 @@ export function markExited( exitReason?: TerminationReason, noOutputTimedOut?: boolean, ) { + // Visibility can be cleared before process termination. Keep suspension + // blocked until the process owner reports the actual terminal transition. + activeBackgroundExecSessionIds.delete(session.id); session.exited = true; session.exitCode = exitCode; session.exitSignal = exitSignal; @@ -206,6 +212,14 @@ export function markExited( /** Marks a running session as reconnectable after the exec call returns. */ export function markBackgrounded(session: ProcessSession) { session.backgrounded = true; + if (!session.exited) { + activeBackgroundExecSessionIds.add(session.id); + } +} + +/** Returns the number of live background exec sessions without exposing process details. */ +export function getActiveBackgroundExecSessionCount(): number { + return activeBackgroundExecSessionIds.size; } function moveToFinished(session: ProcessSession, status: ProcessStatus) { @@ -334,6 +348,7 @@ export function listFinishedSessions() { export function resetProcessRegistryForTests() { runningSessions.clear(); finishedSessions.clear(); + activeBackgroundExecSessionIds.clear(); stopSweeper(); } diff --git a/src/agents/bash-tools.exec-host-gateway.test.ts b/src/agents/bash-tools.exec-host-gateway.test.ts index 2594543d568c..4c0efc049ee2 100644 --- a/src/agents/bash-tools.exec-host-gateway.test.ts +++ b/src/agents/bash-tools.exec-host-gateway.test.ts @@ -7,7 +7,7 @@ import crypto from "node:crypto"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { beforeAll, beforeEach, describe, expect, it, vi, type Mock } from "vitest"; +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi, type Mock } from "vitest"; import { onInternalDiagnosticEvent, resetDiagnosticEventsForTest, @@ -29,8 +29,18 @@ import { } from "../infra/exec-authorization-plan.js"; import { buildAuthorizedShellCommandFromPlan } from "../infra/exec-authorization-render.js"; import { resolvePolicyTargetCandidatePath } from "../infra/exec-command-resolution.js"; +import { createSafeGatewayRestartPreflight } from "../infra/restart-coordinator.js"; +import { + getActiveGatewayRootWorkCount, + markGatewayRestartDraining, + resetGatewayWorkAdmission, + tryBeginGatewaySuspendAdmission, +} from "../process/gateway-work-admission.js"; import type { ExecApprovalFollowupTarget } from "./bash-tools.exec-host-shared.js"; -import type { ExecApprovalFollowupFactory } from "./bash-tools.exec-types.js"; +import type { + ExecApprovalFollowupFactory, + ExecApprovalFollowupOutcome, +} from "./bash-tools.exec-types.js"; type StrictInlineEvalBoundary = typeof import("./bash-tools.exec-host-shared.js").enforceStrictInlineEvalApprovalBoundary; @@ -155,6 +165,7 @@ const resolveExecHostApprovalContextMock = vi.hoisted(() => ), ); const runExecProcessMock = vi.hoisted(() => vi.fn()); +const markBackgroundedMock = vi.hoisted(() => vi.fn()); const sendExecApprovalFollowupResultMock = vi.hoisted(() => vi.fn(async () => undefined), ); @@ -222,7 +233,8 @@ vi.mock("./bash-tools.exec-runtime.js", () => ({ })); vi.mock("./bash-process-registry.js", () => ({ - markBackgrounded: vi.fn(), + getActiveBackgroundExecSessionCount: vi.fn(() => 0), + markBackgrounded: markBackgroundedMock, tail: vi.fn((value) => value), })); @@ -291,6 +303,7 @@ describe("processGatewayAllowlist", () => { }); beforeEach(() => { + resetGatewayWorkAdmission(); resetDiagnosticEventsForTest(); buildExecApprovalPendingToolResultMock.mockReset(); buildExecApprovalFollowupTargetMock.mockReset(); @@ -341,6 +354,7 @@ describe("processGatewayAllowlist", () => { askFallback: "deny", }); runExecProcessMock.mockReset(); + markBackgroundedMock.mockReset(); sendExecApprovalFollowupResultMock.mockReset(); enforceStrictInlineEvalApprovalBoundaryMock.mockReset(); enforceStrictInlineEvalApprovalBoundaryMock.mockImplementation((value) => ({ @@ -367,6 +381,10 @@ describe("processGatewayAllowlist", () => { }); }); + afterEach(() => { + resetGatewayWorkAdmission(); + }); + function runGatewayAllowlist( overrides: Partial & Pick, ) { @@ -2010,6 +2028,136 @@ EOF`, }, ); + it("waits outside admission, then atomically hands an approved process to the registry", async () => { + let resolveApproval: (decision: ExecApprovalDecision) => void = () => {}; + const approval = new Promise((resolve) => { + resolveApproval = resolve; + }); + let resolveOutcome: (outcome: ExecApprovalFollowupOutcome) => void = () => {}; + const outcome = new Promise((resolve) => { + resolveOutcome = resolve; + }); + let allowSpawn: () => void = () => {}; + const spawnAllowed = new Promise((resolve) => { + allowSpawn = resolve; + }); + let announceSpawn: () => void = () => {}; + const spawnStarted = new Promise((resolve) => { + announceSpawn = resolve; + }); + resolveApprovalDecisionOrUndefinedMock.mockReturnValue(approval); + createExecApprovalDecisionStateMock.mockReturnValue({ + baseDecision: { timedOut: false }, + approvedByAsk: true, + deniedReason: null, + }); + commitExecAuthorizationMock.mockImplementation(async () => { + expect(getActiveGatewayRootWorkCount()).toBe(1); + }); + runExecProcessMock.mockImplementation(async () => { + expect(getActiveGatewayRootWorkCount()).toBe(1); + announceSpawn(); + await spawnAllowed; + return { session: { id: "sess-atomic" }, promise: outcome }; + }); + markBackgroundedMock.mockImplementation(() => { + expect(getActiveGatewayRootWorkCount()).toBe(1); + }); + buildExecApprovalFollowupTargetMock.mockImplementation((value) => value); + + const result = await runGatewayAllowlist({ + command: "find . -maxdepth 1", + turnSourceChannel: "feishu", + }); + expect(result.pendingResult?.details.status).toBe("approval-pending"); + await vi.waitFor(() => { + expect(resolveApprovalDecisionOrUndefinedMock).toHaveBeenCalledOnce(); + }); + expect(getActiveGatewayRootWorkCount()).toBe(0); + + const suspension = tryBeginGatewaySuspendAdmission(() => {}); + expect(suspension?.commit()).toBe(true); + resolveApproval("allow-once"); + await Promise.resolve(); + await Promise.resolve(); + expect(commitExecAuthorizationMock).not.toHaveBeenCalled(); + expect(runExecProcessMock).not.toHaveBeenCalled(); + expect(markBackgroundedMock).not.toHaveBeenCalled(); + + suspension?.release(); + await spawnStarted; + expect( + createSafeGatewayRestartPreflight({ + getQueueSize: () => 0, + getPendingReplies: () => 0, + getEmbeddedRuns: () => 0, + getCronRuns: () => 0, + getBackgroundExecSessions: () => 0, + getActiveTasks: () => 0, + getTaskBlockers: () => [], + }), + ).toMatchObject({ + safe: false, + counts: { rootRequests: 1, totalActive: 1 }, + blockers: [{ kind: "root-request", count: 1 }], + }); + allowSpawn(); + await vi.waitFor(() => { + expect(markBackgroundedMock).toHaveBeenCalledOnce(); + expect(getActiveGatewayRootWorkCount()).toBe(0); + }); + expect(commitExecAuthorizationMock).toHaveBeenCalledOnce(); + expect(runExecProcessMock).toHaveBeenCalledOnce(); + + resolveOutcome({ + status: "completed", + exitCode: 0, + timedOut: false, + aggregated: "done", + }); + await vi.waitFor(() => { + expect(sendExecApprovalFollowupResultMock).toHaveBeenCalledOnce(); + }); + }); + + it("denies a detached approved process when restart drain wins admission", async () => { + let resolveApproval: (decision: ExecApprovalDecision) => void = () => {}; + resolveApprovalDecisionOrUndefinedMock.mockReturnValue( + new Promise((resolve) => { + resolveApproval = resolve; + }), + ); + createExecApprovalDecisionStateMock.mockReturnValue({ + baseDecision: { timedOut: false }, + approvedByAsk: true, + deniedReason: null, + }); + buildExecApprovalFollowupTargetMock.mockImplementation((value) => value); + + const result = await runGatewayAllowlist({ + command: "find . -maxdepth 1", + turnSourceChannel: "feishu", + }); + expect(result.pendingResult?.details.status).toBe("approval-pending"); + await vi.waitFor(() => { + expect(resolveApprovalDecisionOrUndefinedMock).toHaveBeenCalledOnce(); + }); + + markGatewayRestartDraining(); + resolveApproval("allow-once"); + await vi.waitFor(() => { + expect(sendExecApprovalFollowupResultMock).toHaveBeenCalledWith( + expect.anything(), + "Exec denied (gateway id=req-1, gateway-draining): find . -maxdepth 1", + ); + }); + expect(sendExecApprovalFollowupResultMock).toHaveBeenCalledOnce(); + expect(commitExecAuthorizationMock).not.toHaveBeenCalled(); + expect(runExecProcessMock).not.toHaveBeenCalled(); + expect(markBackgroundedMock).not.toHaveBeenCalled(); + expect(getActiveGatewayRootWorkCount()).toBe(0); + }); + it("keeps the fire-and-forget path for channels without native approval clients", async () => { resolveApprovalDecisionOrUndefinedMock.mockResolvedValue("allow-once"); createExecApprovalDecisionStateMock.mockReturnValue({ diff --git a/src/agents/bash-tools.exec-host-gateway.ts b/src/agents/bash-tools.exec-host-gateway.ts index 612bea0feb98..894db0771b5c 100644 --- a/src/agents/bash-tools.exec-host-gateway.ts +++ b/src/agents/bash-tools.exec-host-gateway.ts @@ -39,6 +39,10 @@ import { type ExecAutoReviewInput, } from "../infra/exec-auto-review.js"; import type { SafeBinProfile } from "../infra/exec-safe-bin-policy.js"; +import { + GatewayDrainingError, + runWithGatewayIndependentRootWorkAdmission, +} from "../process/gateway-work-admission.js"; import { isNativeApprovalChannel, normalizeMessageChannel } from "../utils/message-channel.js"; import { markBackgrounded, tail } from "./bash-process-registry.js"; import { @@ -1120,40 +1124,75 @@ export async function processGatewayAllowlist( return; } + let admitted: + | { status: "started"; run: Awaited> } + | { status: "approval-state-write-failed" } + | { status: "spawn-failed" }; try { - await commitExecutionAuthorization({ - source: approvalDecision.authorizationSource, - resolvedPath: resolvedPath ?? undefined, - ...(approvalDecision.allowAlwaysDecision - ? { allowAlwaysDecision: approvalDecision.allowAlwaysDecision } - : {}), + admitted = await runWithGatewayIndependentRootWorkAdmission(async () => { + try { + await commitExecutionAuthorization({ + source: approvalDecision.authorizationSource, + resolvedPath: resolvedPath ?? undefined, + ...(approvalDecision.allowAlwaysDecision + ? { allowAlwaysDecision: approvalDecision.allowAlwaysDecision } + : {}), + }); + } catch { + return { status: "approval-state-write-failed" as const }; + } + + let run: Awaited>; + try { + run = await runExecProcess({ + command: params.command, + execCommand: approvalDecision.execCommandOverride, + workdir: params.workdir, + env: params.env, + pathPrepend: params.pathPrepend, + sandbox: undefined, + containerWorkdir: null, + usePty: params.pty, + warnings: params.warnings, + maxOutput: params.maxOutput, + pendingMaxOutput: params.pendingMaxOutput, + notifyOnExit: false, + notifyOnExitEmptySuccess: false, + scopeKey: params.scopeKey, + sessionKey: params.notifySessionKey ?? params.sessionKey, + timeoutSec: effectiveTimeout, + }); + } catch { + return { status: "spawn-failed" as const }; + } + + // Keep the admitted root until the registry owns the live process. + // Suspension must observe one side of this handoff at every instant. + markBackgrounded(run.session); + return { status: "started" as const, run }; }); - } catch { + } catch (error) { + if ( + error instanceof GatewayDrainingError || + (error instanceof Error && error.message === "gateway is draining for restart") + ) { + await sendExecApprovalFollowupResult( + followupTarget, + `Exec denied (gateway id=${approvalId}, gateway-draining): ${params.command}`, + ); + return; + } + // Detached approval work must always settle through a follow-up. Treat + // any unexpected admission failure as a spawn failure, never an + // unhandled rejection from this fire-and-forget chain. + admitted = { status: "spawn-failed" }; + } + + if (admitted.status === "approval-state-write-failed") { await denyApprovalStateWriteFailure(); return; } - - let run: Awaited> | null; - try { - run = await runExecProcess({ - command: params.command, - execCommand: approvalDecision.execCommandOverride, - workdir: params.workdir, - env: params.env, - pathPrepend: params.pathPrepend, - sandbox: undefined, - containerWorkdir: null, - usePty: params.pty, - warnings: params.warnings, - maxOutput: params.maxOutput, - pendingMaxOutput: params.pendingMaxOutput, - notifyOnExit: false, - notifyOnExitEmptySuccess: false, - scopeKey: params.scopeKey, - sessionKey: params.notifySessionKey ?? params.sessionKey, - timeoutSec: effectiveTimeout, - }); - } catch { + if (admitted.status === "spawn-failed") { await sendExecApprovalFollowupResult( followupTarget, `Exec denied (gateway id=${approvalId}, spawn-failed): ${params.command}`, @@ -1161,7 +1200,7 @@ export async function processGatewayAllowlist( return; } - markBackgrounded(run.session); + const { run } = admitted; const outcome = await run.promise; const dynamicFollowupText = await resolveGatewayExecApprovalFollowupText({ diff --git a/src/gateway/server-reload-handlers.test.ts b/src/gateway/server-reload-handlers.test.ts index 0b18bd6754d2..894675edb8cc 100644 --- a/src/gateway/server-reload-handlers.test.ts +++ b/src/gateway/server-reload-handlers.test.ts @@ -2,6 +2,13 @@ * Gateway config reload handler tests. */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + addSession, + markBackgrounded, + markExited, + resetProcessRegistryForTests, +} from "../agents/bash-process-registry.js"; +import { createProcessSessionFixture } from "../agents/bash-process-registry.test-helpers.js"; import type { ConfigWriteNotification } from "../config/config.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { @@ -16,6 +23,7 @@ import { import { isGatewayWorkAdmissionClosed, resetGatewayWorkAdmission, + tryBeginGatewayIndependentRootWorkAdmission, tryBeginGatewayRootWorkAdmission, } from "../process/gateway-work-admission.js"; import { createEmptyRuntimeWebToolsMetadata } from "../secrets/runtime-fast-path.js"; @@ -234,12 +242,14 @@ function createReloadHandlersForTest( // shared vitest worker imports those helpers before this file runs, the leaked // env routes reloads into the skip branch and channel restarts never fire. beforeEach(() => { + resetProcessRegistryForTests(); delete process.env.OPENCLAW_SKIP_CHANNELS; delete process.env.OPENCLAW_SKIP_PROVIDERS; }); afterEach(() => { vi.useRealTimers(); + resetProcessRegistryForTests(); hoisted.startGmailWatcherWithLogs.mockClear(); hoisted.stopGmailWatcher.mockClear(); hoisted.activeTaskCount.value = 0; @@ -500,6 +510,111 @@ describe("gateway restart deferral preflight", () => { } }); + it("defers config restart until a background exec actually exits", async () => { + restartTesting.resetSigusr1State(); + resetGatewayWorkAdmission(); + const logReload = { info: vi.fn(), warn: vi.fn() }; + const { requestGatewayRestart } = createReloadHandlersForTest(logReload); + const session = createProcessSessionFixture({ + id: "background-restart-blocker", + command: "private command", + pid: 12345, + }); + addSession(session); + markBackgrounded(session); + const signalSpy = vi.fn(); + process.once("SIGUSR1", signalSpy); + vi.useFakeTimers(); + + try { + expect( + requestGatewayRestart( + { + changedPaths: ["gateway.port"], + restartGateway: true, + restartReasons: ["gateway.port"], + hotReasons: [], + reloadHooks: false, + restartGmailWatcher: false, + restartCron: false, + restartHeartbeat: false, + restartHealthMonitor: false, + reloadPlugins: false, + restartChannels: new Set(), + disposeMcpRuntimes: false, + noopPaths: [], + }, + {}, + ), + ).toBe(true); + + expect(signalSpy).not.toHaveBeenCalled(); + expect(logReload.warn).toHaveBeenCalledWith( + "config change requires gateway restart (gateway.port) — deferring until 1 background exec session(s) complete", + ); + + markExited(session, 0, null, "completed"); + await vi.advanceTimersByTimeAsync(500); + + expect(signalSpy).toHaveBeenCalledOnce(); + expect(logReload.info).toHaveBeenCalledWith( + "all operations and replies completed; restarting gateway now", + ); + } finally { + process.removeListener("SIGUSR1", signalSpy); + restartTesting.resetSigusr1State(); + resetGatewayWorkAdmission(); + } + }); + + it("defers config restart across an admitted process handoff", async () => { + restartTesting.resetSigusr1State(); + resetGatewayWorkAdmission(); + const logReload = { info: vi.fn(), warn: vi.fn() }; + const { requestGatewayRestart } = createReloadHandlersForTest(logReload); + const handoff = tryBeginGatewayIndependentRootWorkAdmission(); + const signalSpy = vi.fn(); + process.once("SIGUSR1", signalSpy); + vi.useFakeTimers(); + + try { + expect( + requestGatewayRestart( + { + changedPaths: ["gateway.port"], + restartGateway: true, + restartReasons: ["gateway.port"], + hotReasons: [], + reloadHooks: false, + restartGmailWatcher: false, + restartCron: false, + restartHeartbeat: false, + restartHealthMonitor: false, + reloadPlugins: false, + restartChannels: new Set(), + disposeMcpRuntimes: false, + noopPaths: [], + }, + {}, + ), + ).toBe(true); + expect(signalSpy).not.toHaveBeenCalled(); + expect(logReload.warn).toHaveBeenCalledWith( + "config change requires gateway restart (gateway.port) — deferring until 1 gateway request(s) complete", + ); + + handoff?.release(); + await vi.advanceTimersByTimeAsync(500); + + expect(signalSpy).toHaveBeenCalledOnce(); + } finally { + handoff?.release(); + process.removeListener("SIGUSR1", signalSpy); + restartTesting.resetSigusr1State(); + resetGatewayWorkAdmission(); + } + }); + it("defers channel hot reload until active embedded work drains", async () => { const previousSkipChannels = process.env.OPENCLAW_SKIP_CHANNELS; const previousSkipProviders = process.env.OPENCLAW_SKIP_PROVIDERS; diff --git a/src/gateway/server-reload-handlers.ts b/src/gateway/server-reload-handlers.ts index 5e0e142cba09..4260211c3ba2 100644 --- a/src/gateway/server-reload-handlers.ts +++ b/src/gateway/server-reload-handlers.ts @@ -1,6 +1,7 @@ // Gateway hot-reload handlers. // Applies config reload plans to hooks, cron, heartbeat, plugins, channels, and restarts. import { disposeAllSessionMcpRuntimes } from "../agents/agent-bundle-mcp-tools.js"; +import { getActiveBackgroundExecSessionCount } from "../agents/bash-process-registry.js"; import { refreshContextWindowCache } from "../agents/context.js"; import { getActiveEmbeddedRunCount, @@ -28,7 +29,10 @@ import { setGatewaySigusr1RestartPolicy, } from "../infra/restart.js"; import { getTotalQueueSize } from "../process/command-queue.js"; -import { runWithGatewayIndependentRootWorkAdmission } from "../process/gateway-work-admission.js"; +import { + getActiveGatewayRootWorkCount, + runWithGatewayIndependentRootWorkAdmission, +} from "../process/gateway-work-admission.js"; import { clearSecretsRuntimeSnapshot, getActiveSecretsRuntimeSnapshot, @@ -234,13 +238,23 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) const queueSize = getTotalQueueSize(); const pendingReplies = getTotalPendingReplies(); const embeddedRuns = getActiveEmbeddedRunCount(); + const backgroundExecSessions = getActiveBackgroundExecSessionCount(); + const rootRequests = getActiveGatewayRootWorkCount({ excludeCurrent: true }); const activeTasks = getInspectableActiveTaskRestartBlockers().length; return { queueSize, pendingReplies, embeddedRuns, + backgroundExecSessions, + rootRequests, activeTasks, - totalActive: queueSize + pendingReplies + embeddedRuns + activeTasks, + totalActive: + queueSize + + pendingReplies + + embeddedRuns + + backgroundExecSessions + + rootRequests + + activeTasks, }; }; const formatActiveDetails = (counts: ReturnType) => { @@ -254,6 +268,12 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) if (counts.embeddedRuns > 0) { details.push(`${counts.embeddedRuns} embedded run(s)`); } + if (counts.backgroundExecSessions > 0) { + details.push(`${counts.backgroundExecSessions} background exec session(s)`); + } + if (counts.rootRequests > 0) { + details.push(`${counts.rootRequests} gateway request(s)`); + } if (counts.activeTasks > 0) { details.push(`${counts.activeTasks} background task run(s)`); } diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts index 9eb529b91b5d..8edf28405bc4 100644 --- a/src/gateway/server.impl.ts +++ b/src/gateway/server.impl.ts @@ -3,6 +3,7 @@ import type { IncomingMessage, ServerResponse } from "node:http"; // and WebSocket surfaces, config reload hooks, and graceful restart/shutdown. import { monitorEventLoopDelay, performance } from "node:perf_hooks"; import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; +import { getActiveBackgroundExecSessionCount } from "../agents/bash-process-registry.js"; import { getActiveEmbeddedRunCount, resolveActiveEmbeddedRunSessionId, @@ -57,6 +58,7 @@ import { pinActivePluginSessionExtensionRegistry, } from "../plugins/runtime.js"; import { getTotalQueueSize, isGatewayDraining } from "../process/command-queue.js"; +import { getActiveGatewayRootWorkCount } from "../process/gateway-work-admission.js"; import type { RuntimeEnv } from "../runtime.js"; import { clearSecretsRuntimeSnapshot, @@ -651,6 +653,8 @@ export async function startGatewayServer( getTotalPendingReplies() + getActiveEmbeddedRunCount() + getActiveCronJobCount() + + getActiveBackgroundExecSessionCount() + + getActiveGatewayRootWorkCount() + getActiveTaskCount(), ); // Unconditional startup migration: seed gateway.controlUi.allowedOrigins for existing diff --git a/src/gateway/server/plugins-http.suspension-admission.test.ts b/src/gateway/server/plugins-http.suspension-admission.test.ts index ab52491f1643..09c22c533720 100644 --- a/src/gateway/server/plugins-http.suspension-admission.test.ts +++ b/src/gateway/server/plugins-http.suspension-admission.test.ts @@ -57,6 +57,7 @@ function createRootOnlyInspectors(): GatewayActiveWorkInspectors { getQueueSize: () => 0, getPendingReplies: () => 0, getEmbeddedRuns: () => 0, + getBackgroundExecSessions: () => 0, getCronRuns: () => 0, getActiveTasks: () => 0, getTaskBlockers: () => [], diff --git a/src/infra/gateway-active-work.ts b/src/infra/gateway-active-work.ts index a72f6e632fe3..bc04c4224a87 100644 --- a/src/infra/gateway-active-work.ts +++ b/src/infra/gateway-active-work.ts @@ -1,4 +1,5 @@ // Collects process activity shared by restart and host-suspension decisions. +import { getActiveBackgroundExecSessionCount } from "../agents/bash-process-registry.js"; import { getActiveEmbeddedRunCount } from "../agents/embedded-agent-runner/run-state.js"; import { getTotalPendingReplies } from "../auto-reply/reply/dispatcher-registry.js"; import { getActiveCronJobCount } from "../cron/active-jobs.js"; @@ -19,6 +20,7 @@ export type GatewayActiveWorkCounts = { queueSize: number; pendingReplies: number; embeddedRuns: number; + backgroundExecSessions: number; cronRuns: number; activeTasks: number; rootRequests: number; @@ -37,6 +39,7 @@ export type GatewayActiveWorkBlocker = { | "queue" | "reply" | "embedded-run" + | "background-exec" | "cron-run" | "task" | "root-request" @@ -61,6 +64,7 @@ export type GatewayActiveWorkInspectors = { getQueueSize: () => number; getPendingReplies: () => number; getEmbeddedRuns: () => number; + getBackgroundExecSessions: () => number; getCronRuns: () => number; getActiveTasks: () => number; getTaskBlockers: () => ActiveTaskRestartBlocker[]; @@ -77,6 +81,7 @@ const defaultInspectors: GatewayActiveWorkInspectors = { getQueueSize: getTotalQueueSize, getPendingReplies: getTotalPendingReplies, getEmbeddedRuns: getActiveEmbeddedRunCount, + getBackgroundExecSessions: getActiveBackgroundExecSessionCount, getCronRuns: () => Math.max(getActiveCronJobCount(), getSuspensionVisibleCronTaskRunCount()), getActiveTasks: () => getInspectableActiveTaskRestartBlockers().length, getTaskBlockers: getInspectableActiveTaskRestartBlockers, @@ -101,6 +106,7 @@ export function createGatewayActiveWorkSnapshot( queueSize: normalizeCount(resolved.getQueueSize()), pendingReplies: normalizeCount(resolved.getPendingReplies()), embeddedRuns: normalizeCount(resolved.getEmbeddedRuns()), + backgroundExecSessions: normalizeCount(resolved.getBackgroundExecSessions()), cronRuns: normalizeCount(resolved.getCronRuns()), activeTasks: normalizeCount(resolved.getActiveTasks()), rootRequests: normalizeCount(resolved.getRootRequests()), @@ -130,6 +136,11 @@ export function createGatewayActiveWorkSnapshot( `${counts.pendingReplies} pending reply delivery operation(s)`, ); add(counts.embeddedRuns, "embedded-run", `${counts.embeddedRuns} active embedded run(s)`); + add( + counts.backgroundExecSessions, + "background-exec", + `${counts.backgroundExecSessions} active background exec session(s)`, + ); add(counts.cronRuns, "cron-run", `${counts.cronRuns} active cron run(s)`); add(counts.rootRequests, "root-request", `${counts.rootRequests} active gateway request(s)`); add( diff --git a/src/infra/gateway-suspend-coordinator.test.ts b/src/infra/gateway-suspend-coordinator.test.ts index a0b087a4ce8a..83641b6e85ea 100644 --- a/src/infra/gateway-suspend-coordinator.test.ts +++ b/src/infra/gateway-suspend-coordinator.test.ts @@ -1,5 +1,14 @@ // Covers atomic refuse-only suspension preparation, renewal, and release. import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + addSession, + deleteSession, + getActiveBackgroundExecSessionCount, + markBackgrounded, + markExited, + resetProcessRegistryForTests, +} from "../agents/bash-process-registry.js"; +import { createProcessSessionFixture } from "../agents/bash-process-registry.test-helpers.js"; import { isGatewayWorkAdmissionClosed, markGatewayRestartDraining, @@ -7,6 +16,7 @@ import { } from "../process/gateway-work-admission.js"; import type { GatewayActiveWorkInspectors } from "./gateway-active-work.js"; import { + GATEWAY_SUSPEND_RETRY_AFTER_MS, GATEWAY_SUSPEND_TTL_MS, getGatewaySuspendStatus, prepareGatewaySuspend, @@ -21,6 +31,7 @@ function inspectors( getQueueSize: () => 0, getPendingReplies: () => 0, getEmbeddedRuns: () => 0, + getBackgroundExecSessions: () => 0, getCronRuns: () => 0, getActiveTasks: () => 0, getTaskBlockers: () => [], @@ -36,11 +47,13 @@ function inspectors( } beforeEach(() => { + resetProcessRegistryForTests(); resetGatewaySuspendCoordinatorForTest(); resetGatewayWorkAdmission(); }); afterEach(() => { + resetProcessRegistryForTests(); resetGatewaySuspendCoordinatorForTest(); resetGatewayWorkAdmission(); }); @@ -85,6 +98,50 @@ describe("gateway suspend coordinator", () => { expect(isGatewayWorkAdmissionClosed()).toBe(false); }); + it("stays busy after a background session is hidden until its process exits", () => { + const session = createProcessSessionFixture({ + id: "private-background-session", + command: "private command", + }); + addSession(session); + markBackgrounded(session); + deleteSession(session.id); + + const inspect = inspectors({ + getBackgroundExecSessions: getActiveBackgroundExecSessionCount, + }); + const busy = prepareGatewaySuspend({ + requestId: "request-background-exec", + pauseScheduling: vi.fn(), + resumeScheduling: vi.fn(), + inspect, + }); + + expect(busy).toEqual({ + status: "busy", + reason: "active-work", + retryAfterMs: GATEWAY_SUSPEND_RETRY_AFTER_MS, + activeCount: 1, + blockers: [ + { + kind: "background-exec", + count: 1, + message: "1 active background exec session(s)", + }, + ], + }); + + markExited(session, 0, null, "completed"); + expect( + prepareGatewaySuspend({ + requestId: "request-background-exec", + pauseScheduling: vi.fn(), + resumeScheduling: vi.fn(), + inspect, + }), + ).toMatchObject({ status: "ready", activeCount: 0, blockers: [] }); + }); + it("keeps admission closed until a failed busy rollback resumes scheduling", () => { vi.useFakeTimers(); try { diff --git a/src/infra/restart-coordinator.test.ts b/src/infra/restart-coordinator.test.ts index f4a1d2c9ff19..f0556039c569 100644 --- a/src/infra/restart-coordinator.test.ts +++ b/src/infra/restart-coordinator.test.ts @@ -1,5 +1,9 @@ // Covers safe gateway restart preflight and requests. -import { describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + resetGatewayWorkAdmission, + tryBeginGatewayIndependentRootWorkAdmission, +} from "../process/gateway-work-admission.js"; import { createSafeGatewayRestartPreflight, requestSafeGatewayRestart, @@ -11,6 +15,14 @@ vi.mock("./restart.js", () => ({ scheduleGatewaySigusr1Restart: (opts: unknown) => scheduleGatewaySigusr1Restart(opts), })); +beforeEach(() => { + resetGatewayWorkAdmission(); +}); + +afterEach(() => { + resetGatewayWorkAdmission(); +}); + describe("safe gateway restart coordinator", () => { it("reports safe when no restart blockers are active", () => { const preflight = createSafeGatewayRestartPreflight({ @@ -18,6 +30,8 @@ describe("safe gateway restart coordinator", () => { getPendingReplies: () => 0, getEmbeddedRuns: () => 0, getCronRuns: () => 0, + getBackgroundExecSessions: () => 0, + getRootRequests: () => 0, getActiveTasks: () => 0, getTaskBlockers: () => [], }); @@ -29,6 +43,8 @@ describe("safe gateway restart coordinator", () => { pendingReplies: 0, embeddedRuns: 0, cronRuns: 0, + backgroundExecSessions: 0, + rootRequests: 0, activeTasks: 0, totalActive: 0, }, @@ -43,6 +59,8 @@ describe("safe gateway restart coordinator", () => { getPendingReplies: () => 1, getEmbeddedRuns: () => 1, getCronRuns: () => 1, + getBackgroundExecSessions: () => 0, + getRootRequests: () => 1, getActiveTasks: () => 1, getTaskBlockers: () => [ { @@ -57,18 +75,79 @@ describe("safe gateway restart coordinator", () => { }); expect(preflight.safe).toBe(false); - expect(preflight.counts.totalActive).toBe(6); + expect(preflight.counts.totalActive).toBe(7); expect(preflight.blockers.map((blocker) => blocker.kind)).toEqual([ "queue", "reply", "embedded-run", "cron-run", + "root-request", "task", ]); expect(preflight.summary).toContain("restart deferred"); expect(preflight.summary).toContain("taskId=task-1"); }); + it("defers restart for aggregate background exec sessions", () => { + const preflight = createSafeGatewayRestartPreflight({ + getQueueSize: () => 0, + getPendingReplies: () => 0, + getEmbeddedRuns: () => 0, + getCronRuns: () => 0, + getBackgroundExecSessions: () => 2, + getRootRequests: () => 0, + getActiveTasks: () => 0, + getTaskBlockers: () => [], + }); + + expect(preflight.safe).toBe(false); + expect(preflight.counts).toMatchObject({ + backgroundExecSessions: 2, + totalActive: 2, + }); + expect(preflight.blockers).toEqual([ + { + kind: "background-exec", + count: 2, + message: "2 active background exec session(s)", + }, + ]); + expect(preflight.summary).toBe("restart deferred: 2 active background exec session(s)"); + }); + + it("counts an admitted spawn handoff while excluding the preflight request", async () => { + const handoff = tryBeginGatewayIndependentRootWorkAdmission(); + const request = tryBeginGatewayIndependentRootWorkAdmission(); + expect(handoff).not.toBeNull(); + expect(request).not.toBeNull(); + + try { + await request?.run(async () => { + const preflight = createSafeGatewayRestartPreflight({ + getQueueSize: () => 0, + getPendingReplies: () => 0, + getEmbeddedRuns: () => 0, + getCronRuns: () => 0, + getBackgroundExecSessions: () => 0, + getActiveTasks: () => 0, + getTaskBlockers: () => [], + }); + + expect(preflight.counts).toMatchObject({ rootRequests: 1, totalActive: 1 }); + expect(preflight.blockers).toEqual([ + { + kind: "root-request", + count: 1, + message: "1 active gateway request(s)", + }, + ]); + }); + } finally { + request?.release(); + handoff?.release(); + } + }); + it("keeps truncated task titles on complete UTF-16 code points", () => { const preflight = createSafeGatewayRestartPreflight({ getQueueSize: () => 0, diff --git a/src/infra/restart-coordinator.ts b/src/infra/restart-coordinator.ts index d477406cc39d..4d75ef4a44f0 100644 --- a/src/infra/restart-coordinator.ts +++ b/src/infra/restart-coordinator.ts @@ -1,3 +1,4 @@ +import { getActiveGatewayRootWorkCount } from "../process/gateway-work-admission.js"; import { createGatewayActiveWorkSnapshot, type GatewayActiveWorkBlocker, @@ -12,11 +13,20 @@ export type SafeGatewayRestartCounts = { pendingReplies: number; embeddedRuns: number; cronRuns: number; + backgroundExecSessions: number; + rootRequests: number; activeTasks: number; totalActive: number; }; export type SafeGatewayRestartBlocker = Omit & { - kind: "queue" | "reply" | "embedded-run" | "cron-run" | "task"; + kind: + | "queue" + | "reply" + | "embedded-run" + | "cron-run" + | "background-exec" + | "root-request" + | "task"; }; type SafeRestartInspectors = Pick< @@ -25,6 +35,8 @@ type SafeRestartInspectors = Pick< | "getPendingReplies" | "getEmbeddedRuns" | "getCronRuns" + | "getBackgroundExecSessions" + | "getRootRequests" | "getActiveTasks" | "getTaskBlockers" >; @@ -48,7 +60,10 @@ export function createSafeGatewayRestartPreflight( ): SafeGatewayRestartPreflight { const snapshot = createGatewayActiveWorkSnapshot({ ...inspectors, - getRootRequests: () => 0, + // Restart RPC preflight itself owns a root. Count every other admitted + // handoff so signal emission cannot split spawn from durable ownership. + getRootRequests: + inspectors.getRootRequests ?? (() => getActiveGatewayRootWorkCount({ excludeCurrent: true })), getSessionAdmissions: () => 0, getSessionMutations: () => 0, getChatRuns: () => 0, @@ -61,12 +76,16 @@ export function createSafeGatewayRestartPreflight( pendingReplies: snapshot.counts.pendingReplies, embeddedRuns: snapshot.counts.embeddedRuns, cronRuns: snapshot.counts.cronRuns, + backgroundExecSessions: snapshot.counts.backgroundExecSessions, + rootRequests: snapshot.counts.rootRequests, activeTasks: snapshot.counts.activeTasks, totalActive: snapshot.counts.queueSize + snapshot.counts.pendingReplies + snapshot.counts.embeddedRuns + snapshot.counts.cronRuns + + snapshot.counts.backgroundExecSessions + + snapshot.counts.rootRequests + snapshot.counts.activeTasks, }; const blockers = snapshot.blockers as SafeGatewayRestartBlocker[]; @@ -83,7 +102,7 @@ export function createSafeGatewayRestartPreflight( }; } -/** Schedule a gateway restart after collecting queue/reply/task blockers. */ +/** Schedule a gateway restart after collecting tracked active-work blockers. */ export function requestSafeGatewayRestart( opts: { reason?: string; diff --git a/src/infra/restart-suspension.test.ts b/src/infra/restart-suspension.test.ts index ee8012a198ec..6ed07bd34171 100644 --- a/src/infra/restart-suspension.test.ts +++ b/src/infra/restart-suspension.test.ts @@ -24,6 +24,7 @@ function inspectors(): GatewayActiveWorkInspectors { getQueueSize: () => 0, getPendingReplies: () => 0, getEmbeddedRuns: () => 0, + getBackgroundExecSessions: () => 0, getCronRuns: () => 0, getActiveTasks: () => 0, getTaskBlockers: () => [],