authorgravatar for mwolson@member.fsf.orgMike Olson <mwolson@member.fsf.org> 2026-06-26 02:34:08-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-25 23:34:08-07:00
log52b04b947e3604e426386be53e6d20c6a4366fef
treefa346512aafa42b1ec4768f42d54d5e0e505273d
parente9ed70c5b935d8653c887c6a2f5c1f93013579d4
signature Signed by PGP key B5690EEEBB952194

fix(grok): Harden ACP resume with replay-idle load readiness (#3156)

Co-authored-by: Julius Marminge <julius0216@outlook.com> Co-authored-by: codex <codex@users.noreply.github.com>

20 files changed, 2802 insertions(+), 133 deletions(-)

apps/server/scripts/acp-mock-agent.ts+260-13
......@@ -19,6 +19,23 @@ const emitInterleavedAssistantToolCalls =
1919const emitGenericToolPlaceholders = process.env.T3_ACP_EMIT_GENERIC_TOOL_PLACEHOLDERS === "1";
2020const emitAskQuestion = process.env.T3_ACP_EMIT_ASK_QUESTION === "1";
2121const emitXAiAskUserQuestion = process.env.T3_ACP_EMIT_XAI_ASK_USER_QUESTION === "1";
22const emitXAiPromptCompleteThenHang = process.env.T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG === "1";
23const emitForeignSessionUpdates = process.env.T3_ACP_EMIT_FOREIGN_SESSION_UPDATES === "1";
24const hangPromptForever = process.env.T3_ACP_HANG_PROMPT_FOREVER === "1";
25const hangFirstPromptForever = process.env.T3_ACP_HANG_FIRST_PROMPT_FOREVER === "1";
26const emitLateUpdateAfterCancel = process.env.T3_ACP_EMIT_LATE_UPDATE_AFTER_CANCEL === "1";
27const omitXAiPromptCompleteStopReason =
28 process.env.T3_ACP_OMIT_XAI_PROMPT_COMPLETE_STOP_REASON === "1";
29const failLoadSession = process.env.T3_ACP_FAIL_LOAD_SESSION === "1";
30const emitLoadReplay = process.env.T3_ACP_EMIT_LOAD_REPLAY === "1";
31const hangLoadSessionAfterReplay = process.env.T3_ACP_HANG_LOAD_SESSION_AFTER_REPLAY === "1";
32const delayLoadSessionAfterReplay = process.env.T3_ACP_DELAY_LOAD_SESSION_AFTER_REPLAY === "1";
33const loadSessionDelayMs = Number(process.env.T3_ACP_LOAD_SESSION_DELAY_MS ?? "5000");
34const emitStaleXAiPromptCompleteBeforeSecondHang =
35 process.env.T3_ACP_EMIT_STALE_XAI_PROMPT_COMPLETE_BEFORE_SECOND_HANG === "1";
36const emitOverlappingXAiPromptCompleteOutOfOrder =
37 process.env.T3_ACP_EMIT_OVERLAPPING_XAI_PROMPT_COMPLETE_OUT_OF_ORDER === "1";
38const failPrompt = process.env.T3_ACP_FAIL_PROMPT === "1";
2239const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1";
2340const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1";
2441const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT;
......@@ -36,8 +53,21 @@ let parameterizedModelPicker = false;
3653let currentReasoning = "medium";
3754let currentContext = "272k";
3855let currentFast = false;
56let promptCount = 0;
57let overlappingFirstPromptId: string | undefined;
3958const cancelledSessions = new Set<string>();
4059
60function promptIdFromRequestMeta(
61 request: Pick<AcpSchema.PromptRequest, "_meta">,
62): string | undefined {
63 const meta = request._meta;
64 if (meta === null || typeof meta !== "object") {
65 return undefined;
66 }
67 const promptId = meta.promptId ?? meta.requestId;
68 return typeof promptId === "string" && promptId.length > 0 ? promptId : undefined;
69}
70
4171function logExit(reason: string): void {
4272 if (!exitLogPath) {
4373 return;
......@@ -45,6 +75,10 @@ function logExit(reason: string): void {
4575 NodeFS.appendFileSync(exitLogPath, `${reason}\n`, "utf8");
4676}
4777
78function writeJsonRpcNotification(method: string, params: unknown): void {
79 process.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", method, params })}\n`);
80}
81
4882process.once("SIGTERM", () => {
4983 logExit("SIGTERM");
5084 process.exit(0);
......@@ -284,22 +318,66 @@ const program = Effect.gen(function* () {
284318 }),
285319 );
286320
321 const emitLoadReplayNotifications = (requestedSessionId: string) => {
322 writeJsonRpcNotification("session/update", {
323 _meta: { isReplay: true },
324 sessionId: requestedSessionId,
325 update: {
326 sessionUpdate: "tool_call",
327 toolCallId: "replay-tool-1",
328 title: "Replay tool",
329 kind: "search",
330 status: "completed",
331 },
332 });
333 writeJsonRpcNotification("session/update", {
334 _meta: { isReplay: true },
335 sessionId: requestedSessionId,
336 update: {
337 sessionUpdate: "agent_message_chunk",
338 content: { type: "text", text: "replayed assistant text" },
339 },
340 });
341 };
342
287343 yield* agent.handleLoadSession((request) =>
288 agent.client
289 .sessionUpdate({
290 sessionId: String(request.sessionId ?? sessionId),
344 Effect.gen(function* () {
345 const requestedSessionId = String(request.sessionId ?? sessionId);
346 if (failLoadSession) {
347 return yield* AcpError.AcpRequestError.internalError("Mock load session failure");
348 }
349 if (hangLoadSessionAfterReplay || delayLoadSessionAfterReplay) {
350 emitLoadReplayNotifications(requestedSessionId);
351 yield* agent.client.sessionUpdate({
352 sessionId: requestedSessionId,
353 update: {
354 sessionUpdate: "user_message_chunk",
355 content: { type: "text", text: "replay-tail" },
356 },
357 });
358 yield* Effect.sleep(loadSessionDelayMs);
359 return {
360 modes: modeState(),
361 models: modelState(),
362 configOptions: configOptions(),
363 };
364 }
365 if (emitLoadReplay) {
366 emitLoadReplayNotifications(requestedSessionId);
367 }
368 yield* agent.client.sessionUpdate({
369 sessionId: requestedSessionId,
291370 update: {
292371 sessionUpdate: "user_message_chunk",
293372 content: { type: "text", text: "replay" },
294373 },
295 })
296 .pipe(
297 Effect.as({
298 modes: modeState(),
299 models: modelState(),
300 configOptions: configOptions(),
301 }),
302 ),
374 });
375 return {
376 modes: modeState(),
377 models: modelState(),
378 configOptions: configOptions(),
379 };
380 }),
303381 );
304382
305383 yield* agent.handleSetSessionModel((request) =>
......@@ -356,19 +434,152 @@ const program = Effect.gen(function* () {
356434 );
357435
358436 yield* agent.handleCancel(({ sessionId }) =>
359 Effect.sync(() => {
360 cancelledSessions.add(String(sessionId ?? "mock-session-1"));
437 Effect.gen(function* () {
438 const cancelledSessionId = String(sessionId ?? "mock-session-1");
439 cancelledSessions.add(cancelledSessionId);
440 if (emitLateUpdateAfterCancel) {
441 yield* Effect.sleep("50 millis");
442 yield* Effect.sync(() => {
443 writeJsonRpcNotification("session/update", {
444 sessionId: cancelledSessionId,
445 update: {
446 sessionUpdate: "agent_message_chunk",
447 content: { type: "text", text: "late after cancel" },
448 },
449 });
450 });
451 }
361452 }),
362453 );
363454
364455 yield* agent.handlePrompt((request) =>
365456 Effect.gen(function* () {
366457 const requestedSessionId = String(request.sessionId ?? sessionId);
458 promptCount += 1;
367459
368460 if (Number.isFinite(promptDelayMs) && promptDelayMs > 0) {
369461 yield* Effect.sleep(`${promptDelayMs} millis`);
370462 }
371463
464 if (failPrompt) {
465 return yield* AcpError.AcpRequestError.internalError("Mock prompt failure");
466 }
467
468 if (emitStaleXAiPromptCompleteBeforeSecondHang && promptCount === 1) {
469 return {
470 stopReason: "end_turn",
471 _meta: {
472 promptId: "mock-stale-xai-prompt-1",
473 requestId: "mock-stale-xai-prompt-1",
474 },
475 };
476 }
477
478 if (emitStaleXAiPromptCompleteBeforeSecondHang && promptCount === 2) {
479 const currentPromptId = promptIdFromRequestMeta(request) ?? "mock-current-xai-prompt-2";
480 writeJsonRpcNotification("_x.ai/session/prompt_complete", {
481 sessionId: requestedSessionId,
482 promptId: "mock-stale-xai-prompt-1",
483 stopReason: "end_turn",
484 agentResult: null,
485 });
486
487 writeJsonRpcNotification("_x.ai/session/prompt_complete", {
488 sessionId: requestedSessionId,
489 promptId: currentPromptId,
490 stopReason: "end_turn",
491 agentResult: null,
492 });
493
494 return yield* Effect.never;
495 }
496
497 if (emitOverlappingXAiPromptCompleteOutOfOrder && promptCount === 1) {
498 overlappingFirstPromptId = promptIdFromRequestMeta(request);
499 return yield* Effect.never;
500 }
501
502 if (emitOverlappingXAiPromptCompleteOutOfOrder && promptCount === 2) {
503 const secondPromptId = promptIdFromRequestMeta(request);
504 if (overlappingFirstPromptId !== undefined && secondPromptId !== undefined) {
505 writeJsonRpcNotification("_x.ai/session/prompt_complete", {
506 sessionId: requestedSessionId,
507 promptId: secondPromptId,
508 stopReason: "end_turn",
509 agentResult: null,
510 });
511 writeJsonRpcNotification("_x.ai/session/prompt_complete", {
512 sessionId: requestedSessionId,
513 promptId: overlappingFirstPromptId,
514 stopReason: "end_turn",
515 agentResult: null,
516 });
517 }
518 return yield* Effect.never;
519 }
520
521 if (hangPromptForever || (hangFirstPromptForever && promptCount === 1)) {
522 return yield* Effect.never;
523 }
524
525 if (emitXAiPromptCompleteThenHang) {
526 writeJsonRpcNotification("session/update", {
527 sessionId: requestedSessionId,
528 update: {
529 sessionUpdate: "agent_message_chunk",
530 content: { type: "text", text: "hello from " },
531 },
532 });
533
534 if (emitForeignSessionUpdates) {
535 writeJsonRpcNotification("session/update", {
536 sessionId: "mock-child-session-1",
537 update: {
538 sessionUpdate: "agent_message_chunk",
539 content: { type: "text", text: "child before completion" },
540 },
541 });
542 }
543
544 writeJsonRpcNotification("_x.ai/session/prompt_complete", {
545 sessionId: requestedSessionId,
546 promptId: promptIdFromRequestMeta(request) ?? "mock-xai-prompt-1",
547 ...(omitXAiPromptCompleteStopReason ? {} : { stopReason: "end_turn" }),
548 agentResult: null,
549 });
550
551 if (emitForeignSessionUpdates) {
552 writeJsonRpcNotification("session/update", {
553 sessionId: "mock-child-session-1",
554 update: {
555 sessionUpdate: "tool_call",
556 toolCallId: "child-tool-call-1",
557 title: "Child-only tool",
558 kind: "other",
559 status: "pending",
560 rawInput: {},
561 },
562 });
563 writeJsonRpcNotification("session/update", {
564 sessionId: "mock-child-session-1",
565 update: {
566 sessionUpdate: "agent_message_chunk",
567 content: { type: "text", text: "child after completion" },
568 },
569 });
570 }
571
572 writeJsonRpcNotification("session/update", {
573 sessionId: requestedSessionId,
574 update: {
575 sessionUpdate: "agent_message_chunk",
576 content: { type: "text", text: "mock" },
577 },
578 });
579
580 return yield* Effect.never;
581 }
582
372583 if (emitInterleavedAssistantToolCalls) {
373584 const toolCallId = "tool-call-1";
374585
......@@ -599,6 +810,42 @@ const program = Effect.gen(function* () {
599810 return { stopReason: "end_turn" };
600811 }
601812
813 if (emitForeignSessionUpdates) {
814 yield* agent.client.sessionUpdate({
815 sessionId: requestedSessionId,
816 update: {
817 sessionUpdate: "agent_message_chunk",
818 content: { type: "text", text: "root before child" },
819 },
820 });
821 yield* agent.client.sessionUpdate({
822 sessionId: "mock-child-session-1",
823 update: {
824 sessionUpdate: "agent_message_chunk",
825 content: { type: "text", text: "child content" },
826 },
827 });
828 yield* agent.client.sessionUpdate({
829 sessionId: "mock-child-session-1",
830 update: {
831 sessionUpdate: "tool_call",
832 toolCallId: "child-tool-call-1",
833 title: "Child-only tool",
834 kind: "other",
835 status: "pending",
836 rawInput: {},
837 },
838 });
839 yield* agent.client.sessionUpdate({
840 sessionId: requestedSessionId,
841 update: {
842 sessionUpdate: "agent_message_chunk",
843 content: { type: "text", text: " root after child" },
844 },
845 });
846 return { stopReason: "end_turn" };
847 }
848
602849 yield* agent.client.sessionUpdate({
603850 sessionId: requestedSessionId,
604851 update: {
apps/server/src/provider/Layers/CursorAdapter.test.ts+21-1
......@@ -113,6 +113,23 @@ async function waitForFileContent(filePath: string, attempts = 40) {
113113 throw new Error(`Timed out waiting for file content at ${filePath}`);
114114}
115115
116function waitForJsonLogMatch(
117 filePath: string,
118 predicate: (entry: Record<string, unknown>) => boolean,
119 attempts = 40,
120) {
121 return Effect.gen(function* () {
122 for (let attempt = 0; attempt < attempts; attempt += 1) {
123 const requests = yield* Effect.promise(() => readJsonLines(filePath));
124 if (requests.some(predicate)) {
125 return requests;
126 }
127 yield* Effect.yieldNow;
128 }
129 return yield* Effect.promise(() => readJsonLines(filePath));
130 });
131}
132
116133// Tests mutate `ServerSettingsService` mid-flight (e.g. setting
117134// `providers.cursor.binaryPath` to a mock ACP wrapper). The adapter
118135// captures `cursorSettings` once at construction, so without a resolver
......@@ -1004,7 +1021,10 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => {
10041021 assert.equal(turnCompleted.payload.stopReason, "cancelled");
10051022 }
10061023
1007 const requests = yield* Effect.promise(() => readJsonLines(requestLogPath));
1024 const requests = yield* waitForJsonLogMatch(
1025 requestLogPath,
1026 (entry) => entry.method === "session/cancel",
1027 );
10081028 assert.isTrue(requests.some((entry) => entry.method === "session/cancel"));
10091029 assert.isTrue(
10101030 requests.some(
apps/server/src/provider/Layers/CursorAdapter.ts+3
......@@ -785,6 +785,9 @@ export function makeCursorAdapter(
785785 Stream.mapEffect(acp.getEvents(), (event) =>
786786 Effect.gen(function* () {
787787 switch (event._tag) {
788 case "EventStreamBarrier":
789 yield* Deferred.succeed(event.acknowledge, undefined);
790 return;
788791 case "ModeChanged":
789792 return;
790793 case "AssistantItemStarted":
apps/server/src/provider/Layers/GrokAdapter.test.ts+825-4
......@@ -10,20 +10,23 @@ import * as Deferred from "effect/Deferred";
1010import * as Effect from "effect/Effect";
1111import * as Fiber from "effect/Fiber";
1212import * as Layer from "effect/Layer";
13import * as Ref from "effect/Ref";
1314import * as Schema from "effect/Schema";
1415import * as Stream from "effect/Stream";
16import * as TestClock from "effect/testing/TestClock";
1517
1618import {
1719 ApprovalRequestId,
1820 GrokSettings,
1921 ProviderDriverKind,
20 ThreadId,
2122 ProviderInstanceId,
23 ThreadId,
24 TurnId,
2225 type ProviderRuntimeEvent,
2326} from "@t3tools/contracts";
2427
2528import { ServerConfig } from "../../config.ts";
26import { makeGrokAdapter } from "./GrokAdapter.ts";
29import { grokPromptSettlementBelongsToContext, makeGrokAdapter } from "./GrokAdapter.ts";
2730const decodeGrokSettings = Schema.decodeSync(GrokSettings);
2831
2932const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url));
......@@ -45,7 +48,11 @@ exec ${JSON.stringify(mockAgentCommand)} ${JSON.stringify(mockAgentPath)} "$@"
4548 return wrapperPath;
4649}
4750
48function waitForFileContent(filePath: string, attempts = 40): Effect.Effect<string> {
51function waitForFileContent(
52 filePath: string,
53 attempts = 40,
54 expectedContent?: string,
55): Effect.Effect<string> {
4956 const readAttempt = (remainingAttempts: number): Effect.Effect<string> =>
5057 Effect.gen(function* () {
5158 if (remainingAttempts <= 0) {
......@@ -54,7 +61,10 @@ function waitForFileContent(filePath: string, attempts = 40): Effect.Effect<stri
5461 const raw = yield* Effect.tryPromise(() => NodeFSP.readFile(filePath, "utf8")).pipe(
5562 Effect.orElseSucceed(() => ""),
5663 );
57 if (raw.trim().length > 0) {
64 if (
65 raw.trim().length > 0 &&
66 (expectedContent === undefined || raw.includes(expectedContent))
67 ) {
5868 return raw;
5969 }
6070 yield* Effect.sleep("25 millis");
......@@ -79,6 +89,39 @@ const grokAdapterTestLayer = ServerConfig.layerTest(process.cwd(), {
7989const makeTestAdapter = (binaryPath: string, options?: Parameters<typeof makeGrokAdapter>[1]) =>
8090 makeGrokAdapter(decodeGrokSettings({ binaryPath }), options).pipe(Effect.orDie);
8191
92it("requires a settlement to match the live Grok turn", () => {
93 const staleTurnId = TurnId.make("stale-turn");
94 const replacementTurnId = TurnId.make("replacement-turn");
95
96 assert.isFalse(
97 grokPromptSettlementBelongsToContext({
98 liveAcpSessionId: "session-1",
99 expectedAcpSessionId: "session-1",
100 liveActiveTurnId: replacementTurnId,
101 liveSessionActiveTurnId: replacementTurnId,
102 turnId: staleTurnId,
103 }),
104 );
105 assert.isFalse(
106 grokPromptSettlementBelongsToContext({
107 liveAcpSessionId: "replacement-session",
108 expectedAcpSessionId: "stale-session",
109 liveActiveTurnId: staleTurnId,
110 liveSessionActiveTurnId: staleTurnId,
111 turnId: staleTurnId,
112 }),
113 );
114 assert.isTrue(
115 grokPromptSettlementBelongsToContext({
116 liveAcpSessionId: "session-1",
117 expectedAcpSessionId: "session-1",
118 liveActiveTurnId: staleTurnId,
119 liveSessionActiveTurnId: staleTurnId,
120 turnId: staleTurnId,
121 }),
122 );
123});
124
82125it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
83126 it.effect("starts a session and maps mock ACP prompt flow to runtime events", () =>
84127 Effect.gen(function* () {
......@@ -175,6 +218,783 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
175218 }),
176219 );
177220
221 it.effect("reports a Grok session running only while the prompt is in flight", () =>
222 Effect.gen(function* () {
223 const threadId = ThreadId.make("grok-session-ready-after-prompt");
224 const wrapperPath = yield* Effect.promise(() =>
225 makeMockGrokWrapper({
226 T3_ACP_EMIT_TOOL_CALLS: "1",
227 }),
228 );
229 const adapter = yield* makeTestAdapter(wrapperPath);
230 const requestOpened =
231 yield* Deferred.make<Extract<ProviderRuntimeEvent, { type: "request.opened" }>>();
232 const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
233 event.type === "request.opened"
234 ? Deferred.succeed(requestOpened, event).pipe(Effect.ignore)
235 : Effect.void,
236 ).pipe(Effect.forkChild);
237
238 yield* adapter.startSession({
239 threadId,
240 provider: ProviderDriverKind.make("grok"),
241 cwd: process.cwd(),
242 runtimeMode: "approval-required",
243 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
244 });
245
246 const sendTurnFiber = yield* adapter
247 .sendTurn({ threadId, input: "check lifecycle", attachments: [] })
248 .pipe(Effect.forkChild);
249 const requestOpenedEvent = yield* Deferred.await(requestOpened);
250
251 const runningSessions = yield* adapter.listSessions();
252 const runningSession = runningSessions.find((session) => session.threadId === threadId);
253 assert.equal(runningSession?.status, "running");
254 assert.isDefined(runningSession?.activeTurnId);
255
256 yield* adapter.respondToRequest(
257 threadId,
258 ApprovalRequestId.make(String(requestOpenedEvent.requestId)),
259 "accept",
260 );
261 yield* Fiber.join(sendTurnFiber);
262
263 const readySessions = yield* adapter.listSessions();
264 const readySession = readySessions.find((session) => session.threadId === threadId);
265 assert.equal(readySession?.status, "ready");
266 assert.isUndefined(readySession?.activeTurnId);
267
268 yield* Fiber.interrupt(eventsFiber);
269 yield* adapter.stopSession(threadId);
270 }),
271 );
272
273 it.effect("restores ready without completing an unstarted turn when preparation fails", () =>
274 Effect.gen(function* () {
275 const threadId = ThreadId.make("grok-preparation-failure-while-connecting");
276 const wrapperPath = yield* Effect.promise(() => makeMockGrokWrapper());
277 const adapter = yield* makeTestAdapter(wrapperPath);
278
279 const runtimeEvents: ProviderRuntimeEvent[] = [];
280 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
281 Effect.sync(() => {
282 runtimeEvents.push(event);
283 }),
284 ).pipe(Effect.forkChild);
285
286 yield* adapter.startSession({
287 threadId,
288 provider: ProviderDriverKind.make("grok"),
289 cwd: process.cwd(),
290 runtimeMode: "full-access",
291 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
292 });
293
294 const error = yield* Effect.flip(
295 adapter.sendTurn({
296 threadId,
297 input: "prepare invalid attachment",
298 attachments: [
299 {
300 type: "image",
301 id: "missing-image",
302 name: "missing.png",
303 mimeType: "image/png",
304 sizeBytes: 1,
305 },
306 ],
307 }),
308 );
309 for (let yieldAttempt = 0; yieldAttempt < 4; yieldAttempt += 1) {
310 yield* Effect.yieldNow;
311 }
312
313 const turnCompletedEvent = runtimeEvents.find(
314 (event): event is Extract<ProviderRuntimeEvent, { type: "turn.completed" }> =>
315 event.type === "turn.completed",
316 );
317 const readySessions = yield* adapter.listSessions();
318 const readySession = readySessions.find((session) => session.threadId === threadId);
319
320 assert.equal(error._tag, "ProviderAdapterRequestError");
321 assert.isUndefined(turnCompletedEvent);
322 assert.equal(readySession?.status, "ready");
323 assert.isUndefined(readySession?.activeTurnId);
324
325 yield* Fiber.interrupt(runtimeEventsFiber);
326 yield* adapter.stopSession(threadId);
327 }),
328 );
329
330 it.effect("completes a Grok turn from xAI prompt completion when the prompt RPC hangs", () =>
331 Effect.gen(function* () {
332 const threadId = ThreadId.make("grok-xai-prompt-complete-fallback");
333 const wrapperPath = yield* Effect.promise(() =>
334 makeMockGrokWrapper({
335 T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG: "1",
336 T3_ACP_EMIT_FOREIGN_SESSION_UPDATES: "1",
337 }),
338 );
339 const adapter = yield* makeTestAdapter(wrapperPath);
340
341 const runtimeEvents: ProviderRuntimeEvent[] = [];
342 const turnCompleted = yield* Deferred.make<void>();
343 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
344 Effect.sync(() => {
345 runtimeEvents.push(event);
346 }).pipe(
347 Effect.andThen(
348 event.type === "turn.completed"
349 ? Deferred.succeed(turnCompleted, undefined)
350 : Effect.void,
351 ),
352 ),
353 ).pipe(Effect.forkChild);
354
355 yield* adapter.startSession({
356 threadId,
357 provider: ProviderDriverKind.make("grok"),
358 cwd: process.cwd(),
359 runtimeMode: "full-access",
360 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
361 });
362
363 const sendTurnResult = yield* adapter.sendTurn({
364 threadId,
365 input: "exercise fallback",
366 attachments: [],
367 });
368
369 yield* Deferred.await(turnCompleted);
370 for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) {
371 yield* Effect.yieldNow;
372 }
373 const readySessions = yield* adapter.listSessions();
374 const readySession = readySessions.find((session) => session.threadId === threadId);
375 const turnCompletedEvent = runtimeEvents.find(
376 (event): event is Extract<ProviderRuntimeEvent, { type: "turn.completed" }> =>
377 event.type === "turn.completed",
378 );
379 const eventTypes = runtimeEvents.map((event) => event.type);
380 const content = runtimeEvents
381 .filter(
382 (event): event is Extract<ProviderRuntimeEvent, { type: "content.delta" }> =>
383 event.type === "content.delta" && String(event.threadId) === String(threadId),
384 )
385 .map((event) => event.payload.delta)
386 .join("");
387 const terminalIndex = runtimeEvents.findIndex(
388 (event) => event.type === "turn.completed" && String(event.threadId) === String(threadId),
389 );
390 const turnOutputTypes = new Set([
391 "content.delta",
392 "item.started",
393 "item.updated",
394 "item.completed",
395 "turn.plan.updated",
396 ]);
397 const outputAfterTerminal = runtimeEvents
398 .slice(terminalIndex + 1)
399 .filter(
400 (event) => String(event.threadId) === String(threadId) && turnOutputTypes.has(event.type),
401 );
402 const toolTitles = runtimeEvents.flatMap((event) =>
403 event.type === "item.updated" && event.payload.title ? [event.payload.title] : [],
404 );
405
406 assert.equal(sendTurnResult.threadId, threadId);
407 assert.include(eventTypes, "turn.completed");
408 assert.equal(content, "hello from mock");
409 assert.isAtLeast(terminalIndex, 0);
410 assert.deepEqual(outputAfterTerminal, []);
411 assert.notInclude(toolTitles, "Child-only tool");
412 assert.equal(turnCompletedEvent?.payload.stopReason, "end_turn");
413 assert.equal(readySession?.status, "ready");
414 assert.isUndefined(readySession?.activeTurnId);
415
416 yield* Fiber.interrupt(runtimeEventsFiber);
417 yield* adapter.stopSession(threadId);
418 }),
419 );
420
421 it.effect("retains turn transcript when sendTurn is interrupted after prompt success", () =>
422 Effect.gen(function* () {
423 const threadId = ThreadId.make("grok-send-turn-interrupt-after-prompt");
424 const wrapperPath = yield* Effect.promise(() =>
425 makeMockGrokWrapper({
426 T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG: "1",
427 }),
428 );
429 const adapter = yield* makeTestAdapter(wrapperPath);
430 const contentDelta = yield* Deferred.make<void>();
431 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
432 event.type === "content.delta" ? Deferred.succeed(contentDelta, undefined) : Effect.void,
433 ).pipe(Effect.forkChild);
434
435 yield* adapter.startSession({
436 threadId,
437 provider: ProviderDriverKind.make("grok"),
438 cwd: process.cwd(),
439 runtimeMode: "full-access",
440 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
441 });
442
443 const sendTurnFiber = yield* adapter
444 .sendTurn({
445 threadId,
446 input: "interrupt after prompt",
447 attachments: [],
448 })
449 .pipe(Effect.forkChild);
450
451 yield* Deferred.await(contentDelta);
452 for (let yieldAttempt = 0; yieldAttempt < 6; yieldAttempt += 1) {
453 yield* Effect.yieldNow;
454 }
455 yield* Fiber.interrupt(sendTurnFiber);
456 for (let yieldAttempt = 0; yieldAttempt < 4; yieldAttempt += 1) {
457 yield* Effect.yieldNow;
458 }
459
460 const snapshot = yield* adapter.readThread(threadId);
461 assert.equal(snapshot.turns.length, 1);
462 assert.equal(snapshot.turns[0]?.items.length, 1);
463
464 yield* Fiber.interrupt(runtimeEventsFiber);
465 yield* adapter.stopSession(threadId);
466 }),
467 );
468
469 it.effect("does not report a synthetic stop reason when xAI omits one", () =>
470 Effect.gen(function* () {
471 const threadId = ThreadId.make("grok-xai-prompt-complete-missing-stop-reason");
472 const wrapperPath = yield* Effect.promise(() =>
473 makeMockGrokWrapper({
474 T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG: "1",
475 T3_ACP_OMIT_XAI_PROMPT_COMPLETE_STOP_REASON: "1",
476 }),
477 );
478 const adapter = yield* makeTestAdapter(wrapperPath);
479
480 const runtimeEvents: ProviderRuntimeEvent[] = [];
481 const turnCompleted = yield* Deferred.make<void>();
482 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
483 Effect.sync(() => {
484 runtimeEvents.push(event);
485 }).pipe(
486 Effect.andThen(
487 event.type === "turn.completed"
488 ? Deferred.succeed(turnCompleted, undefined)
489 : Effect.void,
490 ),
491 ),
492 ).pipe(Effect.forkChild);
493
494 yield* adapter.startSession({
495 threadId,
496 provider: ProviderDriverKind.make("grok"),
497 cwd: process.cwd(),
498 runtimeMode: "full-access",
499 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
500 });
501
502 yield* adapter.sendTurn({
503 threadId,
504 input: "exercise missing stop reason",
505 attachments: [],
506 });
507
508 yield* Deferred.await(turnCompleted);
509 const turnCompletedEvent = runtimeEvents.find(
510 (event): event is Extract<ProviderRuntimeEvent, { type: "turn.completed" }> =>
511 event.type === "turn.completed",
512 );
513
514 assert.equal(turnCompletedEvent?.payload.state, "completed");
515 assert.isNull(turnCompletedEvent?.payload.stopReason);
516
517 yield* Fiber.interrupt(runtimeEventsFiber);
518 yield* adapter.stopSession(threadId);
519 }),
520 );
521
522 it.effect("lets Stop unblock a fully silent Grok prompt and accept a follow-up turn", () =>
523 Effect.gen(function* () {
524 const threadId = ThreadId.make("grok-stop-after-full-silence");
525 const wrapperPath = yield* Effect.promise(() =>
526 makeMockGrokWrapper({
527 T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1",
528 }),
529 );
530 const adapter = yield* makeTestAdapter(wrapperPath);
531
532 const runtimeEvents: ProviderRuntimeEvent[] = [];
533 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
534 Effect.sync(() => {
535 runtimeEvents.push(event);
536 }),
537 ).pipe(Effect.forkChild);
538
539 yield* adapter.startSession({
540 threadId,
541 provider: ProviderDriverKind.make("grok"),
542 cwd: process.cwd(),
543 runtimeMode: "full-access",
544 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
545 });
546
547 yield* Effect.gen(function* () {
548 yield* Effect.sleep("500 millis");
549 yield* adapter.interruptTurn(threadId);
550 }).pipe(Effect.forkChild({ startImmediately: true }));
551
552 yield* adapter.sendTurn({
553 threadId,
554 input: "hang forever",
555 attachments: [],
556 });
557 for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) {
558 yield* Effect.yieldNow;
559 }
560
561 const cancelledEvents = runtimeEvents.filter(
562 (event): event is Extract<ProviderRuntimeEvent, { type: "turn.completed" }> =>
563 event.type === "turn.completed" && String(event.threadId) === String(threadId),
564 );
565 const readySessions = yield* adapter.listSessions();
566 const readySession = readySessions.find((session) => session.threadId === threadId);
567
568 assert.lengthOf(cancelledEvents, 1);
569 assert.equal(cancelledEvents[0]?.payload.state, "cancelled");
570 assert.equal(readySession?.status, "ready");
571 assert.isUndefined(readySession?.activeTurnId);
572
573 const followUpEventsBefore = runtimeEvents.length;
574 yield* adapter.sendTurn({
575 threadId,
576 input: "continue after stop",
577 attachments: [],
578 });
579 for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) {
580 yield* Effect.yieldNow;
581 }
582
583 const followUpCompletedEvents = runtimeEvents
584 .slice(followUpEventsBefore)
585 .filter(
586 (event): event is Extract<ProviderRuntimeEvent, { type: "turn.completed" }> =>
587 event.type === "turn.completed" && String(event.threadId) === String(threadId),
588 );
589 assert.lengthOf(followUpCompletedEvents, 1);
590 assert.equal(followUpCompletedEvents[0]?.payload.state, "completed");
591
592 yield* Fiber.interrupt(runtimeEventsFiber);
593 yield* adapter.stopSession(threadId);
594 }).pipe(TestClock.withLive),
595 );
596
597 it.effect("does not let a cancelled prompt settlement consume the follow-up prompt slot", () =>
598 Effect.gen(function* () {
599 const threadId = ThreadId.make("grok-cancelled-settlement-before-follow-up");
600 const tempDir = yield* Effect.promise(() =>
601 NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "grok-acp-cancel-race-")),
602 );
603 const requestLogPath = NodePath.join(tempDir, "requests.ndjson");
604 const wrapperPath = yield* Effect.promise(() =>
605 makeMockGrokWrapper({
606 T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1",
607 T3_ACP_REQUEST_LOG_PATH: requestLogPath,
608 }),
609 );
610 const adapter = yield* makeTestAdapter(wrapperPath);
611
612 const runtimeEvents: ProviderRuntimeEvent[] = [];
613 const firstTurnStarted = yield* Deferred.make<TurnId>();
614 const twoTurnsCompleted = yield* Deferred.make<void>();
615 const completedCountRef = yield* Ref.make(0);
616 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
617 Effect.gen(function* () {
618 runtimeEvents.push(event);
619 if (String(event.threadId) !== String(threadId)) {
620 return;
621 }
622 if (event.type === "turn.started" && event.turnId !== undefined) {
623 yield* Deferred.succeed(firstTurnStarted, event.turnId).pipe(Effect.ignore);
624 return;
625 }
626 if (event.type !== "turn.completed") {
627 return;
628 }
629 const completedCount = yield* Ref.updateAndGet(completedCountRef, (count) => count + 1);
630 if (completedCount === 2) {
631 yield* Deferred.succeed(twoTurnsCompleted, undefined);
632 }
633 }),
634 ).pipe(Effect.forkChild);
635
636 yield* adapter.startSession({
637 threadId,
638 provider: ProviderDriverKind.make("grok"),
639 cwd: process.cwd(),
640 runtimeMode: "full-access",
641 });
642
643 const firstSendTurnFiber = yield* adapter
644 .sendTurn({ threadId, input: "cancel this prompt", attachments: [] })
645 .pipe(Effect.forkChild);
646 const firstTurnId = yield* Deferred.await(firstTurnStarted).pipe(Effect.timeout("2 seconds"));
647 yield* waitForFileContent(requestLogPath, 80, '"method":"session/prompt"');
648
649 yield* adapter.interruptTurn(threadId, firstTurnId).pipe(Effect.timeout("2 seconds"));
650 const followUp = yield* adapter
651 .sendTurn({ threadId, input: "complete the follow-up", attachments: [] })
652 .pipe(Effect.timeout("2 seconds"));
653 yield* Fiber.join(firstSendTurnFiber).pipe(Effect.timeout("2 seconds"));
654 yield* Deferred.await(twoTurnsCompleted).pipe(Effect.timeout("2 seconds"));
655
656 const turnCompletedEvents = runtimeEvents.filter(
657 (event): event is Extract<ProviderRuntimeEvent, { type: "turn.completed" }> =>
658 event.type === "turn.completed" && String(event.threadId) === String(threadId),
659 );
660 const readySessions = yield* adapter.listSessions();
661 const readySession = readySessions.find((session) => session.threadId === threadId);
662
663 assert.notEqual(String(followUp.turnId), String(firstTurnId));
664 assert.deepEqual(
665 turnCompletedEvents.map((event) => [String(event.turnId), event.payload.state]),
666 [
667 [String(firstTurnId), "cancelled"],
668 [String(followUp.turnId), "completed"],
669 ],
670 );
671 assert.equal(readySession?.status, "ready");
672 assert.isUndefined(readySession?.activeTurnId);
673
674 yield* Fiber.interrupt(runtimeEventsFiber);
675 yield* adapter.stopSession(threadId);
676 }).pipe(TestClock.withLive),
677 );
678
679 it.effect("drops late ACP notifications after a turn is cancelled", () =>
680 Effect.gen(function* () {
681 const threadId = ThreadId.make("grok-drop-late-cancelled-notifications");
682 const wrapperPath = yield* Effect.promise(() =>
683 makeMockGrokWrapper({
684 T3_ACP_HANG_PROMPT_FOREVER: "1",
685 T3_ACP_EMIT_LATE_UPDATE_AFTER_CANCEL: "1",
686 }),
687 );
688 const lateNativeUpdate = yield* Deferred.make<void>();
689 const adapter = yield* makeTestAdapter(wrapperPath, {
690 nativeEventLogger: {
691 filePath: "memory://grok-cancelled-native-events",
692 write: (record: unknown) =>
693 JSON.stringify(record).includes("late after cancel")
694 ? Deferred.succeed(lateNativeUpdate, undefined).pipe(Effect.asVoid)
695 : Effect.void,
696 close: () => Effect.void,
697 },
698 });
699
700 const runtimeEvents: ProviderRuntimeEvent[] = [];
701 const turnStarted = yield* Deferred.make<TurnId>();
702 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
703 Effect.sync(() => {
704 runtimeEvents.push(event);
705 }).pipe(
706 Effect.andThen(
707 event.type === "turn.started" &&
708 event.turnId !== undefined &&
709 String(event.threadId) === String(threadId)
710 ? Deferred.succeed(turnStarted, event.turnId).pipe(Effect.asVoid)
711 : Effect.void,
712 ),
713 ),
714 ).pipe(Effect.forkChild);
715
716 yield* adapter.startSession({
717 threadId,
718 provider: ProviderDriverKind.make("grok"),
719 cwd: process.cwd(),
720 runtimeMode: "full-access",
721 });
722
723 const sendTurnFiber = yield* adapter
724 .sendTurn({ threadId, input: "cancel before the late update", attachments: [] })
725 .pipe(Effect.forkChild);
726 const turnId = yield* Deferred.await(turnStarted).pipe(Effect.timeout("2 seconds"));
727 yield* adapter.interruptTurn(threadId, turnId).pipe(Effect.timeout("2 seconds"));
728 yield* Fiber.join(sendTurnFiber).pipe(Effect.timeout("2 seconds"));
729 yield* Deferred.await(lateNativeUpdate).pipe(Effect.timeout("2 seconds"));
730 for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) {
731 yield* Effect.yieldNow;
732 }
733
734 const cancelledIndex = runtimeEvents.findIndex(
735 (event) =>
736 event.type === "turn.completed" &&
737 String(event.threadId) === String(threadId) &&
738 String(event.turnId) === String(turnId) &&
739 event.payload.state === "cancelled",
740 );
741 const turnOutputTypes = new Set([
742 "content.delta",
743 "item.started",
744 "item.updated",
745 "item.completed",
746 "turn.plan.updated",
747 ]);
748 const outputAfterCancellation = runtimeEvents
749 .slice(cancelledIndex + 1)
750 .filter(
751 (event) => String(event.threadId) === String(threadId) && turnOutputTypes.has(event.type),
752 );
753
754 assert.isAtLeast(cancelledIndex, 0);
755 assert.deepEqual(outputAfterCancellation, []);
756
757 yield* Fiber.interrupt(runtimeEventsFiber);
758 yield* adapter.stopSession(threadId);
759 }).pipe(TestClock.withLive),
760 );
761
762 it.effect("lets Stop cancel during the xAI completion drain window", () =>
763 Effect.gen(function* () {
764 const threadId = ThreadId.make("grok-stop-during-completion-drain");
765 const wrapperPath = yield* Effect.promise(() =>
766 makeMockGrokWrapper({
767 T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG: "1",
768 }),
769 );
770 const adapter = yield* makeTestAdapter(wrapperPath);
771
772 const runtimeEvents: ProviderRuntimeEvent[] = [];
773 const activeTurnIdRef = yield* Ref.make<TurnId | undefined>(undefined);
774 const trailingChunkTurnId = yield* Deferred.make<TurnId>();
775 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
776 Effect.gen(function* () {
777 runtimeEvents.push(event);
778 if (String(event.threadId) !== String(threadId)) {
779 return;
780 }
781 if (event.type === "turn.started") {
782 yield* Ref.set(activeTurnIdRef, event.turnId);
783 }
784 if (event.type !== "content.delta" || event.payload.delta !== "mock") {
785 return;
786 }
787 const turnId = event.turnId ?? (yield* Ref.get(activeTurnIdRef));
788 if (turnId === undefined) {
789 return;
790 }
791 yield* Deferred.succeed(trailingChunkTurnId, turnId).pipe(Effect.ignore);
792 }),
793 ).pipe(Effect.forkChild);
794
795 yield* adapter.startSession({
796 threadId,
797 provider: ProviderDriverKind.make("grok"),
798 cwd: process.cwd(),
799 runtimeMode: "full-access",
800 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
801 });
802
803 const sendTurnFiber = yield* adapter
804 .sendTurn({
805 threadId,
806 input: "cancel during completion drain",
807 attachments: [],
808 })
809 .pipe(Effect.forkChild);
810
811 const turnId = yield* Deferred.await(trailingChunkTurnId).pipe(Effect.timeout("2 seconds"));
812 yield* adapter.interruptTurn(threadId, turnId).pipe(Effect.timeout("2 seconds"));
813 yield* Fiber.join(sendTurnFiber).pipe(Effect.timeout("2 seconds"));
814
815 const turnCompletedEvents = runtimeEvents.filter(
816 (event): event is Extract<ProviderRuntimeEvent, { type: "turn.completed" }> =>
817 event.type === "turn.completed" && String(event.threadId) === String(threadId),
818 );
819 const readySessions = yield* adapter.listSessions();
820 const readySession = readySessions.find((session) => session.threadId === threadId);
821
822 assert.lengthOf(turnCompletedEvents, 1);
823 assert.equal(turnCompletedEvents[0]?.payload.state, "cancelled");
824 assert.equal(readySession?.status, "ready");
825 assert.isUndefined(readySession?.activeTurnId);
826
827 yield* Fiber.interrupt(runtimeEventsFiber);
828 yield* adapter.stopSession(threadId);
829 }),
830 );
831
832 it.effect("settles the in-flight prompt before emitting completion", () =>
833 Effect.gen(function* () {
834 const threadId = ThreadId.make("grok-completion-before-next-turn");
835 const wrapperPath = yield* Effect.promise(() => makeMockGrokWrapper());
836 const adapter = yield* makeTestAdapter(wrapperPath);
837 const completedCountRef = yield* Ref.make(0);
838 const secondTurnCompleted = yield* Deferred.make<void>();
839
840 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => {
841 if (event.type !== "turn.completed" || String(event.threadId) !== String(threadId)) {
842 return Effect.void;
843 }
844
845 return Ref.modify(completedCountRef, (count) => {
846 const nextCount = count + 1;
847 return [nextCount, nextCount] as const;
848 }).pipe(
849 Effect.flatMap((count) => {
850 if (count === 1) {
851 return adapter
852 .sendTurn({
853 threadId,
854 input: "second turn after completion",
855 attachments: [],
856 })
857 .pipe(Effect.forkChild, Effect.asVoid);
858 }
859 if (count === 2) {
860 return Deferred.succeed(secondTurnCompleted, undefined).pipe(Effect.asVoid);
861 }
862 return Effect.void;
863 }),
864 );
865 }).pipe(Effect.forkChild);
866
867 yield* adapter.startSession({
868 threadId,
869 provider: ProviderDriverKind.make("grok"),
870 cwd: process.cwd(),
871 runtimeMode: "full-access",
872 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
873 });
874
875 yield* adapter.sendTurn({
876 threadId,
877 input: "first turn",
878 attachments: [],
879 });
880 yield* Deferred.await(secondTurnCompleted);
881
882 const completedCount = yield* Ref.get(completedCountRef);
883 const readySessions = yield* adapter.listSessions();
884 const readySession = readySessions.find((session) => session.threadId === threadId);
885
886 assert.equal(completedCount, 2);
887 assert.equal(readySession?.status, "ready");
888 assert.isUndefined(readySession?.activeTurnId);
889
890 yield* Fiber.interrupt(runtimeEventsFiber);
891 yield* adapter.stopSession(threadId);
892 }),
893 );
894
895 it.effect("restores a Grok session to ready when the prompt RPC fails", () =>
896 Effect.gen(function* () {
897 const threadId = ThreadId.make("grok-prompt-failure-ready");
898 const wrapperPath = yield* Effect.promise(() =>
899 makeMockGrokWrapper({
900 T3_ACP_FAIL_PROMPT: "1",
901 }),
902 );
903 const adapter = yield* makeTestAdapter(wrapperPath);
904 const runtimeEvents: ProviderRuntimeEvent[] = [];
905 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
906 Effect.sync(() => {
907 runtimeEvents.push(event);
908 }),
909 ).pipe(Effect.forkChild);
910
911 yield* adapter.startSession({
912 threadId,
913 provider: ProviderDriverKind.make("grok"),
914 cwd: process.cwd(),
915 runtimeMode: "full-access",
916 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
917 });
918
919 const error = yield* Effect.flip(
920 adapter.sendTurn({
921 threadId,
922 input: "fail prompt",
923 attachments: [],
924 }),
925 );
926 const readySessions = yield* adapter.listSessions();
927 const readySession = readySessions.find((session) => session.threadId === threadId);
928 const failedTurnCompleted = runtimeEvents.find(
929 (event) => event.type === "turn.completed" && event.threadId === threadId,
930 );
931
932 assert.equal(error._tag, "ProviderAdapterRequestError");
933 assert.equal(readySession?.status, "ready");
934 assert.isUndefined(readySession?.activeTurnId);
935 assert.equal(failedTurnCompleted?.type, "turn.completed");
936 if (failedTurnCompleted?.type === "turn.completed") {
937 assert.equal(failedTurnCompleted.payload.state, "failed");
938 assert.isString(failedTurnCompleted.payload.errorMessage);
939 }
940
941 yield* Fiber.interrupt(runtimeEventsFiber);
942 yield* adapter.stopSession(threadId);
943 }),
944 );
945
946 it.effect("ignores replayed session/load updates when resuming a Grok session", () =>
947 Effect.gen(function* () {
948 const threadId = ThreadId.make("grok-load-replay-filter");
949 const wrapperPath = yield* Effect.promise(() =>
950 makeMockGrokWrapper({
951 T3_ACP_EMIT_LOAD_REPLAY: "1",
952 }),
953 );
954 const adapter = yield* makeTestAdapter(wrapperPath);
955 const runtimeEvents: ProviderRuntimeEvent[] = [];
956 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
957 Effect.sync(() => {
958 runtimeEvents.push(event);
959 }),
960 ).pipe(Effect.forkChild);
961
962 const session = yield* adapter.startSession({
963 threadId,
964 provider: ProviderDriverKind.make("grok"),
965 cwd: process.cwd(),
966 runtimeMode: "full-access",
967 modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" },
968 resumeCursor: { schemaVersion: 1, sessionId: "mock-session-1" },
969 });
970
971 yield* adapter.sendTurn({
972 threadId,
973 input: "after resume",
974 attachments: [],
975 });
976
977 assert.deepStrictEqual(session.resumeCursor, {
978 schemaVersion: 1,
979 sessionId: "mock-session-1",
980 });
981 assert.isFalse(
982 runtimeEvents.some(
983 (event) => event.type === "item.completed" && event.payload.title === "Replay tool",
984 ),
985 );
986 assert.isFalse(
987 runtimeEvents.some(
988 (event) =>
989 event.type === "content.delta" && event.payload.delta === "replayed assistant text",
990 ),
991 );
992
993 yield* Fiber.interrupt(runtimeEventsFiber);
994 yield* adapter.stopSession(threadId);
995 }),
996 );
997
178998 it.effect("rejects startSession when provider mismatches", () =>
179999 Effect.gen(function* () {
1801000 const wrapperPath = yield* Effect.promise(() => makeMockGrokWrapper());
......@@ -331,6 +1151,7 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
3311151 assert.deepEqual(resolvedEvent.payload.answers, {
3321152 "Which scope should Grok use?": "Workspace",
3331153 });
1154 assert.equal(String(resolvedEvent.turnId), String(requestedEvent.turnId));
3341155 yield* Fiber.join(sendTurnFiber);
3351156
3361157 yield* Fiber.interrupt(eventsFiber);
apps/server/src/provider/Layers/GrokAdapter.ts+512-56
......@@ -22,6 +22,7 @@ import * as FileSystem from "effect/FileSystem";
2222import * as Option from "effect/Option";
2323import * as Path from "effect/Path";
2424import * as PubSub from "effect/PubSub";
25import * as Ref from "effect/Ref";
2526import * as Schema from "effect/Schema";
2627import * as Scope from "effect/Scope";
2728import * as Semaphore from "effect/Semaphore";
......@@ -62,6 +63,7 @@ import {
6263 extractXAiAskUserQuestions,
6364 makeXAiAskUserQuestionCancelledResponse,
6465 makeXAiAskUserQuestionResponse,
66 promptResponseHasMissingXAiStopReason,
6567 XAiAskUserQuestionRequest,
6668} from "../acp/XAiAcpExtension.ts";
6769import { type GrokAdapterShape } from "../Services/GrokAdapter.ts";
......@@ -108,6 +110,8 @@ interface GrokSessionContext {
108110 turns: Array<{ id: TurnId; items: Array<unknown> }>;
109111 lastPlanFingerprint: string | undefined;
110112 activeTurnId: TurnId | undefined;
113 /** Turns already interrupted; late prompt RPCs must not resurrect them. */
114 interruptedTurnIds: Set<TurnId>;
111115 /** Number of sendTurn prompts currently in flight or being prepared.
112116 * >0 means a turn is actively running, so a new sendTurn is a steer that
113117 * continues it, and only the last remaining prompt settles the turn. */
......@@ -136,10 +140,38 @@ function settlePendingUserInputsAsCancelled(
136140 );
137141}
138142
143function appendPromptResultToTurn(
144 ctx: GrokSessionContext,
145 turnId: TurnId,
146 promptParts: ReadonlyArray<EffectAcpSchema.ContentBlock>,
147 result: EffectAcpSchema.PromptResponse,
148): void {
149 const existingTurnRecord = ctx.turns.find((turn) => turn.id === turnId);
150 ctx.turns = existingTurnRecord
151 ? ctx.turns.map((turn) =>
152 turn.id === turnId
153 ? { ...turn, items: [...turn.items, { prompt: promptParts, result }] }
154 : turn,
155 )
156 : [...ctx.turns, { id: turnId, items: [{ prompt: promptParts, result }] }];
157}
158
139159function isRecord(value: unknown): value is Record<string, unknown> {
140160 return typeof value === "object" && value !== null && !Array.isArray(value);
141161}
142162
163const resolveNotificationTurnId = (ctx: GrokSessionContext): TurnId | undefined => ctx.activeTurnId;
164
165const resolveCallbackTurnId = (ctx: GrokSessionContext): TurnId | undefined => ctx.activeTurnId;
166
167const resolveSessionCallbackTurnId = (
168 sessions: ReadonlyMap<ThreadId, GrokSessionContext>,
169 threadId: ThreadId,
170): TurnId | undefined => {
171 const ctx = sessions.get(threadId);
172 return ctx ? resolveCallbackTurnId(ctx) : undefined;
173};
174
143175function parseGrokResume(raw: unknown): { sessionId: string } | undefined {
144176 if (!isRecord(raw)) return undefined;
145177 if (raw.schemaVersion !== GROK_RESUME_VERSION) return undefined;
......@@ -170,6 +202,28 @@ function selectAutoApprovedPermissionOption(
170202 );
171203}
172204
205function completedStopReasonFromPromptResponse(
206 response: EffectAcpSchema.PromptResponse | undefined,
207): EffectAcpSchema.StopReason | null {
208 if (response === undefined || promptResponseHasMissingXAiStopReason(response)) {
209 return null;
210 }
211 return response.stopReason;
212}
213
214export function grokPromptSettlementBelongsToContext(input: {
215 readonly liveAcpSessionId: string;
216 readonly expectedAcpSessionId: string;
217 readonly liveActiveTurnId: TurnId | undefined;
218 readonly liveSessionActiveTurnId: TurnId | undefined;
219 readonly turnId: TurnId;
220}): boolean {
221 return (
222 input.liveAcpSessionId === input.expectedAcpSessionId &&
223 (input.liveActiveTurnId === input.turnId || input.liveSessionActiveTurnId === input.turnId)
224 );
225}
226
173227export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapterLiveOptions) {
174228 return Effect.gen(function* () {
175229 const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("grok");
......@@ -240,6 +294,144 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
240294 const withThreadLock = <A, E, R>(threadId: string, effect: Effect.Effect<A, E, R>) =>
241295 Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect));
242296
297 const settlePromptInFlight = (
298 threadId: ThreadId,
299 turnId: TurnId,
300 expectedAcpSessionId: string,
301 options?: {
302 readonly errorMessage?: string;
303 readonly completedStopReason?: EffectAcpSchema.StopReason | null;
304 readonly emitTurnCompletion?: boolean;
305 /** Interrupt/cancel: drop every outstanding prompt slot and settle once. */
306 readonly settleAllPrompts?: boolean;
307 },
308 ) =>
309 Effect.gen(function* () {
310 const liveCtx = sessions.get(threadId);
311 if (!liveCtx) {
312 return;
313 }
314 const settlementBelongsToLiveContext = grokPromptSettlementBelongsToContext({
315 liveAcpSessionId: liveCtx.acpSessionId,
316 expectedAcpSessionId,
317 liveActiveTurnId: liveCtx.activeTurnId,
318 liveSessionActiveTurnId: liveCtx.session.activeTurnId,
319 turnId,
320 });
321 if (!settlementBelongsToLiveContext) {
322 // interruptTurn already consumed every prompt slot for this turn. A
323 // late prompt result must neither emit a second terminal event nor
324 // consume a slot belonging to a newer turn on the same ACP session.
325 if (
326 liveCtx.acpSessionId !== expectedAcpSessionId ||
327 liveCtx.interruptedTurnIds.has(turnId)
328 ) {
329 return;
330 }
331 if (options?.emitTurnCompletion !== false) {
332 if (options?.errorMessage !== undefined) {
333 yield* offerRuntimeEvent({
334 type: "turn.completed",
335 ...(yield* makeEventStamp()),
336 provider: PROVIDER,
337 threadId,
338 turnId,
339 payload: {
340 state: "failed",
341 errorMessage: options.errorMessage,
342 },
343 });
344 } else if (options?.completedStopReason !== undefined) {
345 yield* offerRuntimeEvent({
346 type: "turn.completed",
347 ...(yield* makeEventStamp()),
348 provider: PROVIDER,
349 threadId,
350 turnId,
351 payload: {
352 state: options.completedStopReason === "cancelled" ? "cancelled" : "completed",
353 stopReason: options.completedStopReason ?? null,
354 },
355 });
356 }
357 }
358 return;
359 }
360 let settleTurnId = turnId;
361 if (options?.settleAllPrompts) {
362 liveCtx.promptsInFlight = 0;
363 if (liveCtx.activeTurnId !== turnId && liveCtx.session.activeTurnId !== turnId) {
364 const fallbackTurnId = liveCtx.activeTurnId ?? liveCtx.session.activeTurnId;
365 if (!fallbackTurnId) {
366 if (liveCtx.session.status === "running" || liveCtx.session.status === "connecting") {
367 const updatedAt = yield* nowIso;
368 const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session;
369 liveCtx.activeTurnId = undefined;
370 liveCtx.session = {
371 ...readySession,
372 status: "ready",
373 updatedAt,
374 };
375 }
376 return;
377 }
378 settleTurnId = fallbackTurnId;
379 }
380 } else {
381 const remainingPrompts = Math.max(0, liveCtx.promptsInFlight - 1);
382 if (
383 remainingPrompts > 0 ||
384 liveCtx.activeTurnId !== settleTurnId ||
385 liveCtx.session.activeTurnId !== settleTurnId
386 ) {
387 liveCtx.promptsInFlight = remainingPrompts;
388 return;
389 }
390 liveCtx.promptsInFlight = remainingPrompts;
391 }
392 const updatedAt = yield* nowIso;
393 const canEmitTurnCompletion =
394 liveCtx.session.status === "running" || liveCtx.session.status === "connecting";
395 const shouldEmitFailedTurn = options?.errorMessage !== undefined && canEmitTurnCompletion;
396 const shouldEmitCompletedTurn =
397 options?.completedStopReason !== undefined && canEmitTurnCompletion;
398 const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session;
399 liveCtx.activeTurnId = undefined;
400 liveCtx.session = {
401 ...readySession,
402 status: "ready",
403 updatedAt,
404 };
405 if (options?.emitTurnCompletion === false) {
406 return;
407 }
408 if (shouldEmitFailedTurn) {
409 yield* offerRuntimeEvent({
410 type: "turn.completed",
411 ...(yield* makeEventStamp()),
412 provider: PROVIDER,
413 threadId,
414 turnId: settleTurnId,
415 payload: {
416 state: "failed",
417 errorMessage: options.errorMessage,
418 },
419 });
420 } else if (shouldEmitCompletedTurn) {
421 yield* offerRuntimeEvent({
422 type: "turn.completed",
423 ...(yield* makeEventStamp()),
424 provider: PROVIDER,
425 threadId,
426 turnId: settleTurnId,
427 payload: {
428 state: options.completedStopReason === "cancelled" ? "cancelled" : "completed",
429 stopReason: options.completedStopReason ?? null,
430 },
431 });
432 }
433 });
434
243435 const logNative = (threadId: ThreadId, method: string, payload: unknown) =>
244436 Effect.gen(function* () {
245437 if (!nativeEventLogger) return;
......@@ -271,6 +463,8 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
271463
272464 const emitPlanUpdate = (
273465 ctx: GrokSessionContext,
466 turnId: TurnId | undefined,
467 stamp: { readonly eventId: EventId; readonly createdAt: string },
274468 payload: {
275469 readonly explanation?: string | null;
276470 readonly plan: ReadonlyArray<{
......@@ -282,17 +476,17 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
282476 method: string,
283477 ) =>
284478 Effect.gen(function* () {
285 const fingerprint = `${ctx.activeTurnId ?? "no-turn"}:${encodeJsonStringForDiagnostics(payload) ?? "[unserializable payload]"}`;
479 const fingerprint = `${turnId ?? "no-turn"}:${encodeJsonStringForDiagnostics(payload) ?? "[unserializable payload]"}`;
286480 if (ctx.lastPlanFingerprint === fingerprint) {
287481 return;
288482 }
289483 ctx.lastPlanFingerprint = fingerprint;
290484 yield* offerRuntimeEvent(
291485 makeAcpPlanUpdatedEvent({
292 stamp: yield* makeEventStamp(),
486 stamp,
293487 provider: PROVIDER,
294488 threadId: ctx.threadId,
295 turnId: ctx.activeTurnId,
489 turnId,
296490 payload,
297491 source: "acp.jsonrpc",
298492 method,
......@@ -424,13 +618,14 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
424618 const requestId = ApprovalRequestId.make(yield* randomUUIDv4);
425619 const runtimeRequestId = RuntimeRequestId.make(requestId);
426620 const resolution = yield* Deferred.make<PendingUserInputResolution>();
621 const turnId = resolveSessionCallbackTurnId(sessions, input.threadId);
427622 pendingUserInputs.set(requestId, { resolution });
428623 yield* offerRuntimeEvent({
429624 type: "user-input.requested",
430625 ...(yield* makeEventStamp()),
431626 provider: PROVIDER,
432627 threadId: input.threadId,
433 turnId: sessions.get(input.threadId)?.activeTurnId,
628 turnId,
434629 requestId: runtimeRequestId,
435630 payload: { questions: extractXAiAskUserQuestions(params) },
436631 raw: {
......@@ -447,7 +642,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
447642 ...(yield* makeEventStamp()),
448643 provider: PROVIDER,
449644 threadId: input.threadId,
450 turnId: sessions.get(input.threadId)?.activeTurnId,
645 turnId,
451646 requestId: runtimeRequestId,
452647 payload: { answers: resolvedAnswers },
453648 raw: {
......@@ -486,13 +681,14 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
486681 const requestId = ApprovalRequestId.make(yield* randomUUIDv4);
487682 const runtimeRequestId = RuntimeRequestId.make(requestId);
488683 const decision = yield* Deferred.make<ProviderApprovalDecision>();
684 const turnId = resolveSessionCallbackTurnId(sessions, input.threadId);
489685 pendingApprovals.set(requestId, { decision });
490686 yield* offerRuntimeEvent(
491687 makeAcpRequestOpenedEvent({
492688 stamp: yield* makeEventStamp(),
493689 provider: PROVIDER,
494690 threadId: input.threadId,
495 turnId: sessions.get(input.threadId)?.activeTurnId,
691 turnId,
496692 requestId: runtimeRequestId,
497693 permissionRequest,
498694 detail:
......@@ -512,7 +708,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
512708 stamp: yield* makeEventStamp(),
513709 provider: PROVIDER,
514710 threadId: input.threadId,
515 turnId: sessions.get(input.threadId)?.activeTurnId,
711 turnId,
516712 requestId: runtimeRequestId,
517713 permissionRequest,
518714 decision: resolved,
......@@ -578,6 +774,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
578774 turns: [],
579775 lastPlanFingerprint: undefined,
580776 activeTurnId: undefined,
777 interruptedTurnIds: new Set(),
581778 promptsInFlight: 0,
582779 currentModelId: boundModelId,
583780 stopped: false,
......@@ -586,14 +783,39 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
586783 const nf = yield* Stream.runDrain(
587784 Stream.mapEffect(acp.getEvents(), (event) =>
588785 Effect.gen(function* () {
786 if (event._tag === "EventStreamBarrier") {
787 yield* Deferred.succeed(event.acknowledge, undefined);
788 return;
789 }
790 if (
791 event._tag === "PlanUpdated" ||
792 event._tag === "ToolCallUpdated" ||
793 event._tag === "ContentDelta"
794 ) {
795 yield* logNative(ctx.threadId, "session/update", event.rawPayload);
796 }
797
798 if (event._tag === "ModeChanged") {
799 return;
800 }
801
802 const notificationTurnId = resolveNotificationTurnId(ctx);
803 if (
804 notificationTurnId === undefined ||
805 ctx.interruptedTurnIds.has(notificationTurnId)
806 ) {
807 return;
808 }
809 const stamp = yield* makeEventStamp();
810
589811 switch (event._tag) {
590812 case "AssistantItemStarted":
591813 yield* offerRuntimeEvent(
592814 makeAcpAssistantItemEvent({
593 stamp: yield* makeEventStamp(),
815 stamp,
594816 provider: PROVIDER,
595817 threadId: ctx.threadId,
596 turnId: ctx.activeTurnId,
818 turnId: notificationTurnId,
597819 itemId: event.itemId,
598820 lifecycle: "item.started",
599821 }),
......@@ -602,40 +824,44 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
602824 case "AssistantItemCompleted":
603825 yield* offerRuntimeEvent(
604826 makeAcpAssistantItemEvent({
605 stamp: yield* makeEventStamp(),
827 stamp,
606828 provider: PROVIDER,
607829 threadId: ctx.threadId,
608 turnId: ctx.activeTurnId,
830 turnId: notificationTurnId,
609831 itemId: event.itemId,
610832 lifecycle: "item.completed",
611833 }),
612834 );
613835 return;
614836 case "PlanUpdated":
615 yield* logNative(ctx.threadId, "session/update", event.rawPayload);
616 yield* emitPlanUpdate(ctx, event.payload, event.rawPayload, "session/update");
837 yield* emitPlanUpdate(
838 ctx,
839 notificationTurnId,
840 stamp,
841 event.payload,
842 event.rawPayload,
843 "session/update",
844 );
617845 return;
618846 case "ToolCallUpdated":
619 yield* logNative(ctx.threadId, "session/update", event.rawPayload);
620847 yield* offerRuntimeEvent(
621848 makeAcpToolCallEvent({
622 stamp: yield* makeEventStamp(),
849 stamp,
623850 provider: PROVIDER,
624851 threadId: ctx.threadId,
625 turnId: ctx.activeTurnId,
852 turnId: notificationTurnId,
626853 toolCall: event.toolCall,
627854 rawPayload: event.rawPayload,
628855 }),
629856 );
630857 return;
631858 case "ContentDelta":
632 yield* logNative(ctx.threadId, "session/update", event.rawPayload);
633859 yield* offerRuntimeEvent(
634860 makeAcpContentDeltaEvent({
635 stamp: yield* makeEventStamp(),
861 stamp,
636862 provider: PROVIDER,
637863 threadId: ctx.threadId,
638 turnId: ctx.activeTurnId,
864 turnId: notificationTurnId,
639865 ...(event.itemId ? { itemId: event.itemId } : {}),
640866 text: event.text,
641867 rawPayload: event.rawPayload,
......@@ -697,6 +923,15 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
697923 // resolving from here on does not settle the turn; decremented on
698924 // preparation failure here, and after the prompt below otherwise.
699925 ctx.promptsInFlight += 1;
926 // Bind the turn id before cooperative yields so interruptTurn can
927 // settle this prompt even if stop arrives during preparation.
928 ctx.activeTurnId = turnId;
929 ctx.session = {
930 ...ctx.session,
931 status: steeringTurnId === undefined ? "connecting" : "running",
932 activeTurnId: turnId,
933 updatedAt: yield* nowIso,
934 };
700935
701936 return yield* Effect.gen(function* () {
702937 const turnModelSelection =
......@@ -765,12 +1000,27 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
7651000 const displayModel = currentModelId
7661001 ? resolveGrokAcpBaseModelId(currentModelId)
7671002 : undefined;
768 ctx.activeTurnId = turnId;
1003 for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) {
1004 yield* Effect.yieldNow;
1005 }
1006 if (ctx.interruptedTurnIds.has(turnId)) {
1007 yield* settlePromptInFlight(input.threadId, turnId, ctx.acpSessionId, {
1008 completedStopReason: "cancelled",
1009 emitTurnCompletion: false,
1010 settleAllPrompts: true,
1011 });
1012 return yield* new ProviderAdapterRequestError({
1013 provider: PROVIDER,
1014 method: "session/prompt",
1015 detail: "Grok prompt was interrupted during preparation.",
1016 });
1017 }
7691018 if (steeringTurnId === undefined) {
7701019 ctx.lastPlanFingerprint = undefined;
7711020 }
7721021 ctx.session = {
7731022 ...ctx.session,
1023 status: "running",
7741024 activeTurnId: turnId,
7751025 updatedAt: yield* nowIso,
7761026 ...(displayModel ? { model: displayModel } : {}),
......@@ -796,13 +1046,27 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
7961046 };
7971047 }).pipe(
7981048 Effect.tapCause(() =>
799 Effect.sync(() => {
800 ctx.promptsInFlight = Math.max(0, ctx.promptsInFlight - 1);
1049 Effect.gen(function* () {
1050 const liveCtx = sessions.get(input.threadId);
1051 if (!liveCtx) {
1052 return;
1053 }
1054 yield* settlePromptInFlight(input.threadId, turnId, liveCtx.acpSessionId, {
1055 errorMessage: "Grok prompt preparation failed.",
1056 emitTurnCompletion: false,
1057 });
8011058 }),
8021059 ),
8031060 );
8041061 }),
8051062 );
1063 const promptSettled = yield* Ref.make(false);
1064 const promptRpcSucceeded = yield* Ref.make(false);
1065 const promptResultRef = yield* Ref.make<EffectAcpSchema.PromptResponse | undefined>(
1066 undefined,
1067 );
1068
1069 const promptFailureMessageRef = yield* Ref.make<string | undefined>(undefined);
8061070
8071071 return yield* Effect.gen(function* () {
8081072 const result = yield* prepared.acp
......@@ -810,6 +1074,18 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
8101074 prompt: prepared.promptParts,
8111075 })
8121076 .pipe(
1077 Effect.tap((promptResult) =>
1078 Effect.all([
1079 Ref.set(promptRpcSucceeded, true),
1080 Ref.set(promptResultRef, promptResult),
1081 ]),
1082 ),
1083 Effect.tapError((error) =>
1084 Ref.set(
1085 promptFailureMessageRef,
1086 mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error).message,
1087 ).pipe(Effect.andThen(prepared.acp.drainEvents)),
1088 ),
8131089 Effect.mapError((error) =>
8141090 mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error),
8151091 ),
......@@ -820,38 +1096,88 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
8201096 Effect.gen(function* () {
8211097 const ctx = yield* requireSession(input.threadId);
8221098 if (ctx.acpSessionId !== prepared.acpSessionId) {
1099 yield* settlePromptInFlight(
1100 input.threadId,
1101 prepared.turnId,
1102 prepared.acpSessionId,
1103 {
1104 errorMessage: "Grok session changed before the turn completed.",
1105 settleAllPrompts: true,
1106 },
1107 );
1108 yield* Ref.set(promptSettled, true);
8231109 return yield* new ProviderAdapterRequestError({
8241110 provider: PROVIDER,
8251111 method: "session/prompt",
8261112 detail: "Grok session changed before the turn completed.",
8271113 });
8281114 }
1115 // Keep prompt settlement atomic with respect to Stop and steering.
1116 // interruptTurn marks its target before waiting for this lock, so
1117 // cancellation can still win while queued ACP events are drained.
1118 for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) {
1119 yield* Effect.yieldNow;
1120 }
1121 yield* prepared.acp.drainEvents;
1122 if (ctx.interruptedTurnIds.has(prepared.turnId)) {
1123 yield* Ref.set(promptSettled, true);
1124 return {
1125 threadId: input.threadId,
1126 turnId: prepared.turnId,
1127 resumeCursor: ctx.session.resumeCursor,
1128 };
1129 }
8291130
830 const existingTurnRecord = ctx.turns.find((turn) => turn.id === prepared.turnId);
831 ctx.turns = existingTurnRecord
832 ? ctx.turns.map((turn) =>
833 turn.id === prepared.turnId
834 ? {
835 ...turn,
836 items: [...turn.items, { prompt: prepared.promptParts, result }],
837 }
838 : turn,
839 )
840 : [
841 ...ctx.turns,
842 { id: prepared.turnId, items: [{ prompt: prepared.promptParts, result }] },
843 ];
1131 if (
1132 ctx.promptsInFlight <= 0 ||
1133 ctx.activeTurnId !== prepared.turnId ||
1134 ctx.session.activeTurnId !== prepared.turnId
1135 ) {
1136 yield* Ref.set(promptSettled, true);
1137 return {
1138 threadId: input.threadId,
1139 turnId: prepared.turnId,
1140 resumeCursor: ctx.session.resumeCursor,
1141 };
1142 }
1143
1144 appendPromptResultToTurn(ctx, prepared.turnId, prepared.promptParts, result);
8441145 ctx.session = {
8451146 ...ctx.session,
1147 status: "running",
8461148 activeTurnId: prepared.turnId,
8471149 updatedAt: yield* nowIso,
8481150 ...(prepared.displayModel ? { model: prepared.displayModel } : {}),
8491151 };
1152 const remainingPrompts = Math.max(0, ctx.promptsInFlight - 1);
1153 ctx.promptsInFlight = remainingPrompts;
8501154
851 // Only the last remaining prompt settles the turn — a steer-
852 // superseded prompt resolving (usually cancelled) while another
853 // is in flight or pending must leave the merged turn running.
854 if (ctx.promptsInFlight === 1) {
1155 // Only the last remaining prompt settles the turn. A steer-
1156 // superseded prompt resolving while another is in flight or
1157 // pending must leave the merged turn running.
1158 if (
1159 remainingPrompts === 0 &&
1160 ctx.activeTurnId === prepared.turnId &&
1161 ctx.session.activeTurnId === prepared.turnId
1162 ) {
1163 if (ctx.interruptedTurnIds.has(prepared.turnId)) {
1164 yield* Ref.set(promptSettled, true);
1165 return {
1166 threadId: input.threadId,
1167 turnId: prepared.turnId,
1168 resumeCursor: ctx.session.resumeCursor,
1169 };
1170 }
1171 const completedAt = yield* nowIso;
1172 const { activeTurnId: _completedTurnId, ...readySession } = ctx.session;
1173 ctx.activeTurnId = undefined;
1174 ctx.session = {
1175 ...readySession,
1176 status: "ready",
1177 updatedAt: completedAt,
1178 ...(prepared.displayModel ? { model: prepared.displayModel } : {}),
1179 };
1180 const completedStopReason = completedStopReasonFromPromptResponse(result);
8551181 yield* offerRuntimeEvent({
8561182 type: "turn.completed",
8571183 ...(yield* makeEventStamp()),
......@@ -860,9 +1186,13 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
8601186 turnId: prepared.turnId,
8611187 payload: {
8621188 state: result.stopReason === "cancelled" ? "cancelled" : "completed",
863 stopReason: result.stopReason ?? null,
1189 stopReason: completedStopReason,
8641190 },
8651191 });
1192 ctx.interruptedTurnIds.delete(prepared.turnId);
1193 yield* Ref.set(promptSettled, true);
1194 } else if (remainingPrompts > 0) {
1195 yield* Ref.set(promptSettled, true);
8661196 }
8671197
8681198 return {
......@@ -874,27 +1204,153 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
8741204 );
8751205 }).pipe(
8761206 Effect.ensuring(
877 Effect.sync(() => {
878 const liveCtx = sessions.get(input.threadId);
879 if (liveCtx) {
880 liveCtx.promptsInFlight = Math.max(0, liveCtx.promptsInFlight - 1);
1207 Effect.gen(function* () {
1208 if (yield* Ref.get(promptSettled)) {
1209 return;
8811210 }
882 }),
1211
1212 if (yield* Ref.get(promptRpcSucceeded)) {
1213 const promptResult = yield* Ref.get(promptResultRef);
1214 if (promptResult === undefined) {
1215 return;
1216 }
1217 yield* withThreadLock(
1218 input.threadId,
1219 Effect.gen(function* () {
1220 const ctx = yield* requireSession(input.threadId);
1221 if (ctx.acpSessionId !== prepared.acpSessionId) {
1222 yield* settlePromptInFlight(
1223 input.threadId,
1224 prepared.turnId,
1225 prepared.acpSessionId,
1226 {
1227 errorMessage: "Grok session changed before the turn completed.",
1228 settleAllPrompts: true,
1229 },
1230 );
1231 return;
1232 }
1233 if (ctx.interruptedTurnIds.has(prepared.turnId)) {
1234 return;
1235 }
1236 if (
1237 ctx.promptsInFlight <= 0 ||
1238 ctx.activeTurnId !== prepared.turnId ||
1239 ctx.session.activeTurnId !== prepared.turnId
1240 ) {
1241 return;
1242 }
1243 appendPromptResultToTurn(
1244 ctx,
1245 prepared.turnId,
1246 prepared.promptParts,
1247 promptResult,
1248 );
1249 yield* settlePromptInFlight(
1250 input.threadId,
1251 prepared.turnId,
1252 prepared.acpSessionId,
1253 {
1254 completedStopReason: completedStopReasonFromPromptResponse(promptResult),
1255 },
1256 );
1257 }),
1258 );
1259 return;
1260 }
1261
1262 const errorMessage = yield* Ref.get(promptFailureMessageRef);
1263 yield* withThreadLock(
1264 input.threadId,
1265 settlePromptInFlight(input.threadId, prepared.turnId, prepared.acpSessionId, {
1266 errorMessage: errorMessage ?? "Grok prompt request failed.",
1267 }),
1268 );
1269 }).pipe(Effect.catch(() => Effect.void)),
8831270 ),
8841271 );
8851272 });
8861273
887 const interruptTurn: GrokAdapterShape["interruptTurn"] = (threadId) =>
1274 const interruptTurn: GrokAdapterShape["interruptTurn"] = (threadId, turnId) =>
8881275 Effect.gen(function* () {
889 const ctx = yield* requireSession(threadId);
890 yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals);
891 yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs);
892 yield* Effect.ignore(
893 ctx.acp.cancel.pipe(
894 Effect.mapError((error) =>
895 mapAcpToAdapterError(PROVIDER, threadId, "session/cancel", error),
896 ),
897 ),
1276 const observed = yield* Effect.sync(() => {
1277 const ctx = sessions.get(threadId);
1278 if (!ctx || ctx.stopped) {
1279 return {
1280 _tag: "Proceed" as const,
1281 acpSessionId: undefined,
1282 interruptedTurnId: turnId,
1283 };
1284 }
1285 const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId;
1286 if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) {
1287 return { _tag: "Ignore" as const };
1288 }
1289 const interruptedTurnId = turnId ?? activeTurnId;
1290 if (interruptedTurnId !== undefined) {
1291 ctx.interruptedTurnIds.add(interruptedTurnId);
1292 }
1293 return {
1294 _tag: "Proceed" as const,
1295 acpSessionId: ctx.acpSessionId,
1296 interruptedTurnId,
1297 };
1298 });
1299 if (observed._tag === "Ignore") {
1300 return;
1301 }
1302
1303 yield* withThreadLock(
1304 threadId,
1305 Effect.gen(function* () {
1306 const ctx = yield* requireSession(threadId);
1307 if (observed.acpSessionId !== undefined && ctx.acpSessionId !== observed.acpSessionId) {
1308 return;
1309 }
1310 const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId;
1311 if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) {
1312 return;
1313 }
1314 if (
1315 observed.interruptedTurnId !== undefined &&
1316 activeTurnId !== undefined &&
1317 activeTurnId !== observed.interruptedTurnId
1318 ) {
1319 return;
1320 }
1321 const interruptedTurnId =
1322 observed.interruptedTurnId ?? turnId ?? activeTurnId ?? ctx.session.activeTurnId;
1323 yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals);
1324 yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs);
1325 yield* Effect.ignore(
1326 ctx.acp.cancel.pipe(
1327 Effect.mapError((error) =>
1328 mapAcpToAdapterError(PROVIDER, threadId, "session/cancel", error),
1329 ),
1330 ),
1331 );
1332 if (interruptedTurnId) {
1333 ctx.interruptedTurnIds.add(interruptedTurnId);
1334 yield* settlePromptInFlight(threadId, interruptedTurnId, ctx.acpSessionId, {
1335 completedStopReason: "cancelled",
1336 settleAllPrompts: true,
1337 });
1338 } else if (
1339 ctx.promptsInFlight > 0 ||
1340 ctx.session.status === "running" ||
1341 ctx.session.status === "connecting"
1342 ) {
1343 const updatedAt = yield* nowIso;
1344 ctx.promptsInFlight = 0;
1345 ctx.activeTurnId = undefined;
1346 const { activeTurnId: _activeTurnId, ...readySession } = ctx.session;
1347 ctx.session = {
1348 ...readySession,
1349 status: "ready",
1350 updatedAt,
1351 };
1352 }
1353 }),
8981354 );
8991355 });
9001356
apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts+222
......@@ -7,6 +7,9 @@ import * as NodeFS from "node:fs";
77import * as NodeServices from "@effect/platform-node/NodeServices";
88import { it } from "@effect/vitest";
99import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Option from "effect/Option";
12import * as TestClock from "effect/testing/TestClock";
1013import * as Stream from "effect/Stream";
1114import { describe, expect } from "vite-plus/test";
1215
......@@ -113,6 +116,122 @@ describe("AcpSessionRuntime", () => {
113116 ),
114117 );
115118
119 it.effect("drops session updates emitted for a child ACP session", () =>
120 Effect.gen(function* () {
121 const runtime = yield* AcpSessionRuntime.AcpSessionRuntime;
122 yield* runtime.start();
123
124 const promptResult = yield* runtime.prompt({
125 prompt: [{ type: "text", text: "hi" }],
126 });
127 expect(promptResult).toMatchObject({ stopReason: "end_turn" });
128
129 const notes = Array.from(yield* Stream.runCollect(Stream.take(runtime.getEvents(), 4)));
130 expect(notes.map((note) => note._tag)).toEqual([
131 "AssistantItemStarted",
132 "ContentDelta",
133 "ContentDelta",
134 "AssistantItemCompleted",
135 ]);
136 expect(
137 notes
138 .filter((note) => note._tag === "ContentDelta")
139 .map((note) => note.text)
140 .join(""),
141 ).toBe("root before child root after child");
142 expect(notes.some((note) => note._tag === "ToolCallUpdated")).toBe(false);
143 }).pipe(
144 Effect.provide(
145 AcpSessionRuntime.layer({
146 spawn: {
147 command: mockAgentCommand,
148 args: mockAgentArgs,
149 env: {
150 T3_ACP_EMIT_FOREIGN_SESSION_UPDATES: "1",
151 },
152 },
153 cwd: process.cwd(),
154 clientInfo: { name: "t3-test", version: "0.0.0" },
155 authMethodId: "test",
156 }),
157 ),
158 Effect.scoped,
159 Effect.provide(NodeServices.layer),
160 ),
161 );
162
163 it.effect("supports successive standard ACP prompts", () =>
164 Effect.gen(function* () {
165 const runtime = yield* AcpSessionRuntime.AcpSessionRuntime;
166 yield* runtime.start();
167
168 const firstPromptResult = yield* runtime.prompt({
169 prompt: [{ type: "text", text: "first" }],
170 });
171 const secondPromptResult = yield* runtime.prompt({
172 prompt: [{ type: "text", text: "second" }],
173 });
174
175 expect(firstPromptResult).toMatchObject({ stopReason: "end_turn" });
176 expect(secondPromptResult).toMatchObject({ stopReason: "end_turn" });
177 }).pipe(
178 Effect.provide(
179 AcpSessionRuntime.layer({
180 spawn: {
181 command: mockAgentCommand,
182 args: mockAgentArgs,
183 },
184 cwd: process.cwd(),
185 clientInfo: { name: "t3-test", version: "0.0.0" },
186 authMethodId: "test",
187 }),
188 ),
189 Effect.scoped,
190 Effect.provide(NodeServices.layer),
191 ),
192 );
193
194 it.effect("releases a fully silent prompt when session/cancel is requested", () =>
195 Effect.gen(function* () {
196 const runtime = yield* AcpSessionRuntime.AcpSessionRuntime;
197 yield* runtime.start();
198
199 const promptFiber = yield* runtime
200 .prompt({
201 prompt: [{ type: "text", text: "hang forever" }],
202 })
203 .pipe(Effect.forkChild({ startImmediately: true }));
204
205 yield* TestClock.adjust("500 millis");
206 yield* runtime.cancel;
207
208 const firstPromptResult = yield* Fiber.join(promptFiber);
209 expect(firstPromptResult).toMatchObject({ stopReason: "cancelled" });
210
211 const secondPromptResult = yield* runtime.prompt({
212 prompt: [{ type: "text", text: "second" }],
213 });
214 expect(secondPromptResult).toMatchObject({ stopReason: "end_turn" });
215 }).pipe(
216 Effect.provide(
217 AcpSessionRuntime.layer({
218 spawn: {
219 command: mockAgentCommand,
220 args: mockAgentArgs,
221 env: {
222 T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1",
223 },
224 },
225 cwd: process.cwd(),
226 clientInfo: { name: "t3-test", version: "0.0.0" },
227 authMethodId: "test",
228 }),
229 ),
230 Effect.scoped,
231 Effect.provide(NodeServices.layer),
232 ),
233 );
234
116235 it.effect("segments assistant text around ACP tool calls", () =>
117236 Effect.gen(function* () {
118237 const runtime = yield* AcpSessionRuntime.AcpSessionRuntime;
......@@ -346,6 +465,109 @@ describe("AcpSessionRuntime", () => {
346465 );
347466 });
348467
468 it.effect("fails session startup when session/load returns an error", () =>
469 Effect.gen(function* () {
470 const runtime = yield* AcpSessionRuntime.AcpSessionRuntime;
471 const error = yield* runtime.start().pipe(Effect.flip);
472
473 expect(error._tag).toBe("AcpRequestError");
474 }).pipe(
475 Effect.provide(
476 AcpSessionRuntime.layer({
477 authMethodId: "test",
478 spawn: {
479 command: mockAgentCommand,
480 args: mockAgentArgs,
481 env: {
482 T3_ACP_FAIL_LOAD_SESSION: "1",
483 },
484 },
485 cwd: process.cwd(),
486 resumeSessionId: "stale-session-id",
487 clientInfo: { name: "t3-test", version: "0.0.0" },
488 }),
489 ),
490 Effect.scoped,
491 Effect.provide(NodeServices.layer),
492 ),
493 );
494
495 it.effect("ignores session/update replay notifications during session/load", () =>
496 Effect.gen(function* () {
497 const runtime = yield* AcpSessionRuntime.AcpSessionRuntime;
498 yield* runtime.start();
499
500 yield* runtime.prompt({
501 prompt: [{ type: "text", text: "hi" }],
502 });
503 const notes = Array.from(yield* Stream.runCollect(Stream.take(runtime.getEvents(), 4)));
504 expect(notes.map((note) => note._tag)).toEqual([
505 "PlanUpdated",
506 "AssistantItemStarted",
507 "ContentDelta",
508 "AssistantItemCompleted",
509 ]);
510 expect(notes.some((note) => note._tag === "ToolCallUpdated")).toBe(false);
511 }).pipe(
512 Effect.provide(
513 AcpSessionRuntime.layer({
514 authMethodId: "test",
515 spawn: {
516 command: mockAgentCommand,
517 args: mockAgentArgs,
518 env: {
519 T3_ACP_EMIT_LOAD_REPLAY: "1",
520 },
521 },
522 cwd: process.cwd(),
523 resumeSessionId: "mock-session-1",
524 clientInfo: { name: "t3-test", version: "0.0.0" },
525 }),
526 ),
527 Effect.scoped,
528 Effect.provide(NodeServices.layer),
529 ),
530 );
531
532 it.effect("completes session/load after replay becomes idle while its RPC stays pending", () =>
533 Effect.gen(function* () {
534 const runtime = yield* AcpSessionRuntime.AcpSessionRuntime;
535 const started = yield* runtime.start().pipe(Effect.timeout("2 seconds"));
536
537 expect(started.sessionId).toBe("mock-session-1");
538 expect(started.sessionSetupResult._meta).toMatchObject({
539 t3SessionLoadReady: "replay_idle",
540 });
541
542 const unexpectedReplayEvent = yield* Stream.runHead(runtime.getEvents()).pipe(
543 Effect.timeoutOption("100 millis"),
544 );
545 expect(Option.isNone(unexpectedReplayEvent)).toBe(true);
546 }).pipe(
547 Effect.provide(
548 AcpSessionRuntime.layer({
549 authMethodId: "test",
550 spawn: {
551 command: mockAgentCommand,
552 args: mockAgentArgs,
553 env: {
554 T3_ACP_HANG_LOAD_SESSION_AFTER_REPLAY: "1",
555 T3_ACP_LOAD_SESSION_DELAY_MS: "10000",
556 },
557 },
558 cwd: process.cwd(),
559 resumeSessionId: "mock-session-1",
560 sessionLoadReplayIdleGap: "50 millis",
561 sessionLoadTimeout: "1 second",
562 clientInfo: { name: "t3-test", version: "0.0.0" },
563 }),
564 ),
565 Effect.scoped,
566 Effect.provide(NodeServices.layer),
567 TestClock.withLive,
568 ),
569 );
570
349571 it.effect("rejects invalid config option values before sending session/set_config_option", () => {
350572 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "acp-runtime-"));
351573 const requestLogPath = NodePath.join(tempDir, "requests.ndjson");
apps/server/src/provider/acp/AcpRuntimeModel.test.ts+91
......@@ -8,6 +8,8 @@ import {
88 parsePermissionRequest,
99 parseSessionModeState,
1010 parseSessionUpdateEvent,
11 sessionUpdateIsReplay,
12 syntheticLoadSessionResponseFromInitialize,
1113} from "./AcpRuntimeModel.ts";
1214
1315describe("AcpRuntimeModel", () => {
......@@ -59,6 +61,95 @@ describe("AcpRuntimeModel", () => {
5961 expect(modelConfigId).toBe("model");
6062 });
6163
64 it("detects Grok session replay updates from _meta.isReplay", () => {
65 expect(
66 sessionUpdateIsReplay({
67 _meta: { isReplay: true },
68 sessionId: "session-1",
69 update: {
70 sessionUpdate: "agent_message_chunk",
71 content: { type: "text", text: "replayed" },
72 },
73 } satisfies EffectAcpSchema.SessionNotification),
74 ).toBe(true);
75 expect(
76 sessionUpdateIsReplay({
77 sessionId: "session-1",
78 update: {
79 sessionUpdate: "agent_message_chunk",
80 content: { type: "text", text: "live" },
81 },
82 } satisfies EffectAcpSchema.SessionNotification),
83 ).toBe(false);
84 });
85
86 it("builds a synthetic load response from initialize model state", () => {
87 const response = syntheticLoadSessionResponseFromInitialize({
88 protocolVersion: 1,
89 _meta: {
90 modelState: {
91 currentModelId: "grok-build",
92 availableModels: [{ modelId: "grok-build", name: "Grok Build" }],
93 },
94 },
95 } satisfies EffectAcpSchema.InitializeResponse);
96
97 expect(response.models?.currentModelId).toBe("grok-build");
98 expect(response._meta).toMatchObject({ t3SessionLoadReady: "replay_idle" });
99 });
100
101 it("accepts initialize model descriptions with null", () => {
102 const response = syntheticLoadSessionResponseFromInitialize({
103 protocolVersion: 1,
104 _meta: {
105 modelState: {
106 currentModelId: "grok-build",
107 availableModels: [{ modelId: "grok-build", name: "Grok Build", description: null }],
108 },
109 },
110 } satisfies EffectAcpSchema.InitializeResponse);
111
112 expect(response.models?.availableModels[0]?.description).toBeNull();
113 });
114
115 it("ignores malformed initialize model state in synthetic load responses", () => {
116 const response = syntheticLoadSessionResponseFromInitialize({
117 protocolVersion: 1,
118 _meta: {
119 modelState: {
120 currentModelId: "grok-build",
121 availableModels: [null],
122 },
123 modeState: {
124 currentModeId: "code",
125 availableModes: [{ id: "code", name: 12 }],
126 },
127 },
128 } as EffectAcpSchema.InitializeResponse);
129
130 expect(response.models).toBeUndefined();
131 expect(response.modes).toBeUndefined();
132 expect(response._meta).toMatchObject({ t3SessionLoadReady: "replay_idle" });
133 });
134
135 it("builds a synthetic load response with initialize mode state", () => {
136 const response = syntheticLoadSessionResponseFromInitialize({
137 protocolVersion: 1,
138 _meta: {
139 modeState: {
140 currentModeId: "code",
141 availableModes: [
142 { id: "ask", name: "Ask" },
143 { id: "code", name: "Code" },
144 ],
145 },
146 },
147 } satisfies EffectAcpSchema.InitializeResponse);
148
149 expect(response.modes?.currentModeId).toBe("code");
150 expect(response.modes?.availableModes).toHaveLength(2);
151 });
152
62153 it("projects typed ACP tool call updates into runtime events", () => {
63154 const created = parseSessionUpdateEvent({
64155 sessionId: "session-1",
apps/server/src/provider/acp/AcpRuntimeModel.ts+91
......@@ -1,3 +1,8 @@
1import * as Clock from "effect/Clock";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Ref from "effect/Ref";
16import type * as EffectAcpSchema from "effect-acp/schema";
27import { deriveToolActivityPresentation } from "@t3tools/shared/toolActivity";
38import type { ToolLifecycleItemType } from "@t3tools/contracts";
......@@ -6,6 +11,40 @@ function isRecord(value: unknown): value is Record<string, unknown> {
611 return typeof value === "object" && value !== null && !Array.isArray(value);
712}
813
14function isSessionModelState(value: unknown): value is EffectAcpSchema.SessionModelState {
15 if (!isRecord(value) || typeof value.currentModelId !== "string") {
16 return false;
17 }
18 if (!Array.isArray(value.availableModels)) {
19 return false;
20 }
21 return value.availableModels.every(
22 (model) =>
23 isRecord(model) &&
24 typeof model.modelId === "string" &&
25 typeof model.name === "string" &&
26 (model.description === undefined ||
27 model.description === null ||
28 typeof model.description === "string"),
29 );
30}
31
32function isSessionModeState(value: unknown): value is EffectAcpSchema.SessionModeState {
33 if (!isRecord(value) || typeof value.currentModeId !== "string") {
34 return false;
35 }
36 if (!Array.isArray(value.availableModes)) {
37 return false;
38 }
39 return value.availableModes.every(
40 (mode) =>
41 isRecord(mode) &&
42 typeof mode.id === "string" &&
43 typeof mode.name === "string" &&
44 (mode.description === undefined || typeof mode.description === "string"),
45 );
46}
47
948export interface AcpSessionMode {
1049 readonly id: string;
1150 readonly name: string;
......@@ -414,6 +453,58 @@ export function parsePermissionRequest(
414453 };
415454}
416455
456export function sessionUpdateIsReplay(params: EffectAcpSchema.SessionNotification): boolean {
457 const meta = params._meta;
458 return isRecord(meta) && meta.isReplay === true;
459}
460
461export interface SessionLoadGate {
462 readonly active: boolean;
463 readonly lastActivityAtMillis: number | undefined;
464 readonly idleGap: Duration.Duration;
465 readonly initializeResult: EffectAcpSchema.InitializeResponse;
466}
467
468export const waitForSessionLoadReplayIdle = (input: {
469 readonly gateRef: Ref.Ref<Option.Option<SessionLoadGate>>;
470}): Effect.Effect<EffectAcpSchema.LoadSessionResponse, never> =>
471 Effect.gen(function* () {
472 const pollInterval = Duration.millis(25);
473 while (true) {
474 const gate = yield* Ref.get(input.gateRef);
475 if (
476 Option.isSome(gate) &&
477 gate.value.active &&
478 gate.value.lastActivityAtMillis !== undefined
479 ) {
480 const idleGapMillis = Duration.toMillis(gate.value.idleGap);
481 const nowMillis = yield* Clock.currentTimeMillis;
482 if (nowMillis - gate.value.lastActivityAtMillis >= idleGapMillis) {
483 return syntheticLoadSessionResponseFromInitialize(gate.value.initializeResult);
484 }
485 }
486 yield* Effect.sleep(pollInterval);
487 }
488 });
489
490export function syntheticLoadSessionResponseFromInitialize(
491 initializeResult: EffectAcpSchema.InitializeResponse,
492): EffectAcpSchema.LoadSessionResponse {
493 const meta = initializeResult._meta;
494 const modelState = isRecord(meta) ? meta.modelState : undefined;
495 const modeState = isRecord(meta) ? meta.modeState : undefined;
496 const models = isSessionModelState(modelState) ? modelState : undefined;
497 const modes = isSessionModeState(modeState) ? modeState : undefined;
498
499 return {
500 ...(models ? { models } : {}),
501 ...(modes ? { modes } : {}),
502 _meta: {
503 t3SessionLoadReady: "replay_idle",
504 },
505 };
506}
507
417508export function parseSessionUpdateEvent(params: EffectAcpSchema.SessionNotification): {
418509 readonly modeId?: string;
419510 readonly events: ReadonlyArray<AcpParsedSessionEvent>;
apps/server/src/provider/acp/AcpSessionRuntime.ts+177-46
......@@ -1,12 +1,16 @@
11import * as Cause from "effect/Cause";
2import * as Clock from "effect/Clock";
23import * as Context from "effect/Context";
34import * as Deferred from "effect/Deferred";
5import * as Duration from "effect/Duration";
46import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
7import * as Fiber from "effect/Fiber";
68import * as Layer from "effect/Layer";
9import * as Option from "effect/Option";
710import * as Queue from "effect/Queue";
811import * as Ref from "effect/Ref";
912import * as Scope from "effect/Scope";
13import * as Semaphore from "effect/Semaphore";
1014import * as Stream from "effect/Stream";
1115import * as ChildProcess from "effect/unstable/process/ChildProcess";
1216import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
......@@ -23,6 +27,9 @@ import {
2327 mergeToolCallState,
2428 parseSessionModeState,
2529 parseSessionUpdateEvent,
30 sessionUpdateIsReplay,
31 waitForSessionLoadReplayIdle,
32 type SessionLoadGate,
2633 type AcpParsedSessionEvent,
2734 type AcpSessionModeState,
2835 type AcpToolCallState,
......@@ -32,6 +39,16 @@ function formatConfigOptionValue(value: string | boolean): string {
3239 return JSON.stringify(value);
3340}
3441
42export interface AcpSessionEventStreamBarrier {
43 readonly _tag: "EventStreamBarrier";
44 readonly acknowledge: Deferred.Deferred<void>;
45}
46
47export type AcpSessionRuntimeEvent = AcpParsedSessionEvent | AcpSessionEventStreamBarrier;
48
49const defaultSessionLoadTimeout = Duration.seconds(90);
50const defaultSessionLoadReplayIdleGap = Duration.seconds(2);
51
3552export interface AcpSpawnInput {
3653 readonly command: string;
3754 readonly args: ReadonlyArray<string>;
......@@ -43,6 +60,8 @@ export interface AcpSessionRuntimeOptions {
4360 readonly spawn: AcpSpawnInput;
4461 readonly cwd: string;
4562 readonly resumeSessionId?: string;
63 readonly sessionLoadTimeout?: Duration.Input;
64 readonly sessionLoadReplayIdleGap?: Duration.Input;
4665 readonly clientCapabilities?: EffectAcpSchema.InitializeRequest["clientCapabilities"];
4766 readonly clientInfo: {
4867 readonly name: string;
......@@ -160,7 +179,9 @@ export class AcpSessionRuntime extends Context.Service<
160179 */
161180 readonly start: () => Effect.Effect<AcpSessionRuntimeStartResult, EffectAcpErrors.AcpError>;
162181 /** Stream of parsed ACP session events emitted after startup. */
163 readonly getEvents: () => Stream.Stream<AcpParsedSessionEvent, never>;
182 readonly getEvents: () => Stream.Stream<AcpSessionRuntimeEvent, never>;
183 /** Waits until the current event consumer has processed every queued event. */
184 readonly drainEvents: Effect.Effect<void>;
164185 /** Latest mode state observed from session setup and `session/update` notifications. */
165186 readonly getModeState: Effect.Effect<AcpSessionModeState | undefined>;
166187 /** Latest configuration options observed from session setup and configuration writes. */
......@@ -254,12 +275,17 @@ export const make = (
254275 Effect.gen(function* () {
255276 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
256277 const runtimeScope = yield* Scope.Scope;
257 const eventQueue = yield* Queue.unbounded<AcpParsedSessionEvent>();
278 const eventQueue = yield* Queue.unbounded<AcpSessionRuntimeEvent>();
258279 const modeStateRef = yield* Ref.make<AcpSessionModeState | undefined>(undefined);
259280 const toolCallsRef = yield* Ref.make(new Map<string, AcpToolCallState>());
260281 const assistantSegmentRef = yield* Ref.make<AcpAssistantSegmentState>({ nextSegmentIndex: 0 });
261282 const configOptionsRef = yield* Ref.make(sessionConfigOptionsFromSetup(undefined));
262283 const startStateRef = yield* Ref.make<AcpStartState>({ _tag: "NotStarted" });
284 const promptSerializationSemaphore = yield* Semaphore.make(1);
285 const activePromptFiberRef = yield* Ref.make<
286 Option.Option<Fiber.Fiber<EffectAcpSchema.PromptResponse, EffectAcpErrors.AcpError>>
287 >(Option.none());
288 const sessionLoadGateRef = yield* Ref.make<Option.Option<SessionLoadGate>>(Option.none());
263289
264290 const logRequest = (event: AcpSessionRequestLogEvent) =>
265291 options.requestLogger ? options.requestLogger(event) : Effect.void;
......@@ -331,15 +357,40 @@ export const make = (
331357 const acp = yield* Effect.service(EffectAcpClient.AcpClient).pipe(Effect.provide(acpContext));
332358
333359 yield* acp.handleSessionUpdate((notification) =>
334 handleSessionUpdate({
335 queue: eventQueue,
336 modeStateRef,
337 toolCallsRef,
338 assistantSegmentRef,
339 params: notification,
360 Effect.gen(function* () {
361 const gate = yield* Ref.get(sessionLoadGateRef);
362 if (Option.isSome(gate) && gate.value.active) {
363 const lastActivityAtMillis = yield* Clock.currentTimeMillis;
364 yield* Ref.set(
365 sessionLoadGateRef,
366 Option.some({
367 ...gate.value,
368 lastActivityAtMillis,
369 }),
370 );
371 return;
372 }
373 if (sessionUpdateIsReplay(notification)) {
374 return;
375 }
376 const startState = yield* Ref.get(startStateRef);
377 // One runtime projects one root ACP session. Child-session updates need
378 // explicit lineage routing and must never be flattened into this stream.
379 if (
380 startState._tag !== "Started" ||
381 notification.sessionId !== startState.result.sessionId
382 ) {
383 return;
384 }
385 yield* handleSessionUpdate({
386 queue: eventQueue,
387 modeStateRef,
388 toolCallsRef,
389 assistantSegmentRef,
390 params: notification,
391 });
340392 }),
341393 );
342
343394 const initializeClientCapabilities = {
344395 fs: {
345396 readTextFile: false,
......@@ -499,27 +550,74 @@ export const make = (
499550 cwd: options.cwd,
500551 mcpServers: options.mcpServers ?? [],
501552 } satisfies EffectAcpSchema.LoadSessionRequest;
502 const resumed = yield* runLoggedRequest(
503 "session/load",
504 loadPayload,
505 acp.agent.loadSession(loadPayload),
506 ).pipe(Effect.exit);
507 if (Exit.isSuccess(resumed)) {
508 sessionId = options.resumeSessionId;
509 sessionSetupResult = resumed.value;
510 } else {
511 const createPayload = {
512 cwd: options.cwd,
513 mcpServers: options.mcpServers ?? [],
514 } satisfies EffectAcpSchema.NewSessionRequest;
515 const created = yield* runLoggedRequest(
516 "session/new",
517 createPayload,
518 acp.agent.createSession(createPayload),
553 const sessionLoadTimeout = Duration.fromInputUnsafe(
554 options.sessionLoadTimeout ?? defaultSessionLoadTimeout,
555 );
556 const sessionLoadReplayIdleGap = Duration.fromInputUnsafe(
557 options.sessionLoadReplayIdleGap ?? defaultSessionLoadReplayIdleGap,
558 );
559
560 yield* Ref.set(
561 sessionLoadGateRef,
562 Option.some({
563 active: true,
564 lastActivityAtMillis: undefined,
565 idleGap: sessionLoadReplayIdleGap,
566 initializeResult,
567 }),
568 );
569
570 sessionId = options.resumeSessionId;
571 sessionSetupResult = yield* Effect.gen(function* () {
572 yield* logRequest({
573 method: "session/load",
574 payload: loadPayload,
575 status: "started",
576 });
577
578 const idleFiber = yield* waitForSessionLoadReplayIdle({
579 gateRef: sessionLoadGateRef,
580 }).pipe(Effect.forkIn(runtimeScope));
581 const loaded = yield* Effect.raceFirst(
582 acp.agent.loadSession(loadPayload),
583 Fiber.join(idleFiber),
584 ).pipe(
585 Effect.ensuring(Fiber.interrupt(idleFiber).pipe(Effect.ignore)),
586 Effect.timeoutOption(sessionLoadTimeout),
587 Effect.flatMap((result) =>
588 Option.match(result, {
589 onNone: () =>
590 Effect.fail(
591 new EffectAcpErrors.AcpTransportError({
592 operation: "call-rpc",
593 method: "session/load",
594 detail: "session/load timed out waiting for RPC response or replay idle gap",
595 cause: undefined,
596 }),
597 ),
598 onSome: Effect.succeed,
599 }),
600 ),
601 Effect.tap((result) =>
602 logRequest({
603 method: "session/load",
604 payload: loadPayload,
605 status: "succeeded",
606 result,
607 }),
608 ),
609 Effect.onError((cause) =>
610 logRequest({
611 method: "session/load",
612 payload: loadPayload,
613 status: "failed",
614 cause,
615 }),
616 ),
519617 );
520 sessionId = created.sessionId;
521 sessionSetupResult = created;
522 }
618
619 return loaded;
620 }).pipe(Effect.ensuring(Ref.set(sessionLoadGateRef, Option.none())));
523621 } else {
524622 const createPayload = {
525623 cwd: options.cwd,
......@@ -596,25 +694,48 @@ export const make = (
596694 handleExtNotification: acp.handleExtNotification,
597695 start: () => start,
598696 getEvents: () => Stream.fromQueue(eventQueue),
697 drainEvents: Effect.gen(function* () {
698 const acknowledge = yield* Deferred.make<void>();
699 yield* Queue.offer(eventQueue, {
700 _tag: "EventStreamBarrier",
701 acknowledge,
702 });
703 yield* Deferred.await(acknowledge);
704 }),
599705 getModeState: Ref.get(modeStateRef),
600706 getConfigOptions: Ref.get(configOptionsRef),
601707 prompt: (payload) =>
602 getStartedState.pipe(
603 Effect.flatMap((started) => {
708 promptSerializationSemaphore.withPermit(
709 Effect.gen(function* () {
710 const started = yield* getStartedState;
711 yield* closeActiveAssistantSegment({
712 queue: eventQueue,
713 assistantSegmentRef,
714 });
604715 const requestPayload = {
605716 sessionId: started.sessionId,
606717 ...payload,
607718 } satisfies EffectAcpSchema.PromptRequest;
608 return closeActiveAssistantSegment({
609 queue: eventQueue,
610 assistantSegmentRef,
611 }).pipe(
612 Effect.andThen(
613 runLoggedRequest(
614 "session/prompt",
615 requestPayload,
616 acp.agent.prompt(requestPayload),
617 ),
719 const cancelledResponse = {
720 stopReason: "cancelled",
721 } satisfies EffectAcpSchema.PromptResponse;
722 const promptRpcFiber = yield* runLoggedRequest(
723 "session/prompt",
724 requestPayload,
725 acp.agent.prompt(requestPayload),
726 ).pipe(Effect.forkIn(runtimeScope));
727 yield* Ref.set(activePromptFiberRef, Option.some(promptRpcFiber));
728 return yield* Fiber.join(promptRpcFiber).pipe(
729 Effect.catchCause((cause) =>
730 Cause.hasInterruptsOnly(cause)
731 ? Effect.succeed(cancelledResponse)
732 : Effect.failCause(cause),
733 ),
734 Effect.ensuring(
735 Effect.gen(function* () {
736 yield* Fiber.interrupt(promptRpcFiber).pipe(Effect.ignore);
737 yield* Ref.set(activePromptFiberRef, Option.none());
738 }),
618739 ),
619740 Effect.tap(() =>
620741 closeActiveAssistantSegment({
......@@ -626,7 +747,17 @@ export const make = (
626747 }),
627748 ),
628749 cancel: getStartedState.pipe(
629 Effect.flatMap((started) => acp.agent.cancel({ sessionId: started.sessionId })),
750 Effect.flatMap((started) =>
751 Effect.gen(function* () {
752 const activePromptFiber = yield* Ref.get(activePromptFiberRef);
753 if (Option.isSome(activePromptFiber)) {
754 yield* Fiber.interrupt(activePromptFiber.value).pipe(Effect.ignore);
755 }
756 yield* acp.agent
757 .cancel({ sessionId: started.sessionId })
758 .pipe(Effect.ignore, Effect.forkIn(runtimeScope));
759 }),
760 ),
630761 ),
631762 setMode: (modeId) =>
632763 Ref.get(modeStateRef).pipe(
......@@ -705,7 +836,7 @@ const handleSessionUpdate = ({
705836 assistantSegmentRef,
706837 params,
707838}: {
708 readonly queue: Queue.Queue<AcpParsedSessionEvent>;
839 readonly queue: Queue.Queue<AcpSessionRuntimeEvent>;
709840 readonly modeStateRef: Ref.Ref<AcpSessionModeState | undefined>;
710841 readonly toolCallsRef: Ref.Ref<Map<string, AcpToolCallState>>;
711842 readonly assistantSegmentRef: Ref.Ref<AcpAssistantSegmentState>;
......@@ -801,7 +932,7 @@ const ensureActiveAssistantSegment = ({
801932 assistantSegmentRef,
802933 sessionId,
803934}: {
804 readonly queue: Queue.Queue<AcpParsedSessionEvent>;
935 readonly queue: Queue.Queue<AcpSessionRuntimeEvent>;
805936 readonly assistantSegmentRef: Ref.Ref<AcpAssistantSegmentState>;
806937 readonly sessionId: string;
807938}) =>
......@@ -838,7 +969,7 @@ const closeActiveAssistantSegment = ({
838969 queue,
839970 assistantSegmentRef,
840971}: {
841 readonly queue: Queue.Queue<AcpParsedSessionEvent>;
972 readonly queue: Queue.Queue<AcpSessionRuntimeEvent>;
842973 readonly assistantSegmentRef: Ref.Ref<AcpAssistantSegmentState>;
843974}) =>
844975 Ref.modify(assistantSegmentRef, (current) => {
apps/server/src/provider/acp/GrokAcpSupport.ts+3-1
......@@ -8,6 +8,7 @@ import type * as EffectAcpSchema from "effect-acp/schema";
88import { normalizeModelSlug } from "@t3tools/shared/model";
99
1010import * as AcpSessionRuntime from "./AcpSessionRuntime.ts";
11import { makeXAiPromptCompletionRuntime } from "./XAiAcpExtension.ts";
1112
1213const GROK_API_KEY_ENV = "XAI_API_KEY";
1314const GROK_OAUTH2_REFERRER_ENV = "GROK_OAUTH2_REFERRER";
......@@ -68,9 +69,10 @@ export const makeGrokAcpRuntime = (
6869 ),
6970 ),
7071 );
71 return yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe(
72 const runtime = yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe(
7273 Effect.provide(acpContext),
7374 );
75 return yield* makeXAiPromptCompletionRuntime(runtime);
7476 });
7577
7678export function resolveGrokAcpBaseModelId(model: string | null | undefined): string {
apps/server/src/provider/acp/XAiAcpExtension.test.ts+83-1
......@@ -1,12 +1,39 @@
1import { describe, expect, it } from "vite-plus/test";
1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodePath from "node:path";
3import * as NodeURL from "node:url";
4
5import * as NodeServices from "@effect/platform-node/NodeServices";
6import { it } from "@effect/vitest";
7import * as Effect from "effect/Effect";
28import * as Schema from "effect/Schema";
9import { describe, expect } from "vite-plus/test";
310
411import {
512 extractXAiAskUserQuestions,
613 makeXAiAskUserQuestionCancelledResponse,
714 makeXAiAskUserQuestionResponse,
15 makeXAiPromptCompletionRuntime,
816 XAiAskUserQuestionRequest,
917} from "./XAiAcpExtension.ts";
18import * as AcpSessionRuntime from "./AcpSessionRuntime.ts";
19
20const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url));
21const mockAgentPath = NodePath.join(__dirname, "../../../scripts/acp-mock-agent.ts");
22
23const makePromptCompletionRuntime = (env: NodeJS.ProcessEnv) =>
24 Effect.gen(function* () {
25 const runtime = yield* AcpSessionRuntime.make({
26 spawn: {
27 command: process.execPath,
28 args: [mockAgentPath],
29 env,
30 },
31 cwd: process.cwd(),
32 clientInfo: { name: "t3-test", version: "0.0.0" },
33 authMethodId: "test",
34 });
35 return yield* makeXAiPromptCompletionRuntime(runtime);
36 });
1037
1138const decodeXAiAskUserQuestionRequest = Schema.decodeUnknownSync(XAiAskUserQuestionRequest);
1239
......@@ -247,4 +274,59 @@ describe("XAiAcpExtension", () => {
247274 },
248275 });
249276 });
277
278 it.effect("resolves a hung standard prompt from xAI prompt completion", () =>
279 Effect.gen(function* () {
280 const runtime = yield* makePromptCompletionRuntime({
281 T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG: "1",
282 });
283 yield* runtime.start();
284
285 const promptResult = yield* runtime.prompt({
286 prompt: [{ type: "text", text: "hi" }],
287 });
288 const promptId = promptResult._meta?.promptId;
289
290 expect(typeof promptId).toBe("string");
291 expect(promptResult).toMatchObject({
292 stopReason: "end_turn",
293 _meta: {
294 sessionId: "mock-session-1",
295 promptId,
296 requestId: promptId,
297 },
298 });
299 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
300 );
301
302 it.effect("ignores stale xAI completion from an already settled prompt", () =>
303 Effect.gen(function* () {
304 const runtime = yield* makePromptCompletionRuntime({
305 T3_ACP_EMIT_STALE_XAI_PROMPT_COMPLETE_BEFORE_SECOND_HANG: "1",
306 });
307 yield* runtime.start();
308
309 const firstPromptResult = yield* runtime.prompt({
310 prompt: [{ type: "text", text: "first" }],
311 });
312 expect(firstPromptResult).toMatchObject({
313 stopReason: "end_turn",
314 _meta: { promptId: "mock-stale-xai-prompt-1" },
315 });
316
317 const secondPromptResult = yield* runtime.prompt({
318 prompt: [{ type: "text", text: "second" }],
319 });
320 const secondPromptId = secondPromptResult._meta?.promptId;
321 expect(typeof secondPromptId).toBe("string");
322 expect(secondPromptId).not.toBe("mock-stale-xai-prompt-1");
323 expect(secondPromptResult).toMatchObject({
324 stopReason: "end_turn",
325 _meta: {
326 promptId: secondPromptId,
327 requestId: secondPromptId,
328 },
329 });
330 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
331 );
250332});
apps/server/src/provider/acp/XAiAcpExtension.ts+259
......@@ -1,5 +1,29 @@
11import type { ProviderUserInputAnswers, UserInputQuestion } from "@t3tools/contracts";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Ref from "effect/Ref";
25import * as Schema from "effect/Schema";
6import type * as EffectAcpSchema from "effect-acp/schema";
7
8import type * as AcpSessionRuntime from "./AcpSessionRuntime.ts";
9
10const XAiPromptCompleteNotification = Schema.Struct({
11 sessionId: Schema.String,
12 promptId: Schema.optional(Schema.String),
13 stopReason: Schema.optional(Schema.String),
14 agentResult: Schema.optional(Schema.NullOr(Schema.Unknown)),
15});
16
17type XAiPromptCompleteNotification = typeof XAiPromptCompleteNotification.Type;
18
19interface PendingXAiPromptCompletion {
20 readonly sessionId: string;
21 readonly promptId: string;
22 readonly deferred: Deferred.Deferred<EffectAcpSchema.PromptResponse>;
23}
24
25const completedXAiPromptIdLimit = 128;
26const xAiStopReasonMissingMetaKey = "xAiStopReasonMissing";
327
428const XAiAskUserQuestionOption = Schema.Struct({
529 label: Schema.String,
......@@ -171,3 +195,238 @@ export function makeXAiAskUserQuestionResponse(
171195export function makeXAiAskUserQuestionCancelledResponse(): XAiAskUserQuestionCancelledResponse {
172196 return { outcome: "cancelled" };
173197}
198
199/**
200 * Adds Grok's private prompt-completion fallback around a standards-only ACP runtime.
201 * The underlying runtime remains unaware of xAI methods and metadata.
202 */
203export const makeXAiPromptCompletionRuntime = Effect.fn("makeXAiPromptCompletionRuntime")(
204 function* (runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]) {
205 const activeSessionIdRef = yield* Ref.make<string | undefined>(undefined);
206 const pendingRef = yield* Ref.make<ReadonlyArray<PendingXAiPromptCompletion>>([]);
207 const completedPromptIdsRef = yield* Ref.make<ReadonlyArray<string>>([]);
208 let nextPromptFallbackId = 0;
209 const allocatePromptFallbackId = Effect.sync(() => {
210 nextPromptFallbackId += 1;
211 return `t3-xai-prompt-${nextPromptFallbackId}`;
212 });
213
214 yield* runtime.handleExtNotification(
215 "_x.ai/session/prompt_complete",
216 XAiPromptCompleteNotification,
217 (notification) =>
218 resolveXAiPromptCompletionFallback({
219 pendingRef,
220 completedPromptIdsRef,
221 notification,
222 }),
223 );
224
225 return {
226 ...runtime,
227 start: () =>
228 runtime
229 .start()
230 .pipe(Effect.tap((started) => Ref.set(activeSessionIdRef, started.sessionId))),
231 prompt: (payload) =>
232 Effect.gen(function* () {
233 const sessionId = yield* Ref.get(activeSessionIdRef);
234 if (sessionId === undefined) {
235 return yield* runtime.prompt(payload);
236 }
237
238 const promptId = yield* allocatePromptFallbackId;
239 const fallback = yield* registerXAiPromptCompletionFallback(
240 pendingRef,
241 sessionId,
242 promptId,
243 );
244 const requestPayload = {
245 ...payload,
246 _meta: {
247 ...payload._meta,
248 promptId: fallback.promptId,
249 requestId: fallback.promptId,
250 },
251 } satisfies Omit<EffectAcpSchema.PromptRequest, "sessionId">;
252
253 return yield* Effect.raceFirst(
254 runtime.prompt(requestPayload),
255 Deferred.await(fallback.deferred),
256 ).pipe(
257 Effect.tap((response) =>
258 rememberCompletedXAiPromptId(completedPromptIdsRef, response, fallback.promptId),
259 ),
260 Effect.ensuring(unregisterXAiPromptCompletionFallback(pendingRef, fallback.deferred)),
261 );
262 }),
263 cancel: Ref.get(activeSessionIdRef).pipe(
264 Effect.flatMap((sessionId) =>
265 sessionId === undefined
266 ? runtime.cancel
267 : abortPendingPromptCompletions(pendingRef, sessionId).pipe(
268 Effect.andThen(runtime.cancel),
269 ),
270 ),
271 ),
272 } satisfies AcpSessionRuntime.AcpSessionRuntime["Service"];
273 },
274);
275
276const registerXAiPromptCompletionFallback = (
277 pendingRef: Ref.Ref<ReadonlyArray<PendingXAiPromptCompletion>>,
278 sessionId: string,
279 promptId: string,
280) =>
281 Deferred.make<EffectAcpSchema.PromptResponse>().pipe(
282 Effect.tap((deferred) =>
283 Ref.update(pendingRef, (pending) => [...pending, { sessionId, promptId, deferred }]),
284 ),
285 Effect.map((deferred) => ({ deferred, promptId })),
286 );
287
288const unregisterXAiPromptCompletionFallback = (
289 pendingRef: Ref.Ref<ReadonlyArray<PendingXAiPromptCompletion>>,
290 deferred: Deferred.Deferred<EffectAcpSchema.PromptResponse>,
291) => Ref.update(pendingRef, (pending) => pending.filter((entry) => entry.deferred !== deferred));
292
293const abortPendingPromptCompletions = (
294 pendingRef: Ref.Ref<ReadonlyArray<PendingXAiPromptCompletion>>,
295 sessionId: string,
296) =>
297 Ref.modify(pendingRef, (pending) => {
298 const [toAbort, remaining] = pending.reduce<
299 [ReadonlyArray<PendingXAiPromptCompletion>, ReadonlyArray<PendingXAiPromptCompletion>]
300 >(
301 ([aborting, kept], entry) =>
302 entry.sessionId === sessionId ? [[...aborting, entry], kept] : [aborting, [...kept, entry]],
303 [[], []],
304 );
305 if (toAbort.length === 0) {
306 return [Effect.void, pending] as const;
307 }
308 return [
309 Effect.forEach(
310 toAbort,
311 (entry) =>
312 Deferred.succeed(
313 entry.deferred,
314 promptResponseFromXAi({
315 sessionId: entry.sessionId,
316 promptId: entry.promptId,
317 stopReason: "cancelled",
318 agentResult: null,
319 }),
320 ),
321 { concurrency: "unbounded" },
322 ).pipe(Effect.asVoid),
323 remaining,
324 ] as const;
325 }).pipe(Effect.flatten);
326
327const resolveXAiPromptCompletionFallback = ({
328 pendingRef,
329 completedPromptIdsRef,
330 notification,
331}: {
332 readonly pendingRef: Ref.Ref<ReadonlyArray<PendingXAiPromptCompletion>>;
333 readonly completedPromptIdsRef: Ref.Ref<ReadonlyArray<string>>;
334 readonly notification: XAiPromptCompleteNotification;
335}) =>
336 Ref.get(completedPromptIdsRef).pipe(
337 Effect.flatMap((completedPromptIds) => {
338 if (
339 notification.promptId !== undefined &&
340 completedPromptIds.includes(notification.promptId)
341 ) {
342 return Effect.void;
343 }
344 return Ref.modify(pendingRef, (pending) => {
345 const index =
346 notification.promptId !== undefined
347 ? pending.findIndex(
348 (entry) =>
349 entry.sessionId === notification.sessionId &&
350 entry.promptId === notification.promptId,
351 )
352 : pending.findIndex((entry) => entry.sessionId === notification.sessionId);
353 if (index < 0) {
354 return [Effect.void, pending] as const;
355 }
356 const entry = pending[index];
357 if (!entry) {
358 return [Effect.void, pending] as const;
359 }
360 return [
361 Deferred.succeed(entry.deferred, promptResponseFromXAi(notification)).pipe(Effect.asVoid),
362 [...pending.slice(0, index), ...pending.slice(index + 1)],
363 ] as const;
364 }).pipe(Effect.flatten);
365 }),
366 );
367
368const rememberCompletedXAiPromptId = (
369 completedPromptIdsRef: Ref.Ref<ReadonlyArray<string>>,
370 response: EffectAcpSchema.PromptResponse,
371 fallbackPromptId: string,
372) => {
373 const promptId = promptIdFromResponse(response) ?? fallbackPromptId;
374 return Ref.update(completedPromptIdsRef, (completedPromptIds) => {
375 if (completedPromptIds.includes(promptId)) {
376 return completedPromptIds;
377 }
378 return [...completedPromptIds, promptId].slice(-completedXAiPromptIdLimit);
379 });
380};
381
382function promptIdFromResponse(response: EffectAcpSchema.PromptResponse): string | undefined {
383 const meta = response._meta;
384 if (meta === null || typeof meta !== "object") {
385 return undefined;
386 }
387 const promptId = meta.promptId ?? meta.requestId;
388 return typeof promptId === "string" && promptId.length > 0 ? promptId : undefined;
389}
390
391export function promptResponseHasMissingXAiStopReason(
392 response: EffectAcpSchema.PromptResponse,
393): boolean {
394 const meta = response._meta;
395 return meta !== null && typeof meta === "object" && meta[xAiStopReasonMissingMetaKey] === true;
396}
397
398function promptResponseFromXAi(
399 notification: XAiPromptCompleteNotification,
400): EffectAcpSchema.PromptResponse {
401 const stopReason = normalizeXAiStopReason(notification.stopReason);
402 const meta: Record<string, unknown> = {
403 sessionId: notification.sessionId,
404 };
405 if (notification.stopReason === undefined) {
406 meta[xAiStopReasonMissingMetaKey] = true;
407 }
408 if (notification.promptId !== undefined) {
409 meta.promptId = notification.promptId;
410 meta.requestId = notification.promptId;
411 }
412 if (notification.agentResult !== undefined) {
413 meta.agentResult = notification.agentResult;
414 }
415 return {
416 stopReason,
417 _meta: meta,
418 };
419}
420
421function normalizeXAiStopReason(value: string | undefined): EffectAcpSchema.StopReason {
422 switch (value) {
423 case "cancelled":
424 case "end_turn":
425 case "max_tokens":
426 case "max_turn_requests":
427 case "refusal":
428 return value;
429 default:
430 return "end_turn";
431 }
432}
apps/web/src/components/ChatView.logic.test.ts+25
......@@ -6,6 +6,7 @@ import {
66 MAX_HIDDEN_MOUNTED_PREVIEW_THREADS,
77 MAX_HIDDEN_MOUNTED_TERMINAL_THREADS,
88 buildExpiredTerminalContextToastCopy,
9 buildThreadTurnInterruptInput,
910 createLocalDispatchSnapshot,
1011 deriveComposerSendState,
1112 getStartedThreadModelChangeBlockReason,
......@@ -69,6 +70,30 @@ const readySession = {
6970 updatedAt: "2026-03-29T00:00:10.000Z",
7071};
7172
73describe("buildThreadTurnInterruptInput", () => {
74 it("targets the session's active running turn", () => {
75 const activeTurnId = TurnId.make("turn-running");
76
77 expect(
78 buildThreadTurnInterruptInput(
79 makeThread({
80 session: {
81 ...readySession,
82 status: "running",
83 activeTurnId,
84 },
85 }),
86 ),
87 ).toEqual({ threadId, turnId: activeTurnId });
88 });
89
90 it("omits a turn id when the session is not running", () => {
91 expect(buildThreadTurnInterruptInput(makeThread({ session: readySession }))).toEqual({
92 threadId,
93 });
94 });
95});
96
7297describe("deriveComposerSendState", () => {
7398 it("treats expired terminal pills as non-sendable content", () => {
7499 const state = deriveComposerSendState({
apps/web/src/components/ChatView.logic.ts+11
......@@ -74,6 +74,17 @@ export function shouldWriteThreadErrorToCurrentServerThread(input: {
7474 );
7575}
7676
77export function buildThreadTurnInterruptInput(thread: Pick<Thread, "id" | "session">): {
78 threadId: ThreadId;
79 turnId?: TurnId;
80} {
81 const runningTurnId = thread.session?.status === "running" ? thread.session.activeTurnId : null;
82 return {
83 threadId: thread.id,
84 ...(runningTurnId !== null ? { turnId: runningTurnId } : {}),
85 };
86}
87
7788export function reconcileMountedTerminalThreadIds(input: {
7889 currentThreadIds: ReadonlyArray<string>;
7990 openThreadIds: ReadonlyArray<string>;
apps/web/src/components/ChatView.tsx+7-3
......@@ -214,6 +214,7 @@ import {
214214 MAX_HIDDEN_MOUNTED_TERMINAL_THREADS,
215215 buildExpiredTerminalContextToastCopy,
216216 buildLocalDraftThread,
217 buildThreadTurnInterruptInput,
217218 collectUserMessageBlobPreviewUrls,
218219 createLocalDispatchSnapshot,
219220 deriveComposerSendState,
......@@ -3941,9 +3942,7 @@ function ChatViewContent(props: ChatViewProps) {
39413942 if (!activeThread) return;
39423943 const result = await interruptThreadTurn({
39433944 environmentId,
3944 input: {
3945 threadId: activeThread.id,
3946 },
3945 input: buildThreadTurnInterruptInput(activeThread),
39473946 });
39483947 if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) {
39493948 const error = squashAtomCommandFailure(result);
......@@ -4771,6 +4770,11 @@ function ChatViewContent(props: ChatViewProps) {
47714770 listRef={legendListRef}
47724771 timelineEntries={timelineEntries}
47734772 latestTurn={activeLatestTurn}
4773 runningTurnId={
4774 activeThread.session?.status === "running"
4775 ? activeThread.session.activeTurnId
4776 : null
4777 }
47744778 turnDiffSummaryByAssistantMessageId={turnDiffSummaryByAssistantMessageId}
47754779 activeThreadEnvironmentId={activeThread.environmentId}
47764780 routeThreadKey={routeThreadKey}
apps/web/src/components/chat/MessagesTimeline.logic.test.ts+61
......@@ -794,6 +794,67 @@ describe("deriveMessagesTimelineRows", () => {
794794 ]);
795795 });
796796
797 it("does not fold the session's running turn when latestTurn regresses", () => {
798 const rows = deriveMessagesTimelineRows({
799 timelineEntries: [
800 {
801 id: "previous-work-entry",
802 kind: "work",
803 createdAt: "2026-01-01T00:00:05Z",
804 entry: {
805 id: "previous-work",
806 createdAt: "2026-01-01T00:00:05Z",
807 turnId: "turn-1" as never,
808 label: "Read files",
809 tone: "tool" as const,
810 },
811 },
812 {
813 id: "user-followup-entry",
814 kind: "message",
815 createdAt: "2026-01-01T00:01:00Z",
816 message: {
817 id: "user-followup" as never,
818 role: "user",
819 text: "continue",
820 turnId: null,
821 createdAt: "2026-01-01T00:01:00Z",
822 updatedAt: "2026-01-01T00:01:00Z",
823 streaming: false,
824 },
825 },
826 {
827 id: "running-work-entry",
828 kind: "work",
829 createdAt: "2026-01-01T00:01:05Z",
830 entry: {
831 id: "running-work",
832 createdAt: "2026-01-01T00:01:05Z",
833 turnId: "turn-2" as never,
834 label: "Searched files",
835 tone: "tool" as const,
836 },
837 },
838 ],
839 latestTurn: {
840 turnId: "turn-1" as never,
841 state: "completed",
842 startedAt: "2026-01-01T00:00:00Z",
843 completedAt: "2026-01-01T00:00:25Z",
844 },
845 runningTurnId: "turn-2" as never,
846 isWorking: true,
847 activeTurnStartedAt: "2026-01-01T00:01:00Z",
848 turnDiffSummaryByAssistantMessageId: new Map(),
849 revertTurnCountByUserMessageId: new Map(),
850 });
851
852 expect(rows.filter((row) => row.kind === "turn-fold").map((row) => row.turnId)).toEqual([
853 "turn-1",
854 ]);
855 expect(rows.map((row) => row.id)).toContain("running-work-entry");
856 });
857
797858 it("only shows assistant metadata on the terminal assistant message", () => {
798859 const rows = deriveMessagesTimelineRows({
799860 timelineEntries: [
apps/web/src/components/chat/MessagesTimeline.logic.ts+18-7
......@@ -149,13 +149,20 @@ interface TurnFold {
149149}
150150
151151/**
152 * The latest turn counts as unsettled while it is still running (or has not
153 * recorded a completion). This is deliberately keyed on the turn's own
154 * lifecycle rather than transient working state: right after the user sends
155 * a message, the previous turn is still the "active" one until the server
156 * creates the new turn, and folding must not flicker through that window.
152 * The session's running turn is authoritative when latestTurn briefly lags or
153 * regresses behind it. Otherwise, the latest turn counts as unsettled while it
154 * is still running (or has not recorded a completion). This is deliberately
155 * keyed on turn lifecycle rather than transient working state: right after the
156 * user sends a message, the previous turn is still the "active" one until the
157 * server creates the new turn, and folding must not flicker through that window.
157158 */
158function deriveUnsettledTurnId(latestTurn: TimelineLatestTurn | null): TurnId | null {
159function deriveUnsettledTurnId(
160 latestTurn: TimelineLatestTurn | null,
161 runningTurnId: TurnId | null,
162): TurnId | null {
163 if (runningTurnId !== null) {
164 return runningTurnId;
165 }
159166 if (!latestTurn) {
160167 return null;
161168 }
......@@ -291,6 +298,7 @@ function deriveTurnFolds(input: {
291298export function deriveMessagesTimelineRows(input: {
292299 timelineEntries: ReadonlyArray<TimelineEntry>;
293300 latestTurn?: TimelineLatestTurn | null;
301 runningTurnId?: TurnId | null;
294302 expandedTurnIds?: ReadonlySet<TurnId>;
295303 isWorking: boolean;
296304 activeTurnStartedAt: string | null;
......@@ -302,7 +310,10 @@ export function deriveMessagesTimelineRows(input: {
302310 input.timelineEntries.flatMap((entry) => (entry.kind === "message" ? [entry.message] : [])),
303311 );
304312 const terminalAssistantMessageIds = deriveTerminalAssistantMessageIds(input.timelineEntries);
305 const unsettledTurnId = deriveUnsettledTurnId(input.latestTurn ?? null);
313 const unsettledTurnId = deriveUnsettledTurnId(
314 input.latestTurn ?? null,
315 input.runningTurnId ?? null,
316 );
306317 const foldsByAnchorEntryId = deriveTurnFolds({
307318 timelineEntries: input.timelineEntries,
308319 terminalAssistantMessageIds,
apps/web/src/components/chat/MessagesTimeline.test.tsx+1
......@@ -109,6 +109,7 @@ function buildProps() {
109109 activeTurnStartedAt: null,
110110 listRef: createRef<LegendListRef | null>(),
111111 latestTurn: null,
112 runningTurnId: null,
112113 turnDiffSummaryByAssistantMessageId: new Map(),
113114 routeThreadKey: "environment-local:thread-1",
114115 onOpenTurnDiff: () => {},
apps/web/src/components/chat/MessagesTimeline.tsx+4
......@@ -153,6 +153,7 @@ interface MessagesTimelineProps {
153153 listRef: React.RefObject<LegendListRef | null>;
154154 timelineEntries: ReturnType<typeof deriveTimelineEntries>;
155155 latestTurn: TimelineLatestTurn | null;
156 runningTurnId: TurnId | null;
156157 turnDiffSummaryByAssistantMessageId: Map<MessageId, TurnDiffSummary>;
157158 routeThreadKey: string;
158159 onOpenTurnDiff: (turnId: TurnId, filePath?: string) => void;
......@@ -182,6 +183,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({
182183 listRef,
183184 timelineEntries,
184185 latestTurn,
186 runningTurnId,
185187 turnDiffSummaryByAssistantMessageId,
186188 routeThreadKey,
187189 onOpenTurnDiff,
......@@ -272,6 +274,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({
272274 deriveMessagesTimelineRows({
273275 timelineEntries,
274276 latestTurn,
277 runningTurnId,
275278 expandedTurnIds,
276279 isWorking,
277280 activeTurnStartedAt,
......@@ -281,6 +284,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({
281284 [
282285 timelineEntries,
283286 latestTurn,
287 runningTurnId,
284288 expandedTurnIds,
285289 isWorking,
286290 activeTurnStartedAt,
packages/effect-acp/src/client.test.ts+128-1
......@@ -17,18 +17,59 @@ import { it, assert } from "@effect/vitest";
1717import * as AcpClient from "./client.ts";
1818import * as AcpSchema from "./_generated/schema.gen.ts";
1919import * as AcpError from "./errors.ts";
20import { encodeJsonl, jsonRpcRequest, jsonRpcResponse } from "./_internal/shared.ts";
20import {
21 encodeJsonl,
22 jsonRpcNotification,
23 jsonRpcRequest,
24 jsonRpcResponse,
25} from "./_internal/shared.ts";
2126import { makeInMemoryStdio } from "./_internal/stdio.ts";
2227
2328const InitializeRequest = jsonRpcRequest("initialize", AcpSchema.InitializeRequest);
2429const InitializeResponse = jsonRpcResponse(AcpSchema.InitializeResponse);
2530const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String }));
2631const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));
32const PromptRequest = jsonRpcRequest("session/prompt", AcpSchema.PromptRequest);
33const PromptResponse = jsonRpcResponse(AcpSchema.PromptResponse);
34const decodePromptRequestLine = Schema.decodeEffect(Schema.fromJsonString(PromptRequest));
35const XAiPromptCompleteNotification = jsonRpcNotification(
36 "_x.ai/session/prompt_complete",
37 Schema.Struct({
38 sessionId: Schema.String,
39 promptId: Schema.String,
40 stopReason: Schema.String,
41 agentResult: Schema.NullOr(Schema.Unknown),
42 }),
43);
44const XAiQueueChangedNotification = jsonRpcNotification(
45 "_x.ai/queue/changed",
46 Schema.Struct({
47 sessionId: Schema.String,
48 entries: Schema.Array(Schema.Unknown),
49 }),
50);
51const XAiSessionsChangedNotification = jsonRpcNotification(
52 "_x.ai/sessions/changed",
53 Schema.Struct({
54 upserted: Schema.Array(Schema.Unknown),
55 removed: Schema.Array(Schema.Unknown),
56 }),
57);
2758const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) =>
2859 path.join(import.meta.dirname, "../test/fixtures/acp-mock-peer.ts"),
2960);
3061const mockPeerArgs = (path: string) => [path];
3162
63function concatBytes(chunks: ReadonlyArray<Uint8Array>): Uint8Array {
64 const batch = new Uint8Array(chunks.reduce((total, chunk) => total + chunk.length, 0));
65 let offset = 0;
66 for (const chunk of chunks) {
67 batch.set(chunk, offset);
68 offset += chunk.length;
69 }
70 return batch;
71}
72
3273it.layer(NodeServices.layer)("effect-acp client", (it) => {
3374 const makeHandle = (env?: Record<string, string>) =>
3475 Effect.gen(function* () {
......@@ -446,4 +487,90 @@ it.layer(NodeServices.layer)("effect-acp client", (it) => {
446487 yield* Scope.close(scope, Exit.void);
447488 }),
448489 );
490
491 it.effect(
492 "routes a standard prompt response after Grok extension notifications in the same batch",
493 () =>
494 Effect.gen(function* () {
495 const { stdio, input, output } = yield* makeInMemoryStdio();
496 const scope = yield* Scope.make();
497 const acp = yield* AcpClient.make(stdio).pipe(Effect.provideService(Scope.Scope, scope));
498
499 const promptFiber = yield* acp.agent
500 .prompt({
501 sessionId: "grok-session-1",
502 prompt: [{ type: "text", text: "run the ls command" }],
503 })
504 .pipe(Effect.forkScoped);
505
506 const outbound = yield* Queue.take(output);
507 const decodedPrompt = yield* decodePromptRequestLine(outbound);
508
509 const responseBatch = concatBytes(
510 yield* Effect.all([
511 encodeJsonl(XAiQueueChangedNotification, {
512 jsonrpc: "2.0",
513 method: "_x.ai/queue/changed",
514 params: { sessionId: "grok-session-1", entries: [] },
515 }),
516 encodeJsonl(XAiPromptCompleteNotification, {
517 jsonrpc: "2.0",
518 method: "_x.ai/session/prompt_complete",
519 params: {
520 sessionId: "grok-session-1",
521 promptId: "prompt-1",
522 stopReason: "end_turn",
523 agentResult: null,
524 },
525 }),
526 encodeJsonl(XAiSessionsChangedNotification, {
527 jsonrpc: "2.0",
528 method: "_x.ai/sessions/changed",
529 params: {
530 upserted: [
531 {
532 sessionId: "grok-session-1",
533 title: null,
534 cwd: process.cwd(),
535 isWorktree: false,
536 modelId: "grok-composer-2.5-fast",
537 yolo: false,
538 activity: "idle",
539 resident: true,
540 lastChangeUnixMs: 1_710_000_000_000,
541 origin: { kind: "local" },
542 },
543 ],
544 removed: [],
545 },
546 }),
547 encodeJsonl(PromptResponse, {
548 jsonrpc: "2.0",
549 id: decodedPrompt.id,
550 result: {
551 stopReason: "end_turn",
552 _meta: {
553 sessionId: "grok-session-1",
554 requestId: "prompt-1",
555 promptId: "prompt-1",
556 modelId: "grok-composer-2.5-fast",
557 },
558 },
559 }),
560 ]),
561 );
562 yield* Queue.offer(input, responseBatch);
563
564 assert.deepEqual(yield* Fiber.join(promptFiber), {
565 stopReason: "end_turn",
566 _meta: {
567 sessionId: "grok-session-1",
568 requestId: "prompt-1",
569 promptId: "prompt-1",
570 modelId: "grok-composer-2.5-fast",
571 },
572 });
573 yield* Scope.close(scope, Exit.void);
574 }),
575 );
449576});