History

Add ACP replay harness and session lifecycle support

- Add replay agent and fixture-driven ACP transcript testing - Support session lifecycle, auth methods, and session resume/fork flows - Expand orchestration V2 adapter coverage and debug tooling

Julius Marminge committed 4 months ago

Commit 11615291 parent 9d38de865 files · +9914 −74Browse files

Showing the changes to apps/server/scripts (2 of 65 files). Show all

Unified · Split

2 files
  1. apps/server/scripts/acp-mock-agent.ts +80 −7
  2. apps/server/scripts/acp-replay-agent.ts +285 −0
apps/server/scripts/acp-mock-agent.ts+80−7
@@ -23,6 +23,9 @@ const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1"; · 22 unmodified lines
23const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1";23const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1";
24const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT;24const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT;
25const promptDelayMs = Number(process.env.T3_ACP_PROMPT_DELAY_MS ?? "0");25const promptDelayMs = Number(process.env.T3_ACP_PROMPT_DELAY_MS ?? "0");
26const supportsSessionLifecycle = process.env.T3_ACP_SESSION_LIFECYCLE === "1";
27const advertisedAuthMethodId = process.env.T3_ACP_AUTH_METHOD_ID?.trim();
28const requiresAuthentication = process.env.T3_ACP_REQUIRE_AUTH === "1";
26const permissionOptionIds = {29const permissionOptionIds = {
27 allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once",30 allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once",
28 allowAlways: process.env.T3_ACP_ALLOW_ALWAYS_OPTION_ID ?? "allow-always",31 allowAlways: process.env.T3_ACP_ALLOW_ALWAYS_OPTION_ID ?? "allow-always",
@@ -36,6 +39,7 @@ let parameterizedModelPicker = false; · 7 unmodified lines
36let currentReasoning = "medium";39let currentReasoning = "medium";
37let currentContext = "272k";40let currentContext = "272k";
38let currentFast = false;41let currentFast = false;
42let authenticated = !requiresAuthentication;
39const cancelledSessions = new Set<string>();43const cancelledSessions = new Set<string>();
40 44
41function logExit(reason: string): void {45function logExit(reason: string): void {
@@ -268,19 +272,56 @@ const program = Effect.gen(function* () { · 226 unmodified lines
268 request.clientCapabilities?._meta?.parameterizedModelPicker === true;272 request.clientCapabilities?._meta?.parameterizedModelPicker === true;
269 return {273 return {
270 protocolVersion: 1,274 protocolVersion: 1,
271 agentCapabilities: { loadSession: true },275 agentCapabilities: {
276 loadSession: true,
277 ...(supportsSessionLifecycle
278 ? {
279 sessionCapabilities: {
280 list: {},
281 fork: {},
282 resume: {},
283 close: {},
284 },
285 }
286 : {}),
287 },
288 ...(advertisedAuthMethodId
289 ? {
290 authMethods: [
291 {
292 id: advertisedAuthMethodId,
293 name: "Mock agent authentication",
294 },
295 ],
296 }
297 : {}),
272 };298 };
273 }),299 }),
274 );300 );
275 301
276 yield* agent.handleAuthenticate(() => Effect.succeed({}));302 yield* agent.handleAuthenticate((request) =>
303 Effect.gen(function* () {
304 if (advertisedAuthMethodId && request.methodId !== advertisedAuthMethodId) {
305 return yield* AcpError.AcpRequestError.invalidParams(
306 `Unknown mock authentication method: ${request.methodId}`,
307 );
308 }
309 authenticated = true;
310 return {};
311 }),
312 );
277 313
278 yield* agent.handleCreateSession(() =>314 yield* agent.handleCreateSession(() =>
279 Effect.succeed({315 Effect.gen(function* () {
280 sessionId,316 if (!authenticated) {
281 modes: modeState(),317 return yield* AcpError.AcpRequestError.authRequired();
282 models: modelState(),318 }
283 configOptions: configOptions(),319 return {
320 sessionId,
321 modes: modeState(),
322 models: modelState(),
323 configOptions: configOptions(),
324 };
284 }),325 }),
285 );326 );
286 327
@@ -302,6 +343,38 @@ const program = Effect.gen(function* () { · 15 unmodified lines
302 ),343 ),
303 );344 );
304 345
346 yield* agent.handleListSessions((request) =>
347 Effect.succeed({
348 sessions: [
349 {
350 sessionId,
351 cwd: request.cwd ?? process.cwd(),
352 title: "Mock session",
353 updatedAt: "1970-01-01T00:00:00.000Z",
354 },
355 ],
356 }),
357 );
358
359 yield* agent.handleForkSession((request) =>
360 Effect.succeed({
361 sessionId: `${request.sessionId}-fork`,
362 modes: modeState(),
363 models: modelState(),
364 configOptions: configOptions(),
365 }),
366 );
367
368 yield* agent.handleResumeSession(() =>
369 Effect.succeed({
370 modes: modeState(),
371 models: modelState(),
372 configOptions: configOptions(),
373 }),
374 );
375
376 yield* agent.handleCloseSession(() => Effect.succeed({}));
377
305 yield* agent.handleSetSessionModel((request) =>378 yield* agent.handleSetSessionModel((request) =>
306 Effect.gen(function* () {379 Effect.gen(function* () {
307 if (!grokAcpModels.some((model) => model.modelId === request.modelId)) {380 if (!grokAcpModels.some((model) => model.modelId === request.modelId)) {
apps/server/scripts/acp-replay-agent.ts+285−0
@@ -0,0 +1,285 @@
1#!/usr/bin/env node
2// @effect-diagnostics nodeBuiltinImport:off
3import { writeFileSync } from "node:fs";
4import { createInterface } from "node:readline";
5
6interface ReplayEntry {
7 readonly type: "emit_inbound" | "expect_outbound" | "runtime_exit";
8 readonly label?: string;
9 readonly frame?: unknown;
10 readonly status?: "success" | "error" | "cancelled";
11 readonly error?: unknown;
12}
13
14interface ReplayTranscript {
15 readonly scenario: string;
16 readonly entries: ReadonlyArray<ReplayEntry>;
17}
18
19interface LogicalFrame {
20 readonly kind: "notification" | "request" | "response";
21 readonly method: string;
22 readonly params?: unknown;
23 readonly result?: unknown;
24 readonly error?: unknown;
25}
26
27interface JsonRpcMessage {
28 readonly jsonrpc?: string;
29 readonly id?: string | number | null;
30 readonly method?: string;
31 readonly params?: unknown;
32 readonly result?: unknown;
33 readonly error?: unknown;
34 readonly headers?: ReadonlyArray<unknown>;
35}
36
37const encodedTranscript = process.env.T3_ACP_REPLAY_TRANSCRIPT;
38const statusPath = process.env.T3_ACP_REPLAY_STATUS_PATH;
39const replayWorkspace = process.env.T3_ACP_REPLAY_WORKSPACE ?? process.cwd();
40
41if (encodedTranscript === undefined || statusPath === undefined) {
42 process.stderr.write("ACP replay requires transcript and status environment variables.\n");
43 process.exit(2);
44}
45
46const replayStatusPath = statusPath;
47const transcript = JSON.parse(
48 Buffer.from(encodedTranscript, "base64").toString("utf8"),
49) as ReplayTranscript;
50let cursor = 0;
51let stopped = false;
52let nextAgentRequestId = 1;
53const pendingClientRequestIds = new Map<string, string | number>();
54const pendingAgentRequestMethods = new Map<string, string>();
55
56function writeStatus(failure?: unknown): void {
57 writeFileSync(
58 replayStatusPath,
59 JSON.stringify({
60 scenario: transcript.scenario,
61 cursor,
62 total: transcript.entries.length,
63 ...(failure === undefined ? {} : { failure }),
64 }),
65 "utf8",
66 );
67}
68
69function stableStringify(value: unknown): string {
70 if (Array.isArray(value)) {
71 return `[${value.map(stableStringify).join(",")}]`;
72 }
73 if (typeof value === "object" && value !== null) {
74 const record = value as Record<string, unknown>;
75 return `{${Object.keys(record)
76 .toSorted()
77 .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`)
78 .join(",")}}`;
79 }
80 return JSON.stringify(value);
81}
82
83function expandExpectedString(value: string): string {
84 return value.replaceAll("<workspace>", replayWorkspace);
85}
86
87function matchesExpected(expected: unknown, actual: unknown): boolean {
88 if (expected === "<any>") return true;
89 if (typeof expected === "string" && typeof actual === "string") {
90 const expanded = expandExpectedString(expected);
91 if (!expanded.includes("<any>")) return expanded === actual;
92 const parts = expanded.split("<any>");
93 let offset = 0;
94 for (const part of parts) {
95 const index = actual.indexOf(part, offset);
96 if (index === -1) return false;
97 offset = index + part.length;
98 }
99 return true;
100 }
101 if (Array.isArray(expected)) {
102 return (
103 Array.isArray(actual) &&
104 expected.length === actual.length &&
105 expected.every((entry, index) => matchesExpected(entry, actual[index]))
106 );
107 }
108 if (typeof expected === "object" && expected !== null) {
109 if (typeof actual !== "object" || actual === null || Array.isArray(actual)) return false;
110 const expectedRecord = expected as Record<string, unknown>;
111 const actualRecord = actual as Record<string, unknown>;
112 const expectedKeys = Object.keys(expectedRecord).toSorted();
113 const actualKeys = Object.keys(actualRecord).toSorted();
114 return (
115 stableStringify(expectedKeys) === stableStringify(actualKeys) &&
116 expectedKeys.every((key) => matchesExpected(expectedRecord[key], actualRecord[key]))
117 );
118 }
119 return Object.is(expected, actual);
120}
121
122function stopWithFailure(detail: string, actual?: unknown): void {
123 if (stopped) return;
124 stopped = true;
125 const entry = transcript.entries[cursor];
126 const failure = {
127 detail,
128 cursor,
129 expected: entry,
130 ...(actual === undefined ? {} : { actual }),
131 };
132 writeStatus(failure);
133 process.stderr.write(`ACP replay mismatch: ${JSON.stringify(failure)}\n`);
134 process.exitCode = 1;
135 process.stdin.pause();
136}
137
138function advance(): void {
139 cursor += 1;
140 writeStatus();
141}
142
143function send(message: JsonRpcMessage): void {
144 process.stdout.write(`${JSON.stringify(message)}\n`);
145}
146
147function pendingClientRequestId(method: string): string | number | undefined {
148 return pendingClientRequestIds.get(method);
149}
150
151function logicalIncoming(message: JsonRpcMessage): LogicalFrame | undefined {
152 if (typeof message.method === "string") {
153 return {
154 kind:
155 message.id === undefined || message.id === null || message.id === ""
156 ? "notification"
157 : "request",
158 method: message.method,
159 ...(message.params === undefined ? {} : { params: message.params }),
160 };
161 }
162 if (message.id === undefined || message.id === null) return undefined;
163 const method = pendingAgentRequestMethods.get(String(message.id));
164 if (method === undefined) return undefined;
165 return {
166 kind: "response",
167 method,
168 ...(message.result === undefined ? {} : { result: message.result }),
169 ...(message.error === undefined ? {} : { error: message.error }),
170 };
171}
172
173function emitInbound(frame: LogicalFrame): void {
174 switch (frame.kind) {
175 case "notification":
176 send({
177 jsonrpc: "2.0",
178 method: frame.method,
179 ...(frame.params === undefined ? {} : { params: frame.params }),
180 });
181 return;
182 case "request": {
183 const id = nextAgentRequestId;
184 nextAgentRequestId += 1;
185 pendingAgentRequestMethods.set(String(id), frame.method);
186 send({
187 jsonrpc: "2.0",
188 id,
189 method: frame.method,
190 ...(frame.params === undefined ? {} : { params: frame.params }),
191 headers: [],
192 });
193 return;
194 }
195 case "response": {
196 const id = pendingClientRequestId(frame.method);
197 if (id === undefined) {
198 stopWithFailure(`No pending client request for ${frame.method}`, frame);
199 return;
200 }
201 pendingClientRequestIds.delete(frame.method);
202 send({
203 jsonrpc: "2.0",
204 id,
205 ...(frame.result === undefined ? {} : { result: frame.result }),
206 ...(frame.error === undefined ? {} : { error: frame.error }),
207 });
208 }
209 }
210}
211
212function flushInbound(): void {
213 while (!stopped) {
214 const entry = transcript.entries[cursor];
215 if (entry === undefined || entry.type === "expect_outbound") return;
216 if (entry.type === "runtime_exit") {
217 if (entry.status !== "success") {
218 stopWithFailure(`Recorded runtime exit was ${entry.status ?? "unknown"}`, entry.error);
219 return;
220 }
221 advance();
222 continue;
223 }
224 const frame = entry.frame as LogicalFrame;
225 if (
226 typeof frame !== "object" ||
227 frame === null ||
228 !["notification", "request", "response"].includes(frame.kind) ||
229 typeof frame.method !== "string"
230 ) {
231 stopWithFailure("Invalid emit_inbound logical ACP frame", entry.frame);
232 return;
233 }
234 emitInbound(frame);
235 if (stopped) return;
236 advance();
237 }
238}
239
240function handleMessage(message: JsonRpcMessage): void {
241 if (stopped) return;
242 const actual = logicalIncoming(message);
243 if (actual === undefined) {
244 stopWithFailure("Could not identify outbound ACP frame", message);
245 return;
246 }
247 const entry = transcript.entries[cursor];
248 if (entry?.type !== "expect_outbound" || !matchesExpected(entry.frame, actual)) {
249 if (actual.kind === "request" && message.id !== undefined && message.id !== null) {
250 send({
251 jsonrpc: "2.0",
252 id: message.id,
253 error: { code: -32603, message: "ACP replay frame mismatch" },
254 });
255 }
256 stopWithFailure("Unexpected outbound ACP frame", actual);
257 return;
258 }
259 if (actual.kind === "request" && message.id !== undefined && message.id !== null) {
260 pendingClientRequestIds.set(actual.method, message.id);
261 } else if (actual.kind === "response" && message.id !== undefined && message.id !== null) {
262 pendingAgentRequestMethods.delete(String(message.id));
263 }
264 advance();
265 flushInbound();
266}
267
268writeStatus();
269flushInbound();
270
271const input = createInterface({ input: process.stdin, crlfDelay: Infinity });
272input.on("line", (line) => {
273 if (stopped || line.trim().length === 0) return;
274 try {
275 handleMessage(JSON.parse(line) as JsonRpcMessage);
276 } catch (cause) {
277 stopWithFailure("Failed to decode outbound ACP JSON-RPC", String(cause));
278 }
279});
280
281input.on("close", () => {
282 if (!stopped && cursor !== transcript.entries.length) {
283 stopWithFailure("ACP replay input closed before transcript completion");
284 }
285});