mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor: split ACP manager session flows
Split ACP manager session-flow ownership into focused helpers for initialization, status reads, cancellation, and startup identity reconciliation. Verification: - `node scripts/run-oxlint.mjs src/acp/control-plane/manager.core.ts src/acp/control-plane/manager.initialize-session.ts src/acp/control-plane/manager.status.ts src/acp/control-plane/manager.cancel-session.ts src/acp/control-plane/manager.startup-identity-reconcile.ts src/acp/control-plane/manager.close-session.ts src/acp/control-plane/manager.turn-runner.ts src/acp/control-plane/manager.runtime-options-commands.ts src/acp/control-plane/manager.types.ts src/acp/control-plane/manager.test.ts src/acp/control-plane/manager.initialize-session.test.ts src/acp/control-plane/manager.cancel-session.test.ts src/acp/control-plane/manager.startup-identity-reconcile.test.ts src/acp/control-plane/manager.runtime-config.test.ts` - `pnpm tsgo:prod` - `pnpm test src/acp/control-plane/manager.test.ts src/acp/control-plane/manager.initialize-session.test.ts src/acp/control-plane/manager.cancel-session.test.ts src/acp/control-plane/manager.startup-identity-reconcile.test.ts src/acp/control-plane/manager.runtime-config.test.ts src/acp/control-plane/manager.runtime-handles.test.ts` - `pnpm format:check src/acp/control-plane/manager.core.ts src/acp/control-plane/manager.initialize-session.ts src/acp/control-plane/manager.status.ts src/acp/control-plane/manager.cancel-session.ts src/acp/control-plane/manager.startup-identity-reconcile.ts src/acp/control-plane/manager.close-session.ts src/acp/control-plane/manager.turn-runner.ts src/acp/control-plane/manager.runtime-options-commands.ts src/acp/control-plane/manager.types.ts src/acp/control-plane/manager.test.ts src/acp/control-plane/manager.initialize-session.test.ts src/acp/control-plane/manager.cancel-session.test.ts src/acp/control-plane/manager.startup-identity-reconcile.test.ts src/acp/control-plane/manager.runtime-config.test.ts` - `git diff --check` - `pnpm check:test-types` - `.agents/skills/autoreview/scripts/autoreview --mode branch --base origin/main` - GitHub PR checks for #88752 passed Real behavior proof: Behavior addressed: ACP manager session-flow ownership is split out of `AcpSessionManager` without changing initialization, status, cancel, startup identity reconciliation, close, turn, or runtime-option behavior. Real environment tested: Local OpenClaw checkout, Node/pnpm repo toolchain, GitHub Actions PR CI. Exact steps or command run after this patch: Focused ACP manager/runtime config/runtime handle tests plus prod/test type checks, lint, format check, diff check, autoreview, and PR CI. Evidence after fix: All listed local commands passed, autoreview reported no accepted/actionable findings, and GitHub PR checks passed. Observed result after fix: `manager.core.ts` is down to 612 LOC, with init/status/cancel/startup identity flows in focused modules and matching focused tests. What was not tested: Live ACP backend session initialization/cancel/status against a real external ACP provider.
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
8cfccca4de
commit
118b9cacf6
@@ -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<void>((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");
|
||||
});
|
||||
});
|
||||
@@ -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<string, ActiveTurnState>;
|
||||
withSessionActor: WithManagerSessionActor;
|
||||
resolveSession: ResolveManagerSession;
|
||||
ensureRuntimeHandle: EnsureManagerRuntimeHandle;
|
||||
setSessionState: SetManagerSessionState;
|
||||
}): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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.",
|
||||
});
|
||||
}
|
||||
@@ -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<AcpSessionManagerDeps, "getRuntimeBackend">;
|
||||
runtimeHandles: ManagerRuntimeHandleCache;
|
||||
resolveSession: (params: { cfg: OpenClawConfig; sessionKey: string }) => AcpSessionResolution;
|
||||
ensureRuntimeHandle: EnsureRuntimeHandle;
|
||||
resolveSession: ResolveManagerSession;
|
||||
ensureRuntimeHandle: EnsureManagerRuntimeHandle;
|
||||
writeSessionMeta: WriteManagerSessionMeta;
|
||||
}): Promise<AcpCloseSessionResult> {
|
||||
const { input, sessionKey } = params;
|
||||
|
||||
@@ -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<AcpStartupIdentityReconcileResult> {
|
||||
let checked = 0;
|
||||
let resolved = 0;
|
||||
let failed = 0;
|
||||
|
||||
let acpSessions: Awaited<ReturnType<AcpSessionManagerDeps["listAcpSessions"]>>;
|
||||
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),
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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<AcpSessionManagerDeps, "requireRuntimeBackend">;
|
||||
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<SessionEntry | null> {
|
||||
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<void> {
|
||||
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)}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
@@ -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({
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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<AcpSessionManagerDeps, "listAcpSessions">;
|
||||
withSessionActor: WithManagerSessionActor;
|
||||
resolveSession: ResolveManagerSession;
|
||||
ensureRuntimeHandle: EnsureManagerRuntimeHandle;
|
||||
reconcileRuntimeSessionIdentifiers: ReconcileManagerRuntimeSessionIdentifiers;
|
||||
}): Promise<AcpStartupIdentityReconcileResult> {
|
||||
let checked = 0;
|
||||
let resolved = 0;
|
||||
let failed = 0;
|
||||
|
||||
let acpSessions: Awaited<ReturnType<AcpSessionManagerDeps["listAcpSessions"]>>;
|
||||
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 };
|
||||
}
|
||||
@@ -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<AcpRuntimeCapabilities>;
|
||||
reconcileRuntimeSessionIdentifiers: ReconcileManagerRuntimeSessionIdentifiers;
|
||||
}): Promise<AcpSessionStatus> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
@@ -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<void>((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({
|
||||
|
||||
@@ -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<void>;
|
||||
|
||||
type SetSessionState = (params: {
|
||||
cfg: AcpRunTurnInput["cfg"];
|
||||
sessionKey: string;
|
||||
state: SessionAcpMeta["state"];
|
||||
lastError?: string;
|
||||
clearLastError?: boolean;
|
||||
}) => Promise<void>;
|
||||
|
||||
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<string, ActiveTurnState>;
|
||||
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<void> {
|
||||
const { input, sessionKey } = params;
|
||||
|
||||
@@ -159,6 +159,41 @@ export type WriteManagerSessionMeta = (params: {
|
||||
takeCacheOwnership?: boolean;
|
||||
}) => Promise<SessionEntry | null>;
|
||||
|
||||
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<void>;
|
||||
|
||||
export type WithManagerSessionActor = <T>(sessionKey: string, op: () => Promise<T>) => Promise<T>;
|
||||
|
||||
export const DEFAULT_DEPS: AcpSessionManagerDeps = {
|
||||
listAcpSessions: listAcpSessionEntries,
|
||||
readSessionEntry: readAcpSessionEntry,
|
||||
|
||||
Reference in New Issue
Block a user