diff --git a/src/acp/control-plane/manager.cancel-session.test.ts b/src/acp/control-plane/manager.cancel-session.test.ts new file mode 100644 index 000000000000..3657b39087fc --- /dev/null +++ b/src/acp/control-plane/manager.cancel-session.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it, vi } from "vitest"; +import { + AcpSessionManager, + baseCfg, + createRuntime, + expectRecordFields, + extractStatesFromUpserts, + hoisted, + installAcpSessionManagerTestLifecycle, + mockCallArg, + readySessionMeta, +} from "./manager.test-helpers.js"; + +describe("AcpSessionManager cancelSession", () => { + installAcpSessionManagerTestLifecycle(); + + it("preempts an active turn on cancel and returns to idle state", async () => { + const runtimeState = createRuntime(); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + hoisted.readAcpSessionEntryMock.mockReturnValue({ + sessionKey: "agent:codex:acp:session-1", + storeSessionKey: "agent:codex:acp:session-1", + acp: readySessionMeta(), + }); + + let enteredRun = false; + runtimeState.runTurn.mockImplementation(async function* (input: { signal?: AbortSignal }) { + enteredRun = true; + await new Promise((resolve) => { + if (input.signal?.aborted) { + resolve(); + return; + } + input.signal?.addEventListener("abort", () => resolve(), { once: true }); + }); + yield { type: "done" as const, stopReason: "cancel" }; + }); + + const manager = new AcpSessionManager(); + const runPromise = manager.runTurn({ + cfg: baseCfg, + sessionKey: "agent:codex:acp:session-1", + text: "long task", + mode: "prompt", + requestId: "run-1", + }); + await vi.waitFor( + () => { + expect(enteredRun).toBe(true); + }, + { interval: 1 }, + ); + + await manager.cancelSession({ + cfg: baseCfg, + sessionKey: "agent:codex:acp:session-1", + reason: "manual-cancel", + }); + await runPromise; + + expect(runtimeState.cancel).toHaveBeenCalledTimes(1); + expectRecordFields(mockCallArg(runtimeState.cancel), { + reason: "manual-cancel", + }); + const states = extractStatesFromUpserts(); + expect(states).toContain("running"); + expect(states).toContain("idle"); + expect(states).not.toContain("error"); + }); +}); diff --git a/src/acp/control-plane/manager.cancel-session.ts b/src/acp/control-plane/manager.cancel-session.ts new file mode 100644 index 000000000000..f4aeaccad8c9 --- /dev/null +++ b/src/acp/control-plane/manager.cancel-session.ts @@ -0,0 +1,113 @@ +import type { AcpRuntime, AcpRuntimeHandle } from "@openclaw/acp-core/runtime/types"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { + type AcpRuntimeError, + toAcpRuntimeError, + withAcpRuntimeErrorBoundary, +} from "../runtime/errors.js"; +import type { + ActiveTurnState, + EnsureManagerRuntimeHandle, + ResolveManagerSession, + SetManagerSessionState, + WithManagerSessionActor, +} from "./manager.types.js"; +import { normalizeActorKey, requireReadySessionMeta } from "./manager.utils.js"; + +export async function runManagerCancelSession(params: { + cfg: OpenClawConfig; + sessionKey: string; + reason?: string; + activeTurnBySession: Map; + withSessionActor: WithManagerSessionActor; + resolveSession: ResolveManagerSession; + ensureRuntimeHandle: EnsureManagerRuntimeHandle; + setSessionState: SetManagerSessionState; +}): Promise { + const actorKey = normalizeActorKey(params.sessionKey); + const activeTurn = params.activeTurnBySession.get(actorKey); + if (activeTurn) { + await cancelActiveTurn({ + activeTurn, + reason: params.reason, + }); + return; + } + + await params.withSessionActor(params.sessionKey, async () => { + const resolution = params.resolveSession({ + cfg: params.cfg, + sessionKey: params.sessionKey, + }); + const resolvedMeta = requireReadySessionMeta(resolution); + const { runtime, handle } = await params.ensureRuntimeHandle({ + cfg: params.cfg, + sessionKey: params.sessionKey, + meta: resolvedMeta, + }); + try { + await cancelRuntimeHandle({ + runtime, + handle, + reason: params.reason, + }); + await params.setSessionState({ + cfg: params.cfg, + sessionKey: params.sessionKey, + state: "idle", + clearLastError: true, + }); + } catch (error) { + const acpError = normalizeCancelError(error); + await params.setSessionState({ + cfg: params.cfg, + sessionKey: params.sessionKey, + state: "error", + lastError: acpError.message, + }); + throw acpError; + } + }); +} + +async function cancelActiveTurn(params: { + activeTurn: ActiveTurnState; + reason?: string; +}): Promise { + params.activeTurn.abortController.abort(); + if (!params.activeTurn.cancelPromise) { + params.activeTurn.cancelPromise = params.activeTurn.runtime.cancel({ + handle: params.activeTurn.handle, + reason: params.reason, + }); + } + await withAcpRuntimeErrorBoundary({ + run: async () => await params.activeTurn.cancelPromise!, + fallbackCode: "ACP_TURN_FAILED", + fallbackMessage: "ACP cancel failed before completion.", + }); +} + +async function cancelRuntimeHandle(params: { + runtime: AcpRuntime; + handle: AcpRuntimeHandle; + reason?: string; +}): Promise { + await withAcpRuntimeErrorBoundary({ + run: async () => + await params.runtime.cancel({ + handle: params.handle, + reason: params.reason, + }), + fallbackCode: "ACP_TURN_FAILED", + fallbackMessage: "ACP cancel failed before completion.", + }); +} + +function normalizeCancelError(error: unknown): AcpRuntimeError { + return toAcpRuntimeError({ + error, + fallbackCode: "ACP_TURN_FAILED", + fallbackMessage: "ACP cancel failed before completion.", + }); +} diff --git a/src/acp/control-plane/manager.close-session.ts b/src/acp/control-plane/manager.close-session.ts index 71b8b46efd86..0040712110a7 100644 --- a/src/acp/control-plane/manager.close-session.ts +++ b/src/acp/control-plane/manager.close-session.ts @@ -2,8 +2,6 @@ import { identityHasStableSessionId, resolveSessionIdentityFromMeta, } from "@openclaw/acp-core/runtime/session-identity"; -import type { AcpRuntime, AcpRuntimeHandle } from "@openclaw/acp-core/runtime/types"; -import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { toAcpRuntimeError, withAcpRuntimeErrorBoundary } from "../runtime/errors.js"; import type { ManagerRuntimeHandleCache } from "./manager.runtime-handle-cache.js"; import { @@ -15,25 +13,19 @@ import type { AcpCloseSessionInput, AcpCloseSessionResult, AcpSessionManagerDeps, - AcpSessionResolution, - SessionAcpMeta, + EnsureManagerRuntimeHandle, + ResolveManagerSession, WriteManagerSessionMeta, } from "./manager.types.js"; import { requireReadySessionMeta, resolveAcpSessionResolutionError } from "./manager.utils.js"; -type EnsureRuntimeHandle = (params: { - cfg: OpenClawConfig; - sessionKey: string; - meta: SessionAcpMeta; -}) => Promise<{ runtime: AcpRuntime; handle: AcpRuntimeHandle; meta: SessionAcpMeta }>; - export async function runManagerCloseSession(params: { input: AcpCloseSessionInput; sessionKey: string; deps: Pick; runtimeHandles: ManagerRuntimeHandleCache; - resolveSession: (params: { cfg: OpenClawConfig; sessionKey: string }) => AcpSessionResolution; - ensureRuntimeHandle: EnsureRuntimeHandle; + resolveSession: ResolveManagerSession; + ensureRuntimeHandle: EnsureManagerRuntimeHandle; writeSessionMeta: WriteManagerSessionMeta; }): Promise { const { input, sessionKey } = params; diff --git a/src/acp/control-plane/manager.core.ts b/src/acp/control-plane/manager.core.ts index 34ed7894d070..6fd31a9145e9 100644 --- a/src/acp/control-plane/manager.core.ts +++ b/src/acp/control-plane/manager.core.ts @@ -1,28 +1,17 @@ -import { - createIdentityFromEnsure, - identityHasStableSessionId, - isSessionIdentityPending, - mergeSessionIdentity, - resolveSessionIdentityFromMeta, -} from "@openclaw/acp-core/runtime/session-identity"; import type { AcpRuntime, AcpRuntimeCapabilities, AcpRuntimeHandle, AcpRuntimeStatus, } from "@openclaw/acp-core/runtime/types"; -import { resolveRuntimeConfigCacheKey } from "../../config/runtime-snapshot.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { logVerbose } from "../../globals.js"; -import { normalizeAgentId } from "../../routing/session-key.js"; import { isAcpSessionKey } from "../../sessions/session-key-utils.js"; -import { - AcpRuntimeError, - toAcpRuntimeError, - withAcpRuntimeErrorBoundary, -} from "../runtime/errors.js"; +import { AcpRuntimeError } from "../runtime/errors.js"; +import { runManagerCancelSession } from "./manager.cancel-session.js"; import { runManagerCloseSession } from "./manager.close-session.js"; import { reconcileManagerRuntimeSessionIdentifiers } from "./manager.identity-reconcile.js"; +import { runManagerInitializeSession } from "./manager.initialize-session.js"; import { applyManagerRuntimeControls, resolveManagerRuntimeCapabilities, @@ -36,6 +25,8 @@ import { runUpdateManagerSessionRuntimeOptions, type RuntimeOptionCommandServices, } from "./manager.runtime-options-commands.js"; +import { runManagerStartupIdentityReconcile } from "./manager.startup-identity-reconcile.js"; +import { runManagerGetSessionStatus } from "./manager.status.js"; import { runManagerTurn } from "./manager.turn-runner.js"; import { type AcpCloseSessionInput, @@ -58,13 +49,10 @@ import { canonicalizeAcpSessionKey, normalizeAcpErrorCode, normalizeActorKey, - requireReadySessionMeta, resolveMissingMetaError, } from "./manager.utils.js"; import { - normalizeRuntimeOptions, normalizeText, - resolveRuntimeOptionsFromMeta, validateRuntimeConfigOptionInput, validateRuntimeModeInput, validateRuntimeOptionPatch, @@ -144,69 +132,14 @@ export class AcpSessionManager { async reconcilePendingSessionIdentities(params: { cfg: OpenClawConfig; }): Promise { - let checked = 0; - let resolved = 0; - let failed = 0; - - let acpSessions: Awaited>; - try { - acpSessions = await this.deps.listAcpSessions({ - cfg: params.cfg, - }); - } catch (error) { - logVerbose(`acp-manager: startup identity scan failed: ${String(error)}`); - return { checked, resolved, failed: failed + 1 }; - } - - for (const session of acpSessions) { - if (!session.acp || !session.sessionKey) { - continue; - } - const currentIdentity = resolveSessionIdentityFromMeta(session.acp); - if ( - !isSessionIdentityPending(currentIdentity) || - !identityHasStableSessionId(currentIdentity) - ) { - continue; - } - - checked += 1; - try { - const becameResolved = await this.withSessionActor(session.sessionKey, async () => { - const resolution = this.resolveSession({ - cfg: params.cfg, - sessionKey: session.sessionKey, - }); - if (resolution.kind !== "ready") { - return false; - } - const { runtime, handle, meta } = await this.ensureRuntimeHandle({ - cfg: params.cfg, - sessionKey: session.sessionKey, - meta: resolution.meta, - }); - const reconciled = await this.reconcileRuntimeSessionIdentifiers({ - cfg: params.cfg, - sessionKey: session.sessionKey, - runtime, - handle, - meta, - failOnStatusError: false, - }); - return !isSessionIdentityPending(resolveSessionIdentityFromMeta(reconciled.meta)); - }); - if (becameResolved) { - resolved += 1; - } - } catch (error) { - failed += 1; - logVerbose( - `acp-manager: startup identity reconcile failed for ${session.sessionKey}: ${String(error)}`, - ); - } - } - - return { checked, resolved, failed }; + return await runManagerStartupIdentityReconcile({ + cfg: params.cfg, + deps: this.deps, + withSessionActor: this.withSessionActor.bind(this), + resolveSession: this.resolveSession.bind(this), + ensureRuntimeHandle: this.ensureRuntimeHandle.bind(this), + reconcileRuntimeSessionIdentifiers: this.reconcileRuntimeSessionIdentifiers.bind(this), + }); } async initializeSession(input: AcpInitializeSessionInput): Promise<{ @@ -221,124 +154,16 @@ export class AcpSessionManager { if (!sessionKey) { throw new AcpRuntimeError("ACP_SESSION_INIT_FAILED", "ACP session key is required."); } - const agent = normalizeAgentId(input.agent); await this.evictIdleRuntimeHandles(input.cfg); return await this.withSessionActor(sessionKey, async () => { - const backend = this.deps.requireRuntimeBackend(input.backendId || input.cfg.acp?.backend); - const runtime = backend.runtime; - const initialRuntimeOptions = validateRuntimeOptionPatch({ - ...input.runtimeOptions, - ...(input.cwd !== undefined ? { cwd: input.cwd } : {}), - }); - const requestedCwd = initialRuntimeOptions.cwd; - const requestedModel = initialRuntimeOptions.model; - const requestedThinking = initialRuntimeOptions.thinking; - this.enforceConcurrentSessionLimit({ - cfg: input.cfg, + return await runManagerInitializeSession({ + input, sessionKey, + deps: this.deps, + runtimeHandles: this.runtimeHandles, + enforceConcurrentSessionLimit: this.enforceConcurrentSessionLimit.bind(this), + writeSessionMeta: this.writeSessionMeta.bind(this), }); - const handle = await withAcpRuntimeErrorBoundary({ - run: async () => - await runtime.ensureSession({ - sessionKey, - agent, - mode: input.mode, - resumeSessionId: input.resumeSessionId, - ...(requestedModel ? { model: requestedModel } : {}), - ...(requestedThinking ? { thinking: requestedThinking } : {}), - cwd: requestedCwd, - }), - fallbackCode: "ACP_SESSION_INIT_FAILED", - fallbackMessage: "Could not initialize ACP session runtime.", - }); - const effectiveCwd = normalizeText(handle.cwd) ?? requestedCwd; - const effectiveRuntimeOptions = normalizeRuntimeOptions({ - ...initialRuntimeOptions, - ...(effectiveCwd ? { cwd: effectiveCwd } : {}), - }); - - const identityNow = Date.now(); - const initializedIdentity = - mergeSessionIdentity({ - current: undefined, - incoming: createIdentityFromEnsure({ - handle, - now: identityNow, - }), - now: identityNow, - }) ?? - ({ - state: "pending", - source: "ensure", - lastUpdatedAt: identityNow, - } as const); - const meta: SessionAcpMeta = { - backend: handle.backend || backend.id, - agent, - runtimeSessionName: handle.runtimeSessionName, - identity: initializedIdentity, - mode: input.mode, - ...(Object.keys(effectiveRuntimeOptions).length > 0 - ? { runtimeOptions: effectiveRuntimeOptions } - : {}), - cwd: effectiveCwd, - state: "idle", - lastActivityAt: Date.now(), - }; - - let persisted: SessionEntry | null = null; - try { - persisted = await this.writeSessionMeta({ - cfg: input.cfg, - sessionKey, - mutate: () => meta, - failOnError: true, - }); - } catch (error) { - await runtime - .close({ - handle, - reason: "init-meta-failed", - }) - .catch((closeError) => { - logVerbose( - `acp-manager: cleanup close failed after metadata write error for ${sessionKey}: ${String(closeError)}`, - ); - }); - throw error; - } - - if (!persisted?.acp) { - await runtime - .close({ - handle, - reason: "init-meta-failed", - }) - .catch((closeError) => { - logVerbose( - `acp-manager: cleanup close failed after metadata write error for ${sessionKey}: ${String(closeError)}`, - ); - }); - - throw new AcpRuntimeError( - "ACP_SESSION_INIT_FAILED", - `Could not persist ACP metadata for ${sessionKey}.`, - ); - } - this.runtimeHandles.set(sessionKey, { - runtime, - handle, - backend: handle.backend || backend.id, - agent, - mode: input.mode, - cwd: effectiveCwd, - configSignature: resolveRuntimeConfigCacheKey(input.cfg), - }); - return { - runtime, - handle, - meta, - }; }); } @@ -355,65 +180,17 @@ export class AcpSessionManager { await this.evictIdleRuntimeHandles(params.cfg); return await this.withSessionActor( sessionKey, - async () => { - this.throwIfAborted(params.signal); - const resolution = this.resolveSession({ + async () => + await runManagerGetSessionStatus({ cfg: params.cfg, sessionKey, - }); - const resolvedMeta = requireReadySessionMeta(resolution); - const { - runtime, - handle: ensuredHandle, - meta: ensuredMeta, - } = await this.ensureRuntimeHandle({ - cfg: params.cfg, - sessionKey, - meta: resolvedMeta, - }); - let handle = ensuredHandle; - let meta = ensuredMeta; - const capabilities = await this.resolveRuntimeCapabilities({ runtime, handle }); - let runtimeStatus: AcpRuntimeStatus | undefined; - if (runtime.getStatus) { - runtimeStatus = await withAcpRuntimeErrorBoundary({ - run: async () => { - this.throwIfAborted(params.signal); - const status = await runtime.getStatus!({ - handle, - ...(params.signal ? { signal: params.signal } : {}), - }); - this.throwIfAborted(params.signal); - return status; - }, - fallbackCode: "ACP_TURN_FAILED", - fallbackMessage: "Could not read ACP runtime status.", - }); - } - ({ handle, meta, runtimeStatus } = await this.reconcileRuntimeSessionIdentifiers({ - cfg: params.cfg, - sessionKey, - runtime, - handle, - meta, - runtimeStatus, - failOnStatusError: true, - })); - const identity = resolveSessionIdentityFromMeta(meta); - return { - sessionKey, - backend: handle.backend || meta.backend, - agent: meta.agent, - ...(identity ? { identity } : {}), - state: meta.state, - mode: meta.mode, - runtimeOptions: resolveRuntimeOptionsFromMeta(meta), - capabilities, - runtimeStatus, - lastActivityAt: meta.lastActivityAt, - lastError: meta.lastError, - }; - }, + signal: params.signal, + throwIfAborted: this.throwIfAborted.bind(this), + resolveSession: this.resolveSession.bind(this), + ensureRuntimeHandle: this.ensureRuntimeHandle.bind(this), + resolveRuntimeCapabilities: this.resolveRuntimeCapabilities.bind(this), + reconcileRuntimeSessionIdentifiers: this.reconcileRuntimeSessionIdentifiers.bind(this), + }), params.signal, ); } @@ -546,65 +323,15 @@ export class AcpSessionManager { throw new AcpRuntimeError("ACP_SESSION_INIT_FAILED", "ACP session key is required."); } await this.evictIdleRuntimeHandles(params.cfg); - const actorKey = normalizeActorKey(sessionKey); - const activeTurn = this.activeTurnBySession.get(actorKey); - if (activeTurn) { - activeTurn.abortController.abort(); - if (!activeTurn.cancelPromise) { - activeTurn.cancelPromise = activeTurn.runtime.cancel({ - handle: activeTurn.handle, - reason: params.reason, - }); - } - await withAcpRuntimeErrorBoundary({ - run: async () => await activeTurn.cancelPromise!, - fallbackCode: "ACP_TURN_FAILED", - fallbackMessage: "ACP cancel failed before completion.", - }); - return; - } - - await this.withSessionActor(sessionKey, async () => { - const resolution = this.resolveSession({ - cfg: params.cfg, - sessionKey, - }); - const resolvedMeta = requireReadySessionMeta(resolution); - const { runtime, handle } = await this.ensureRuntimeHandle({ - cfg: params.cfg, - sessionKey, - meta: resolvedMeta, - }); - try { - await withAcpRuntimeErrorBoundary({ - run: async () => - await runtime.cancel({ - handle, - reason: params.reason, - }), - fallbackCode: "ACP_TURN_FAILED", - fallbackMessage: "ACP cancel failed before completion.", - }); - await this.setSessionState({ - cfg: params.cfg, - sessionKey, - state: "idle", - clearLastError: true, - }); - } catch (error) { - const acpError = toAcpRuntimeError({ - error, - fallbackCode: "ACP_TURN_FAILED", - fallbackMessage: "ACP cancel failed before completion.", - }); - await this.setSessionState({ - cfg: params.cfg, - sessionKey, - state: "error", - lastError: acpError.message, - }); - throw acpError; - } + await runManagerCancelSession({ + cfg: params.cfg, + sessionKey, + reason: params.reason, + activeTurnBySession: this.activeTurnBySession, + withSessionActor: this.withSessionActor.bind(this), + resolveSession: this.resolveSession.bind(this), + ensureRuntimeHandle: this.ensureRuntimeHandle.bind(this), + setSessionState: this.setSessionState.bind(this), }); } diff --git a/src/acp/control-plane/manager.initialize-session.test.ts b/src/acp/control-plane/manager.initialize-session.test.ts new file mode 100644 index 000000000000..799360e88ad6 --- /dev/null +++ b/src/acp/control-plane/manager.initialize-session.test.ts @@ -0,0 +1,166 @@ +import { describe, expect, it } from "vitest"; +import { + AcpSessionManager, + baseCfg, + createRuntime, + expectRecordFields, + expectRejectedRecord, + extractRuntimeOptionsFromUpserts, + hoisted, + installAcpSessionManagerTestLifecycle, + mockCallArg, + readySessionMeta, + type OpenClawConfig, +} from "./manager.test-helpers.js"; + +describe("AcpSessionManager initializeSession", () => { + installAcpSessionManagerTestLifecycle(); + + it("enforces acp.maxConcurrentSessions during initializeSession", async () => { + const runtimeState = createRuntime(); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + hoisted.upsertAcpSessionMetaMock.mockResolvedValue({ + sessionKey: "agent:codex:acp:session-a", + storeSessionKey: "agent:codex:acp:session-a", + acp: readySessionMeta(), + }); + const limitedCfg = { + acp: { + ...baseCfg.acp, + maxConcurrentSessions: 1, + }, + } as OpenClawConfig; + + const manager = new AcpSessionManager(); + await manager.initializeSession({ + cfg: limitedCfg, + sessionKey: "agent:codex:acp:session-a", + agent: "codex", + mode: "persistent", + }); + + await expectRejectedRecord( + manager.initializeSession({ + cfg: limitedCfg, + sessionKey: "agent:codex:acp:session-b", + agent: "codex", + mode: "persistent", + }), + { + code: "ACP_SESSION_INIT_FAILED", + message: "ACP max concurrent sessions reached (1/1).", + }, + ); + expect(runtimeState.ensureSession).toHaveBeenCalledTimes(1); + }); + + it("persists runtime options provided during initializeSession", async () => { + const runtimeState = createRuntime(); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + hoisted.upsertAcpSessionMetaMock.mockResolvedValue({ + sessionKey: "agent:codex:acp:session-a", + storeSessionKey: "agent:codex:acp:session-a", + acp: readySessionMeta({ + runtimeOptions: { + model: "openai/gpt-5.4", + thinking: "high", + }, + }), + }); + + const manager = new AcpSessionManager(); + await manager.initializeSession({ + cfg: baseCfg, + sessionKey: "agent:codex:acp:session-a", + agent: "codex", + mode: "persistent", + runtimeOptions: { + model: "openai/gpt-5.4", + thinking: "high", + }, + }); + + expect(extractRuntimeOptionsFromUpserts()).toEqual([ + { + model: "openai/gpt-5.4", + thinking: "high", + }, + ]); + expectRecordFields(mockCallArg(runtimeState.ensureSession), { + sessionKey: "agent:codex:acp:session-a", + model: "openai/gpt-5.4", + thinking: "high", + }); + }); + + it("preserves runtimeOptions cwd when initializeSession cwd is omitted", async () => { + const runtimeState = createRuntime(); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + hoisted.upsertAcpSessionMetaMock.mockResolvedValue({ + sessionKey: "agent:codex:acp:session-cwd-runtime-options", + storeSessionKey: "agent:codex:acp:session-cwd-runtime-options", + acp: readySessionMeta({ + runtimeOptions: { + cwd: "/workspace/from-runtime-options", + }, + cwd: "/workspace/from-runtime-options", + }), + }); + + const manager = new AcpSessionManager(); + await manager.initializeSession({ + cfg: baseCfg, + sessionKey: "agent:codex:acp:session-cwd-runtime-options", + agent: "codex", + mode: "persistent", + runtimeOptions: { + cwd: "/workspace/from-runtime-options", + }, + }); + + expectRecordFields(mockCallArg(runtimeState.ensureSession), { + sessionKey: "agent:codex:acp:session-cwd-runtime-options", + cwd: "/workspace/from-runtime-options", + }); + expect(extractRuntimeOptionsFromUpserts()).toEqual([ + { + cwd: "/workspace/from-runtime-options", + }, + ]); + }); + + it("rolls back ensured runtime sessions when metadata persistence fails", async () => { + const runtimeState = createRuntime(); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + hoisted.upsertAcpSessionMetaMock.mockRejectedValueOnce(new Error("disk full")); + + const manager = new AcpSessionManager(); + await expect( + manager.initializeSession({ + cfg: baseCfg, + sessionKey: "agent:codex:acp:session-1", + agent: "codex", + mode: "persistent", + }), + ).rejects.toThrow("disk full"); + const closeInput = mockCallArg(runtimeState.close); + expectRecordFields(closeInput, { + reason: "init-meta-failed", + }); + expectRecordFields(closeInput.handle, { + sessionKey: "agent:codex:acp:session-1", + }); + }); +}); diff --git a/src/acp/control-plane/manager.initialize-session.ts b/src/acp/control-plane/manager.initialize-session.ts new file mode 100644 index 000000000000..2bd5cc0ceeaf --- /dev/null +++ b/src/acp/control-plane/manager.initialize-session.ts @@ -0,0 +1,173 @@ +import { + createIdentityFromEnsure, + mergeSessionIdentity, +} from "@openclaw/acp-core/runtime/session-identity"; +import type { AcpRuntime, AcpRuntimeHandle } from "@openclaw/acp-core/runtime/types"; +import { resolveRuntimeConfigCacheKey } from "../../config/runtime-snapshot.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { logVerbose } from "../../globals.js"; +import { normalizeAgentId } from "../../routing/session-key.js"; +import { AcpRuntimeError, withAcpRuntimeErrorBoundary } from "../runtime/errors.js"; +import type { ManagerRuntimeHandleCache } from "./manager.runtime-handle-cache.js"; +import type { + AcpInitializeSessionInput, + AcpSessionManagerDeps, + SessionAcpMeta, + SessionEntry, + WriteManagerSessionMeta, +} from "./manager.types.js"; +import { + normalizeRuntimeOptions, + normalizeText, + validateRuntimeOptionPatch, +} from "./runtime-options.js"; + +export async function runManagerInitializeSession(params: { + input: AcpInitializeSessionInput; + sessionKey: string; + deps: Pick; + runtimeHandles: ManagerRuntimeHandleCache; + enforceConcurrentSessionLimit: (params: { cfg: OpenClawConfig; sessionKey: string }) => void; + writeSessionMeta: WriteManagerSessionMeta; +}): Promise<{ + runtime: AcpRuntime; + handle: AcpRuntimeHandle; + meta: SessionAcpMeta; +}> { + const { input, sessionKey } = params; + const backend = params.deps.requireRuntimeBackend(input.backendId || input.cfg.acp?.backend); + const runtime = backend.runtime; + const agent = normalizeAgentId(input.agent); + const initialRuntimeOptions = validateRuntimeOptionPatch({ + ...input.runtimeOptions, + ...(input.cwd !== undefined ? { cwd: input.cwd } : {}), + }); + const requestedCwd = initialRuntimeOptions.cwd; + const requestedModel = initialRuntimeOptions.model; + const requestedThinking = initialRuntimeOptions.thinking; + params.enforceConcurrentSessionLimit({ + cfg: input.cfg, + sessionKey, + }); + const handle = await withAcpRuntimeErrorBoundary({ + run: async () => + await runtime.ensureSession({ + sessionKey, + agent, + mode: input.mode, + resumeSessionId: input.resumeSessionId, + ...(requestedModel ? { model: requestedModel } : {}), + ...(requestedThinking ? { thinking: requestedThinking } : {}), + cwd: requestedCwd, + }), + fallbackCode: "ACP_SESSION_INIT_FAILED", + fallbackMessage: "Could not initialize ACP session runtime.", + }); + const effectiveCwd = normalizeText(handle.cwd) ?? requestedCwd; + const effectiveRuntimeOptions = normalizeRuntimeOptions({ + ...initialRuntimeOptions, + ...(effectiveCwd ? { cwd: effectiveCwd } : {}), + }); + + const identityNow = Date.now(); + const initializedIdentity = + mergeSessionIdentity({ + current: undefined, + incoming: createIdentityFromEnsure({ + handle, + now: identityNow, + }), + now: identityNow, + }) ?? + ({ + state: "pending", + source: "ensure", + lastUpdatedAt: identityNow, + } as const); + const meta: SessionAcpMeta = { + backend: handle.backend || backend.id, + agent, + runtimeSessionName: handle.runtimeSessionName, + identity: initializedIdentity, + mode: input.mode, + ...(Object.keys(effectiveRuntimeOptions).length > 0 + ? { runtimeOptions: effectiveRuntimeOptions } + : {}), + cwd: effectiveCwd, + state: "idle", + lastActivityAt: Date.now(), + }; + + const persisted = await persistInitializedSessionMeta({ + cfg: input.cfg, + sessionKey, + meta, + runtime, + handle, + writeSessionMeta: params.writeSessionMeta, + }); + if (!persisted?.acp) { + throw new AcpRuntimeError( + "ACP_SESSION_INIT_FAILED", + `Could not persist ACP metadata for ${sessionKey}.`, + ); + } + params.runtimeHandles.set(sessionKey, { + runtime, + handle, + backend: handle.backend || backend.id, + agent, + mode: input.mode, + cwd: effectiveCwd, + configSignature: resolveRuntimeConfigCacheKey(input.cfg), + }); + return { + runtime, + handle, + meta, + }; +} + +async function persistInitializedSessionMeta(params: { + cfg: OpenClawConfig; + sessionKey: string; + meta: SessionAcpMeta; + runtime: AcpRuntime; + handle: AcpRuntimeHandle; + writeSessionMeta: WriteManagerSessionMeta; +}): Promise { + try { + const persisted = await params.writeSessionMeta({ + cfg: params.cfg, + sessionKey: params.sessionKey, + mutate: () => params.meta, + failOnError: true, + }); + if (persisted?.acp) { + return persisted; + } + } catch (error) { + await closeRuntimeAfterInitMetaFailure(params); + throw error; + } + + await closeRuntimeAfterInitMetaFailure(params); + return null; +} + +async function closeRuntimeAfterInitMetaFailure(params: { + sessionKey: string; + runtime: AcpRuntime; + handle: AcpRuntimeHandle; +}): Promise { + await params.runtime + .close({ + handle: params.handle, + reason: "init-meta-failed", + }) + .catch((closeError) => { + logVerbose( + `acp-manager: cleanup close failed after metadata write error for ${params.sessionKey}: ${String(closeError)}`, + ); + }); +} diff --git a/src/acp/control-plane/manager.runtime-config.test.ts b/src/acp/control-plane/manager.runtime-config.test.ts index 4056305abefc..8885f1ba8802 100644 --- a/src/acp/control-plane/manager.runtime-config.test.ts +++ b/src/acp/control-plane/manager.runtime-config.test.ts @@ -268,131 +268,6 @@ describe("AcpSessionManager runtime config", () => { }); }); - it("reconciles pending ACP identities during startup scan", async () => { - const runtimeState = createRuntime(); - runtimeState.getStatus.mockResolvedValue({ - summary: "status=alive", - acpxRecordId: "acpx-record-1", - backendSessionId: "acpx-session-1", - agentSessionId: "agent-session-1", - details: { status: "alive" }, - }); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - - let currentMeta: SessionAcpMeta = { - ...readySessionMeta(), - identity: { - state: "pending", - source: "ensure", - acpxSessionId: "acpx-stale", - lastUpdatedAt: Date.now(), - }, - }; - const sessionKey = "agent:codex:acp:session-1"; - hoisted.listAcpSessionEntriesMock.mockResolvedValue([ - { - cfg: baseCfg, - storePath: "/tmp/sessions-acp.json", - sessionKey, - storeSessionKey: sessionKey, - entry: { - sessionId: "session-1", - updatedAt: Date.now(), - acp: currentMeta, - }, - acp: currentMeta, - }, - ]); - hoisted.readAcpSessionEntryMock.mockImplementation((paramsUnknown: unknown) => { - const key = (paramsUnknown as { sessionKey?: string }).sessionKey ?? sessionKey; - return { - sessionKey: key, - storeSessionKey: key, - acp: currentMeta, - }; - }); - hoisted.upsertAcpSessionMetaMock.mockImplementation(async (paramsUnknown: unknown) => { - const params = paramsUnknown as { - mutate: ( - current: SessionAcpMeta | undefined, - entry: { acp?: SessionAcpMeta } | undefined, - ) => SessionAcpMeta | null | undefined; - }; - const next = params.mutate(currentMeta, { acp: currentMeta }); - if (next) { - currentMeta = next; - } - return { - sessionId: "session-1", - updatedAt: Date.now(), - acp: currentMeta, - }; - }); - - const manager = new AcpSessionManager(); - const result = await manager.reconcilePendingSessionIdentities({ cfg: baseCfg }); - - expect(result).toEqual({ checked: 1, resolved: 1, failed: 0 }); - expect(currentMeta.identity?.state).toBe("resolved"); - expect(currentMeta.identity?.acpxRecordId).toBe("acpx-record-1"); - expect(currentMeta.identity?.acpxSessionId).toBe("acpx-session-1"); - expect(currentMeta.identity?.agentSessionId).toBe("agent-session-1"); - }); - - it("skips startup reconcile for pending identities without stable runtime ids", async () => { - const runtimeState = createRuntime(); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - - const sessionKey = "agent:claude:acp:binding:discord:default:9373ab192b2317f4"; - hoisted.listAcpSessionEntriesMock.mockResolvedValue([ - { - cfg: baseCfg, - storePath: "/tmp/sessions-acp.json", - sessionKey, - storeSessionKey: sessionKey, - entry: { - sessionId: "session-1", - updatedAt: Date.now(), - acp: { - ...readySessionMeta({ - agent: "claude", - }), - identity: { - state: "pending", - acpxRecordId: sessionKey, - source: "status", - lastUpdatedAt: Date.now(), - }, - }, - }, - acp: { - ...readySessionMeta({ - agent: "claude", - }), - identity: { - state: "pending", - acpxRecordId: sessionKey, - source: "status", - lastUpdatedAt: Date.now(), - }, - }, - }, - ]); - - const manager = new AcpSessionManager(); - const result = await manager.reconcilePendingSessionIdentities({ cfg: baseCfg }); - - expect(result).toEqual({ checked: 0, resolved: 0, failed: 0 }); - expect(runtimeState.ensureSession).not.toHaveBeenCalled(); - expect(runtimeState.getStatus).not.toHaveBeenCalled(); - }); - it("reconciles prompt-learned agent session IDs even when runtime status omits them", async () => { const runtimeState = createRuntime(); runtimeState.ensureSession.mockResolvedValue({ @@ -470,46 +345,6 @@ describe("AcpSessionManager runtime config", () => { expect(currentMeta.identity?.acpxSessionId).toBe("acpx-stale"); }); - it("skips startup identity reconciliation for already resolved sessions", async () => { - const runtimeState = createRuntime(); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - const sessionKey = "agent:codex:acp:session-1"; - const resolvedMeta: SessionAcpMeta = { - ...readySessionMeta(), - identity: { - state: "resolved", - source: "status", - acpxSessionId: "acpx-sid-1", - agentSessionId: "agent-sid-1", - lastUpdatedAt: Date.now(), - }, - }; - hoisted.listAcpSessionEntriesMock.mockResolvedValue([ - { - cfg: baseCfg, - storePath: "/tmp/sessions-acp.json", - sessionKey, - storeSessionKey: sessionKey, - entry: { - sessionId: "session-1", - updatedAt: Date.now(), - acp: resolvedMeta, - }, - acp: resolvedMeta, - }, - ]); - - const manager = new AcpSessionManager(); - const result = await manager.reconcilePendingSessionIdentities({ cfg: baseCfg }); - - expect(result).toEqual({ checked: 0, resolved: 0, failed: 0 }); - expect(runtimeState.getStatus).not.toHaveBeenCalled(); - expect(runtimeState.ensureSession).not.toHaveBeenCalled(); - }); - it("preserves existing ACP session identifiers when ensure returns none", async () => { const runtimeState = createRuntime(); runtimeState.ensureSession.mockResolvedValue({ diff --git a/src/acp/control-plane/manager.runtime-options-commands.ts b/src/acp/control-plane/manager.runtime-options-commands.ts index ef0043c8d29e..335a0f2dbb44 100644 --- a/src/acp/control-plane/manager.runtime-options-commands.ts +++ b/src/acp/control-plane/manager.runtime-options-commands.ts @@ -4,9 +4,9 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { AcpRuntimeError, withAcpRuntimeErrorBoundary } from "../runtime/errors.js"; import type { ManagerRuntimeHandleCache } from "./manager.runtime-handle-cache.js"; import type { - AcpSessionResolution, AcpSessionRuntimeOptions, - SessionAcpMeta, + EnsureManagerRuntimeHandle, + ResolveManagerSession, WriteManagerSessionMeta, } from "./manager.types.js"; import { createUnsupportedControlError, requireReadySessionMeta } from "./manager.utils.js"; @@ -20,12 +20,8 @@ import { export type RuntimeOptionCommandServices = { runtimeHandles: ManagerRuntimeHandleCache; - resolveSession: (params: { cfg: OpenClawConfig; sessionKey: string }) => AcpSessionResolution; - ensureRuntimeHandle: (params: { - cfg: OpenClawConfig; - sessionKey: string; - meta: SessionAcpMeta; - }) => Promise<{ runtime: AcpRuntime; handle: AcpRuntimeHandle; meta: SessionAcpMeta }>; + resolveSession: ResolveManagerSession; + ensureRuntimeHandle: EnsureManagerRuntimeHandle; resolveRuntimeCapabilities: (params: { runtime: AcpRuntime; handle: AcpRuntimeHandle; diff --git a/src/acp/control-plane/manager.startup-identity-reconcile.test.ts b/src/acp/control-plane/manager.startup-identity-reconcile.test.ts new file mode 100644 index 000000000000..101beda8331e --- /dev/null +++ b/src/acp/control-plane/manager.startup-identity-reconcile.test.ts @@ -0,0 +1,170 @@ +import { describe, expect, it } from "vitest"; +import { + AcpSessionManager, + baseCfg, + createRuntime, + hoisted, + installAcpSessionManagerTestLifecycle, + readySessionMeta, + type SessionAcpMeta, +} from "./manager.test-helpers.js"; + +describe("AcpSessionManager startup identity reconcile", () => { + installAcpSessionManagerTestLifecycle(); + + it("reconciles pending ACP identities during startup scan", async () => { + const runtimeState = createRuntime(); + runtimeState.getStatus.mockResolvedValue({ + summary: "status=alive", + acpxRecordId: "acpx-record-1", + backendSessionId: "acpx-session-1", + agentSessionId: "agent-session-1", + details: { status: "alive" }, + }); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + + let currentMeta: SessionAcpMeta = { + ...readySessionMeta(), + identity: { + state: "pending", + source: "ensure", + acpxSessionId: "acpx-stale", + lastUpdatedAt: Date.now(), + }, + }; + const sessionKey = "agent:codex:acp:session-1"; + hoisted.listAcpSessionEntriesMock.mockResolvedValue([ + { + cfg: baseCfg, + storePath: "/tmp/sessions-acp.json", + sessionKey, + storeSessionKey: sessionKey, + entry: { + sessionId: "session-1", + updatedAt: Date.now(), + acp: currentMeta, + }, + acp: currentMeta, + }, + ]); + hoisted.readAcpSessionEntryMock.mockImplementation((paramsUnknown: unknown) => { + const key = (paramsUnknown as { sessionKey?: string }).sessionKey ?? sessionKey; + return { + sessionKey: key, + storeSessionKey: key, + acp: currentMeta, + }; + }); + hoisted.upsertAcpSessionMetaMock.mockImplementation(async (paramsUnknown: unknown) => { + const params = paramsUnknown as { + mutate: ( + current: SessionAcpMeta | undefined, + entry: { acp?: SessionAcpMeta } | undefined, + ) => SessionAcpMeta | null | undefined; + }; + const next = params.mutate(currentMeta, { acp: currentMeta }); + if (next) { + currentMeta = next; + } + return { + sessionId: "session-1", + updatedAt: Date.now(), + acp: currentMeta, + }; + }); + + const manager = new AcpSessionManager(); + const result = await manager.reconcilePendingSessionIdentities({ cfg: baseCfg }); + + expect(result).toEqual({ checked: 1, resolved: 1, failed: 0 }); + expect(currentMeta.identity?.state).toBe("resolved"); + expect(currentMeta.identity?.acpxRecordId).toBe("acpx-record-1"); + expect(currentMeta.identity?.acpxSessionId).toBe("acpx-session-1"); + expect(currentMeta.identity?.agentSessionId).toBe("agent-session-1"); + }); + + it("skips startup reconcile for pending identities without stable runtime ids", async () => { + const runtimeState = createRuntime(); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + + const sessionKey = "agent:claude:acp:binding:discord:default:9373ab192b2317f4"; + const acp = { + ...readySessionMeta({ + agent: "claude", + }), + identity: { + state: "pending" as const, + acpxRecordId: sessionKey, + source: "status" as const, + lastUpdatedAt: Date.now(), + }, + }; + hoisted.listAcpSessionEntriesMock.mockResolvedValue([ + { + cfg: baseCfg, + storePath: "/tmp/sessions-acp.json", + sessionKey, + storeSessionKey: sessionKey, + entry: { + sessionId: "session-1", + updatedAt: Date.now(), + acp, + }, + acp, + }, + ]); + + const manager = new AcpSessionManager(); + const result = await manager.reconcilePendingSessionIdentities({ cfg: baseCfg }); + + expect(result).toEqual({ checked: 0, resolved: 0, failed: 0 }); + expect(runtimeState.ensureSession).not.toHaveBeenCalled(); + expect(runtimeState.getStatus).not.toHaveBeenCalled(); + }); + + it("skips startup identity reconciliation for already resolved sessions", async () => { + const runtimeState = createRuntime(); + hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ + id: "acpx", + runtime: runtimeState.runtime, + }); + const sessionKey = "agent:codex:acp:session-1"; + const resolvedMeta: SessionAcpMeta = { + ...readySessionMeta(), + identity: { + state: "resolved", + source: "status", + acpxSessionId: "acpx-sid-1", + agentSessionId: "agent-sid-1", + lastUpdatedAt: Date.now(), + }, + }; + hoisted.listAcpSessionEntriesMock.mockResolvedValue([ + { + cfg: baseCfg, + storePath: "/tmp/sessions-acp.json", + sessionKey, + storeSessionKey: sessionKey, + entry: { + sessionId: "session-1", + updatedAt: Date.now(), + acp: resolvedMeta, + }, + acp: resolvedMeta, + }, + ]); + + const manager = new AcpSessionManager(); + const result = await manager.reconcilePendingSessionIdentities({ cfg: baseCfg }); + + expect(result).toEqual({ checked: 0, resolved: 0, failed: 0 }); + expect(runtimeState.getStatus).not.toHaveBeenCalled(); + expect(runtimeState.ensureSession).not.toHaveBeenCalled(); + }); +}); diff --git a/src/acp/control-plane/manager.startup-identity-reconcile.ts b/src/acp/control-plane/manager.startup-identity-reconcile.ts new file mode 100644 index 000000000000..0b79e92d9dc8 --- /dev/null +++ b/src/acp/control-plane/manager.startup-identity-reconcile.ts @@ -0,0 +1,88 @@ +import { + identityHasStableSessionId, + isSessionIdentityPending, + resolveSessionIdentityFromMeta, +} from "@openclaw/acp-core/runtime/session-identity"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { logVerbose } from "../../globals.js"; +import type { + AcpSessionManagerDeps, + AcpStartupIdentityReconcileResult, + EnsureManagerRuntimeHandle, + ReconcileManagerRuntimeSessionIdentifiers, + ResolveManagerSession, + WithManagerSessionActor, +} from "./manager.types.js"; + +export async function runManagerStartupIdentityReconcile(params: { + cfg: OpenClawConfig; + deps: Pick; + withSessionActor: WithManagerSessionActor; + resolveSession: ResolveManagerSession; + ensureRuntimeHandle: EnsureManagerRuntimeHandle; + reconcileRuntimeSessionIdentifiers: ReconcileManagerRuntimeSessionIdentifiers; +}): Promise { + let checked = 0; + let resolved = 0; + let failed = 0; + + let acpSessions: Awaited>; + try { + acpSessions = await params.deps.listAcpSessions({ + cfg: params.cfg, + }); + } catch (error) { + logVerbose(`acp-manager: startup identity scan failed: ${String(error)}`); + return { checked, resolved, failed: failed + 1 }; + } + + for (const session of acpSessions) { + if (!session.acp || !session.sessionKey) { + continue; + } + const currentIdentity = resolveSessionIdentityFromMeta(session.acp); + if ( + !isSessionIdentityPending(currentIdentity) || + !identityHasStableSessionId(currentIdentity) + ) { + continue; + } + + checked += 1; + try { + const becameResolved = await params.withSessionActor(session.sessionKey, async () => { + const resolution = params.resolveSession({ + cfg: params.cfg, + sessionKey: session.sessionKey, + }); + if (resolution.kind !== "ready") { + return false; + } + const { runtime, handle, meta } = await params.ensureRuntimeHandle({ + cfg: params.cfg, + sessionKey: session.sessionKey, + meta: resolution.meta, + }); + const reconciled = await params.reconcileRuntimeSessionIdentifiers({ + cfg: params.cfg, + sessionKey: session.sessionKey, + runtime, + handle, + meta, + failOnStatusError: false, + }); + return !isSessionIdentityPending(resolveSessionIdentityFromMeta(reconciled.meta)); + }); + if (becameResolved) { + resolved += 1; + } + } catch (error) { + failed += 1; + logVerbose( + `acp-manager: startup identity reconcile failed for ${session.sessionKey}: ${String(error)}`, + ); + } + } + + return { checked, resolved, failed }; +} diff --git a/src/acp/control-plane/manager.status.ts b/src/acp/control-plane/manager.status.ts new file mode 100644 index 000000000000..67e908aa00d8 --- /dev/null +++ b/src/acp/control-plane/manager.status.ts @@ -0,0 +1,89 @@ +import { resolveSessionIdentityFromMeta } from "@openclaw/acp-core/runtime/session-identity"; +import type { + AcpRuntime, + AcpRuntimeCapabilities, + AcpRuntimeHandle, + AcpRuntimeStatus, +} from "@openclaw/acp-core/runtime/types"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { withAcpRuntimeErrorBoundary } from "../runtime/errors.js"; +import type { + AcpSessionStatus, + EnsureManagerRuntimeHandle, + ReconcileManagerRuntimeSessionIdentifiers, + ResolveManagerSession, +} from "./manager.types.js"; +import { requireReadySessionMeta } from "./manager.utils.js"; +import { resolveRuntimeOptionsFromMeta } from "./runtime-options.js"; + +export async function runManagerGetSessionStatus(params: { + cfg: OpenClawConfig; + sessionKey: string; + signal?: AbortSignal; + throwIfAborted: (signal?: AbortSignal) => void; + resolveSession: ResolveManagerSession; + ensureRuntimeHandle: EnsureManagerRuntimeHandle; + resolveRuntimeCapabilities: (params: { + runtime: AcpRuntime; + handle: AcpRuntimeHandle; + }) => Promise; + reconcileRuntimeSessionIdentifiers: ReconcileManagerRuntimeSessionIdentifiers; +}): Promise { + params.throwIfAborted(params.signal); + const resolution = params.resolveSession({ + cfg: params.cfg, + sessionKey: params.sessionKey, + }); + const resolvedMeta = requireReadySessionMeta(resolution); + const { + runtime, + handle: ensuredHandle, + meta: ensuredMeta, + } = await params.ensureRuntimeHandle({ + cfg: params.cfg, + sessionKey: params.sessionKey, + meta: resolvedMeta, + }); + let handle = ensuredHandle; + let meta = ensuredMeta; + const capabilities = await params.resolveRuntimeCapabilities({ runtime, handle }); + let runtimeStatus: AcpRuntimeStatus | undefined; + if (runtime.getStatus) { + runtimeStatus = await withAcpRuntimeErrorBoundary({ + run: async () => { + params.throwIfAborted(params.signal); + const status = await runtime.getStatus!({ + handle, + ...(params.signal ? { signal: params.signal } : {}), + }); + params.throwIfAborted(params.signal); + return status; + }, + fallbackCode: "ACP_TURN_FAILED", + fallbackMessage: "Could not read ACP runtime status.", + }); + } + ({ handle, meta, runtimeStatus } = await params.reconcileRuntimeSessionIdentifiers({ + cfg: params.cfg, + sessionKey: params.sessionKey, + runtime, + handle, + meta, + runtimeStatus, + failOnStatusError: true, + })); + const identity = resolveSessionIdentityFromMeta(meta); + return { + sessionKey: params.sessionKey, + backend: handle.backend || meta.backend, + agent: meta.agent, + ...(identity ? { identity } : {}), + state: meta.state, + mode: meta.mode, + runtimeOptions: resolveRuntimeOptionsFromMeta(meta), + capabilities, + runtimeStatus, + lastActivityAt: meta.lastActivityAt, + lastError: meta.lastError, + }; +} diff --git a/src/acp/control-plane/manager.test.ts b/src/acp/control-plane/manager.test.ts index 95b227c41108..2d02e919293e 100644 --- a/src/acp/control-plane/manager.test.ts +++ b/src/acp/control-plane/manager.test.ts @@ -15,7 +15,6 @@ import { createRuntime, expectRecordFields, expectRejectedRecord, - extractRuntimeOptionsFromUpserts, extractStateUpsertPersistenceOptions, extractStatesFromUpserts, flushMicrotasks, @@ -940,47 +939,6 @@ describe("AcpSessionManager", () => { expect(runtimeState.ensureSession).toHaveBeenCalledTimes(1); }); - it("enforces acp.maxConcurrentSessions during initializeSession", async () => { - const runtimeState = createRuntime(); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - hoisted.upsertAcpSessionMetaMock.mockResolvedValue({ - sessionKey: "agent:codex:acp:session-a", - storeSessionKey: "agent:codex:acp:session-a", - acp: readySessionMeta(), - }); - const limitedCfg = { - acp: { - ...baseCfg.acp, - maxConcurrentSessions: 1, - }, - } as OpenClawConfig; - - const manager = new AcpSessionManager(); - await manager.initializeSession({ - cfg: limitedCfg, - sessionKey: "agent:codex:acp:session-a", - agent: "codex", - mode: "persistent", - }); - - await expectRejectedRecord( - manager.initializeSession({ - cfg: limitedCfg, - sessionKey: "agent:codex:acp:session-b", - agent: "codex", - mode: "persistent", - }), - { - code: "ACP_SESSION_INIT_FAILED", - message: "ACP max concurrent sessions reached (1/1).", - }, - ); - expect(runtimeState.ensureSession).toHaveBeenCalledTimes(1); - }); - it("uses metadata backend when global acp.backend is unset", async () => { const runtimeState = createRuntime(); runtimeState.ensureSession.mockImplementation(async (input) => ({ @@ -1028,87 +986,6 @@ describe("AcpSessionManager", () => { expect(runtimeState.runTurn).toHaveBeenCalledTimes(1); }); - it("persists runtime options provided during initializeSession", async () => { - const runtimeState = createRuntime(); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - hoisted.upsertAcpSessionMetaMock.mockResolvedValue({ - sessionKey: "agent:codex:acp:session-a", - storeSessionKey: "agent:codex:acp:session-a", - acp: readySessionMeta({ - runtimeOptions: { - model: "openai/gpt-5.4", - thinking: "high", - }, - }), - }); - - const manager = new AcpSessionManager(); - await manager.initializeSession({ - cfg: baseCfg, - sessionKey: "agent:codex:acp:session-a", - agent: "codex", - mode: "persistent", - runtimeOptions: { - model: "openai/gpt-5.4", - thinking: "high", - }, - }); - - expect(extractRuntimeOptionsFromUpserts()).toEqual([ - { - model: "openai/gpt-5.4", - thinking: "high", - }, - ]); - expectRecordFields(mockCallArg(runtimeState.ensureSession), { - sessionKey: "agent:codex:acp:session-a", - model: "openai/gpt-5.4", - thinking: "high", - }); - }); - - it("preserves runtimeOptions cwd when initializeSession cwd is omitted", async () => { - const runtimeState = createRuntime(); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - hoisted.upsertAcpSessionMetaMock.mockResolvedValue({ - sessionKey: "agent:codex:acp:session-cwd-runtime-options", - storeSessionKey: "agent:codex:acp:session-cwd-runtime-options", - acp: readySessionMeta({ - runtimeOptions: { - cwd: "/workspace/from-runtime-options", - }, - cwd: "/workspace/from-runtime-options", - }), - }); - - const manager = new AcpSessionManager(); - await manager.initializeSession({ - cfg: baseCfg, - sessionKey: "agent:codex:acp:session-cwd-runtime-options", - agent: "codex", - mode: "persistent", - runtimeOptions: { - cwd: "/workspace/from-runtime-options", - }, - }); - - expectRecordFields(mockCallArg(runtimeState.ensureSession), { - sessionKey: "agent:codex:acp:session-cwd-runtime-options", - cwd: "/workspace/from-runtime-options", - }); - expect(extractRuntimeOptionsFromUpserts()).toEqual([ - { - cwd: "/workspace/from-runtime-options", - }, - ]); - }); - it("drops cached runtime handles after tolerated close failures", async () => { const closeFailures = [ { @@ -1564,89 +1441,6 @@ describe("AcpSessionManager", () => { expect(snapshot.errorsByCode.ACP_TURN_FAILED).toBe(1); }); - it("rolls back ensured runtime sessions when metadata persistence fails", async () => { - const runtimeState = createRuntime(); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - hoisted.upsertAcpSessionMetaMock.mockRejectedValueOnce(new Error("disk full")); - - const manager = new AcpSessionManager(); - await expect( - manager.initializeSession({ - cfg: baseCfg, - sessionKey: "agent:codex:acp:session-1", - agent: "codex", - mode: "persistent", - }), - ).rejects.toThrow("disk full"); - const closeInput = mockCallArg(runtimeState.close); - expectRecordFields(closeInput, { - reason: "init-meta-failed", - }); - expectRecordFields(closeInput.handle, { - sessionKey: "agent:codex:acp:session-1", - }); - }); - - it("preempts an active turn on cancel and returns to idle state", async () => { - const runtimeState = createRuntime(); - hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ - id: "acpx", - runtime: runtimeState.runtime, - }); - hoisted.readAcpSessionEntryMock.mockReturnValue({ - sessionKey: "agent:codex:acp:session-1", - storeSessionKey: "agent:codex:acp:session-1", - acp: readySessionMeta(), - }); - - let enteredRun = false; - runtimeState.runTurn.mockImplementation(async function* (input: { signal?: AbortSignal }) { - enteredRun = true; - await new Promise((resolve) => { - if (input.signal?.aborted) { - resolve(); - return; - } - input.signal?.addEventListener("abort", () => resolve(), { once: true }); - }); - yield { type: "done" as const, stopReason: "cancel" }; - }); - - const manager = new AcpSessionManager(); - const runPromise = manager.runTurn({ - cfg: baseCfg, - sessionKey: "agent:codex:acp:session-1", - text: "long task", - mode: "prompt", - requestId: "run-1", - }); - await vi.waitFor( - () => { - expect(enteredRun).toBe(true); - }, - { interval: 1 }, - ); - - await manager.cancelSession({ - cfg: baseCfg, - sessionKey: "agent:codex:acp:session-1", - reason: "manual-cancel", - }); - await runPromise; - - expect(runtimeState.cancel).toHaveBeenCalledTimes(1); - expectRecordFields(mockCallArg(runtimeState.cancel), { - reason: "manual-cancel", - }); - const states = extractStatesFromUpserts(); - expect(states).toContain("running"); - expect(states).toContain("idle"); - expect(states).not.toContain("error"); - }); - it("cleans actor-tail bookkeeping after session turns complete", async () => { const runtimeState = createRuntime(); hoisted.requireAcpRuntimeBackendMock.mockReturnValue({ diff --git a/src/acp/control-plane/manager.turn-runner.ts b/src/acp/control-plane/manager.turn-runner.ts index e0ee0331639b..e35b09ef12b2 100644 --- a/src/acp/control-plane/manager.turn-runner.ts +++ b/src/acp/control-plane/manager.turn-runner.ts @@ -28,8 +28,11 @@ import { import type { AcpRunTurnInput, AcpSessionManagerDeps, - AcpSessionResolution, ActiveTurnState, + EnsureManagerRuntimeHandle, + ReconcileManagerRuntimeSessionIdentifiers, + ResolveManagerSession, + SetManagerSessionState, SessionAcpMeta, WriteManagerSessionMeta, } from "./manager.types.js"; @@ -37,12 +40,6 @@ import { normalizeActorKey, requireReadySessionMeta } from "./manager.utils.js"; const ACP_TURN_TIMEOUT_GRACE_MS = 1_000; -type EnsureRuntimeHandle = (params: { - cfg: AcpRunTurnInput["cfg"]; - sessionKey: string; - meta: SessionAcpMeta; -}) => Promise<{ runtime: AcpRuntime; handle: AcpRuntimeHandle; meta: SessionAcpMeta }>; - type ApplyRuntimeControls = (params: { sessionKey: string; runtime: AcpRuntime; @@ -50,41 +47,21 @@ type ApplyRuntimeControls = (params: { meta: SessionAcpMeta; }) => Promise; -type SetSessionState = (params: { - cfg: AcpRunTurnInput["cfg"]; - sessionKey: string; - state: SessionAcpMeta["state"]; - lastError?: string; - clearLastError?: boolean; -}) => Promise; - -type ReconcileRuntimeSessionIdentifiers = (params: { - cfg: AcpRunTurnInput["cfg"]; - sessionKey: string; - runtime: AcpRuntime; - handle: AcpRuntimeHandle; - meta: SessionAcpMeta; - failOnStatusError: boolean; -}) => Promise<{ handle: AcpRuntimeHandle; meta: SessionAcpMeta }>; - export async function runManagerTurn(params: { input: AcpRunTurnInput; sessionKey: string; deps: AcpSessionManagerDeps; runtimeHandles: ManagerRuntimeHandleCache; activeTurnBySession: Map; - resolveSession: (params: { - cfg: AcpRunTurnInput["cfg"]; - sessionKey: string; - }) => AcpSessionResolution; - ensureRuntimeHandle: EnsureRuntimeHandle; + resolveSession: ResolveManagerSession; + ensureRuntimeHandle: EnsureManagerRuntimeHandle; applyRuntimeControls: ApplyRuntimeControls; - setSessionState: SetSessionState; + setSessionState: SetManagerSessionState; recordTurnCompletion: (params: { startedAt: number; errorCode?: AcpRuntimeError["code"]; }) => void; - reconcileRuntimeSessionIdentifiers: ReconcileRuntimeSessionIdentifiers; + reconcileRuntimeSessionIdentifiers: ReconcileManagerRuntimeSessionIdentifiers; writeSessionMeta: WriteManagerSessionMeta; }): Promise { const { input, sessionKey } = params; diff --git a/src/acp/control-plane/manager.types.ts b/src/acp/control-plane/manager.types.ts index 7707ed49d531..087e518dfd4d 100644 --- a/src/acp/control-plane/manager.types.ts +++ b/src/acp/control-plane/manager.types.ts @@ -159,6 +159,41 @@ export type WriteManagerSessionMeta = (params: { takeCacheOwnership?: boolean; }) => Promise; +export type ResolveManagerSession = (params: { + cfg: OpenClawConfig; + sessionKey: string; +}) => AcpSessionResolution; + +export type EnsureManagerRuntimeHandle = (params: { + cfg: OpenClawConfig; + sessionKey: string; + meta: SessionAcpMeta; +}) => Promise<{ runtime: AcpRuntime; handle: AcpRuntimeHandle; meta: SessionAcpMeta }>; + +export type ReconcileManagerRuntimeSessionIdentifiers = (params: { + cfg: OpenClawConfig; + sessionKey: string; + runtime: AcpRuntime; + handle: AcpRuntimeHandle; + meta: SessionAcpMeta; + runtimeStatus?: AcpRuntimeStatus; + failOnStatusError: boolean; +}) => Promise<{ + handle: AcpRuntimeHandle; + meta: SessionAcpMeta; + runtimeStatus?: AcpRuntimeStatus; +}>; + +export type SetManagerSessionState = (params: { + cfg: OpenClawConfig; + sessionKey: string; + state: SessionAcpMeta["state"]; + lastError?: string; + clearLastError?: boolean; +}) => Promise; + +export type WithManagerSessionActor = (sessionKey: string, op: () => Promise) => Promise; + export const DEFAULT_DEPS: AcpSessionManagerDeps = { listAcpSessions: listAcpSessionEntries, readSessionEntry: readAcpSessionEntry,