Skip to content

Commit 9c256f2

Browse files
steipetesteipete
committed
refactor: split ACP manager session flows
1 parent 720071a commit 9c256f2

14 files changed

Lines changed: 960 additions & 732 deletions

‎src/acp/control-plane/manager.cancel-session.test.ts‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
import { describe, expect, it, vi } from "vitest";
2+
import {
3+
AcpSessionManager,
4+
baseCfg,
5+
createRuntime,
6+
expectRecordFields,
7+
extractStatesFromUpserts,
8+
hoisted,
9+
installAcpSessionManagerTestLifecycle,
10+
mockCallArg,
11+
readySessionMeta,
12+
} from "./manager.test-helpers.js";
13+
14+
describe("AcpSessionManager cancelSession", () => {
15+
installAcpSessionManagerTestLifecycle();
16+
17+
it("preempts an active turn on cancel and returns to idle state", async () => {
18+
const runtimeState = createRuntime();
19+
hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
20+
id: "acpx",
21+
runtime: runtimeState.runtime,
22+
});
23+
hoisted.readAcpSessionEntryMock.mockReturnValue({
24+
sessionKey: "agent:codex:acp:session-1",
25+
storeSessionKey: "agent:codex:acp:session-1",
26+
acp: readySessionMeta(),
27+
});
28+
29+
let enteredRun = false;
30+
runtimeState.runTurn.mockImplementation(async function* (input: { signal?: AbortSignal }) {
31+
enteredRun = true;
32+
await new Promise<void>((resolve) => {
33+
if (input.signal?.aborted) {
34+
resolve();
35+
return;
36+
}
37+
input.signal?.addEventListener("abort", () => resolve(), { once: true });
38+
});
39+
yield { type: "done" as const, stopReason: "cancel" };
40+
});
41+
42+
const manager = new AcpSessionManager();
43+
const runPromise = manager.runTurn({
44+
cfg: baseCfg,
45+
sessionKey: "agent:codex:acp:session-1",
46+
text: "long task",
47+
mode: "prompt",
48+
requestId: "run-1",
49+
});
50+
await vi.waitFor(
51+
() => {
52+
expect(enteredRun).toBe(true);
53+
},
54+
{ interval: 1 },
55+
);
56+
57+
await manager.cancelSession({
58+
cfg: baseCfg,
59+
sessionKey: "agent:codex:acp:session-1",
60+
reason: "manual-cancel",
61+
});
62+
await runPromise;
63+
64+
expect(runtimeState.cancel).toHaveBeenCalledTimes(1);
65+
expectRecordFields(mockCallArg(runtimeState.cancel), {
66+
reason: "manual-cancel",
67+
});
68+
const states = extractStatesFromUpserts();
69+
expect(states).toContain("running");
70+
expect(states).toContain("idle");
71+
expect(states).not.toContain("error");
72+
});
73+
});

‎src/acp/control-plane/manager.cancel-session.ts‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
import type { AcpRuntime, AcpRuntimeHandle } from "@openclaw/acp-core/runtime/types";
2+
import type { OpenClawConfig } from "../../config/types.openclaw.js";
3+
import {
4+
type AcpRuntimeError,
5+
toAcpRuntimeError,
6+
withAcpRuntimeErrorBoundary,
7+
} from "../runtime/errors.js";
8+
import type {
9+
ActiveTurnState,
10+
EnsureManagerRuntimeHandle,
11+
ResolveManagerSession,
12+
SetManagerSessionState,
13+
WithManagerSessionActor,
14+
} from "./manager.types.js";
15+
import { normalizeActorKey, requireReadySessionMeta } from "./manager.utils.js";
16+
17+
export async function runManagerCancelSession(params: {
18+
cfg: OpenClawConfig;
19+
sessionKey: string;
20+
reason?: string;
21+
activeTurnBySession: Map<string, ActiveTurnState>;
22+
withSessionActor: WithManagerSessionActor;
23+
resolveSession: ResolveManagerSession;
24+
ensureRuntimeHandle: EnsureManagerRuntimeHandle;
25+
setSessionState: SetManagerSessionState;
26+
}): Promise<void> {
27+
const actorKey = normalizeActorKey(params.sessionKey);
28+
const activeTurn = params.activeTurnBySession.get(actorKey);
29+
if (activeTurn) {
30+
await cancelActiveTurn({
31+
activeTurn,
32+
reason: params.reason,
33+
});
34+
return;
35+
}
36+
37+
await params.withSessionActor(params.sessionKey, async () => {
38+
const resolution = params.resolveSession({
39+
cfg: params.cfg,
40+
sessionKey: params.sessionKey,
41+
});
42+
const resolvedMeta = requireReadySessionMeta(resolution);
43+
const { runtime, handle } = await params.ensureRuntimeHandle({
44+
cfg: params.cfg,
45+
sessionKey: params.sessionKey,
46+
meta: resolvedMeta,
47+
});
48+
try {
49+
await cancelRuntimeHandle({
50+
runtime,
51+
handle,
52+
reason: params.reason,
53+
});
54+
await params.setSessionState({
55+
cfg: params.cfg,
56+
sessionKey: params.sessionKey,
57+
state: "idle",
58+
clearLastError: true,
59+
});
60+
} catch (error) {
61+
const acpError = normalizeCancelError(error);
62+
await params.setSessionState({
63+
cfg: params.cfg,
64+
sessionKey: params.sessionKey,
65+
state: "error",
66+
lastError: acpError.message,
67+
});
68+
throw acpError;
69+
}
70+
});
71+
}
72+
73+
async function cancelActiveTurn(params: {
74+
activeTurn: ActiveTurnState;
75+
reason?: string;
76+
}): Promise<void> {
77+
params.activeTurn.abortController.abort();
78+
if (!params.activeTurn.cancelPromise) {
79+
params.activeTurn.cancelPromise = params.activeTurn.runtime.cancel({
80+
handle: params.activeTurn.handle,
81+
reason: params.reason,
82+
});
83+
}
84+
await withAcpRuntimeErrorBoundary({
85+
run: async () => await params.activeTurn.cancelPromise!,
86+
fallbackCode: "ACP_TURN_FAILED",
87+
fallbackMessage: "ACP cancel failed before completion.",
88+
});
89+
}
90+
91+
async function cancelRuntimeHandle(params: {
92+
runtime: AcpRuntime;
93+
handle: AcpRuntimeHandle;
94+
reason?: string;
95+
}): Promise<void> {
96+
await withAcpRuntimeErrorBoundary({
97+
run: async () =>
98+
await params.runtime.cancel({
99+
handle: params.handle,
100+
reason: params.reason,
101+
}),
102+
fallbackCode: "ACP_TURN_FAILED",
103+
fallbackMessage: "ACP cancel failed before completion.",
104+
});
105+
}
106+
107+
function normalizeCancelError(error: unknown): AcpRuntimeError {
108+
return toAcpRuntimeError({
109+
error,
110+
fallbackCode: "ACP_TURN_FAILED",
111+
fallbackMessage: "ACP cancel failed before completion.",
112+
});
113+
}

‎src/acp/control-plane/manager.close-session.ts‎

Lines changed: 4 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@ import {
22
identityHasStableSessionId,
33
resolveSessionIdentityFromMeta,
44
} from "@openclaw/acp-core/runtime/session-identity";
5-
import type { AcpRuntime, AcpRuntimeHandle } from "@openclaw/acp-core/runtime/types";
6-
import type { OpenClawConfig } from "../../config/types.openclaw.js";
75
import { toAcpRuntimeError, withAcpRuntimeErrorBoundary } from "../runtime/errors.js";
86
import type { ManagerRuntimeHandleCache } from "./manager.runtime-handle-cache.js";
97
import {
@@ -15,25 +13,19 @@ import type {
1513
AcpCloseSessionInput,
1614
AcpCloseSessionResult,
1715
AcpSessionManagerDeps,
18-
AcpSessionResolution,
19-
SessionAcpMeta,
16+
EnsureManagerRuntimeHandle,
17+
ResolveManagerSession,
2018
WriteManagerSessionMeta,
2119
} from "./manager.types.js";
2220
import { requireReadySessionMeta, resolveAcpSessionResolutionError } from "./manager.utils.js";
2321

24-
type EnsureRuntimeHandle = (params: {
25-
cfg: OpenClawConfig;
26-
sessionKey: string;
27-
meta: SessionAcpMeta;
28-
}) => Promise<{ runtime: AcpRuntime; handle: AcpRuntimeHandle; meta: SessionAcpMeta }>;
29-
3022
export async function runManagerCloseSession(params: {
3123
input: AcpCloseSessionInput;
3224
sessionKey: string;
3325
deps: Pick<AcpSessionManagerDeps, "getRuntimeBackend">;
3426
runtimeHandles: ManagerRuntimeHandleCache;
35-
resolveSession: (params: { cfg: OpenClawConfig; sessionKey: string }) => AcpSessionResolution;
36-
ensureRuntimeHandle: EnsureRuntimeHandle;
27+
resolveSession: ResolveManagerSession;
28+
ensureRuntimeHandle: EnsureManagerRuntimeHandle;
3729
writeSessionMeta: WriteManagerSessionMeta;
3830
}): Promise<AcpCloseSessionResult> {
3931
const { input, sessionKey } = params;

0 commit comments

Comments
 (0)