From 216eb4f66ddcae1d58a475a5e20e737ede878ab2 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 7 Sep 2026 23:06:56 +1000 Subject: [PATCH 1/3] fix(grok): retire crashed sessions before retrying --- apps/server/scripts/acp-mock-agent.ts | 6 ++++ .../src/provider/Layers/GrokAdapter.test.ts | 36 +++++++++++++++++++ .../server/src/provider/Layers/GrokAdapter.ts | 35 ++++++++++++++++-- 3 files changed, 74 insertions(+), 3 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index c923fa401021..90af1f9f2675 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -617,6 +617,12 @@ const program = Effect.gen(function* () { Effect.gen(function* () { const requestedSessionId = String(request.sessionId ?? sessionId); promptCount += 1; + if ( + process.env.T3_ACP_CRASH_PROMPT === "1" && + request.prompt.some((part) => part.type === "text" && part.text === "crash now") + ) { + return yield* Effect.sync(() => process.exit(23)); + } if (completeFirstPromptOnCancel && promptCount === 1) { yield* agent.client.sessionUpdate({ diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 9efb28d80628..964d059d8635 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -294,6 +294,42 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { }), ); + it.effect("retires a crashed process so a deliberate retry can resume", () => + Effect.gen(function* () { + const threadId = ThreadId.make("grok-crash-recovery"); + const wrapper = yield* Effect.promise(() => + makeMockGrokWrapper({ T3_ACP_CRASH_PROMPT: "1" }), + ); + const adapter = yield* makeTestAdapter(wrapper); + const exited = + yield* Deferred.make>(); + const events = yield* Stream.runForEach(adapter.streamEvents, (event) => + event.type === "session.exited" + ? Deferred.succeed(exited, event).pipe(Effect.asVoid) + : Effect.void, + ).pipe(Effect.forkChild); + const input = { + threadId, + provider: ProviderDriverKind.make("grok"), + cwd: process.cwd(), + runtimeMode: "full-access" as const, + }; + const session = yield* adapter.startSession(input); + const failure = yield* Effect.flip( + adapter.sendTurn({ threadId, input: "crash now", attachments: [] }), + ); + assert.isDefined(failure); + assert.isFalse(yield* adapter.hasSession(threadId)); + assert.equal((yield* Deferred.await(exited)).payload.exitKind, "error"); + assert.deepStrictEqual(yield* adapter.listSessions(), []); + yield* Fiber.interrupt(events); + yield* adapter.startSession({ ...input, resumeCursor: session.resumeCursor }); + const turn = yield* adapter.sendTurn({ threadId, input: "retry now", attachments: [] }); + assert.equal(turn.threadId, threadId); + yield* adapter.stopSession(threadId); + }), + ); + it.effect("starts a session and maps mock ACP prompt flow to runtime events", () => Effect.gen(function* () { const threadId = ThreadId.make("grok-mock-thread"); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 395d7e546f9a..11b1dbf89476 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -171,6 +171,7 @@ interface GrokSessionContext { currentModelId: string | undefined; currentReasoningEffort: string | undefined; stopped: boolean; + terminated: boolean; } function settlePendingApprovalsAsCancelled( @@ -339,6 +340,7 @@ export function grokPromptSettlementBelongsToContext(input: { export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapterLiveOptions) { return Effect.gen(function* () { + const ownerScope = yield* Effect.scope; const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("grok"); const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; @@ -939,7 +941,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte ...(yield* makeEventStamp()), provider: PROVIDER, threadId: ctx.threadId, - payload: { exitKind: "graceful" }, + payload: { exitKind: ctx.terminated ? "error" : "graceful" }, }); }); @@ -1309,11 +1311,34 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte ? normalizeGrokReasoningEffort(requestedStartReasoningEffort) : currentStartReasoningEffort, stopped: false, + terminated: false, }; const nf = yield* Stream.runDrain( Stream.mapEffect(acp.getEvents(), (event) => Effect.gen(function* () { + if (event._tag === "ConnectionTerminated") { + ctx.terminated = true; + yield* withThreadLock( + ctx.threadId, + Effect.gen(function* () { + if (sessions.get(ctx.threadId) !== ctx) return; + if (ctx.activeTurnId) { + yield* settlePromptInFlight( + ctx.threadId, + ctx.activeTurnId, + ctx.acpSessionId, + { + errorMessage: "Grok connection terminated.", + settleAllPrompts: true, + }, + ); + } + yield* stopSessionInternal(ctx); + }), + ).pipe(Effect.forkIn(ownerScope)); + return; + } if (event._tag === "EventStreamBarrier") { yield* Deferred.succeed(event.acknowledge, undefined); return; @@ -2112,12 +2137,16 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte ); const listSessions: GrokAdapterShape["listSessions"] = () => - Effect.sync(() => Array.from(sessions.values(), (c) => ({ ...c.session }))); + Effect.sync(() => + Array.from(sessions.values()) + .filter((c) => !c.terminated) + .map((c) => ({ ...c.session })), + ); const hasSession: GrokAdapterShape["hasSession"] = (threadId) => Effect.sync(() => { const c = sessions.get(threadId); - return c !== undefined && !c.stopped; + return c !== undefined && !c.stopped && !c.terminated; }); const stopAll: GrokAdapterShape["stopAll"] = () => From 707fd371df1961400b1d787f34335b34755205c5 Mon Sep 17 00:00:00 2001 From: SAPHID Date: Mon, 7 Sep 2026 14:45:35 +0000 Subject: [PATCH 2/3] fix(grok): reject retries during session teardown --- apps/server/src/provider/Layers/GrokAdapter.test.ts | 4 ++++ apps/server/src/provider/Layers/GrokAdapter.ts | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 964d059d8635..794f95cf58be 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -320,6 +320,10 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { ); assert.isDefined(failure); assert.isFalse(yield* adapter.hasSession(threadId)); + const retryDuringTeardown = yield* Effect.flip( + adapter.sendTurn({ threadId, input: "retry during teardown", attachments: [] }), + ); + assert.equal(retryDuringTeardown._tag, "ProviderAdapterSessionNotFoundError"); assert.equal((yield* Deferred.await(exited)).payload.exitKind, "error"); assert.deepStrictEqual(yield* adapter.listSessions(), []); yield* Fiber.interrupt(events); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 11b1dbf89476..579c18dd01b1 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -917,7 +917,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte threadId: ThreadId, ): Effect.Effect => { const ctx = sessions.get(threadId); - if (!ctx || ctx.stopped) { + if (!ctx || ctx.stopped || ctx.terminated) { return Effect.fail( new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId }), ); From 41f4dd0e9cb3d906910a10f96d4cc8a34e57fe0f Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Tue, 8 Sep 2026 17:14:05 +1000 Subject: [PATCH 3/3] test(grok): assert retry loads the saved ACP session --- .../src/provider/Layers/GrokAdapter.test.ts | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 794f95cf58be..e17d53259840 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -297,8 +297,15 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { it.effect("retires a crashed process so a deliberate retry can resume", () => Effect.gen(function* () { const threadId = ThreadId.make("grok-crash-recovery"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "grok-crash-recovery-")), + ); + const requestLogPath = NodePath.join(tempDir, "requests.ndjson"); const wrapper = yield* Effect.promise(() => - makeMockGrokWrapper({ T3_ACP_CRASH_PROMPT: "1" }), + makeMockGrokWrapper({ + T3_ACP_CRASH_PROMPT: "1", + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + }), ); const adapter = yield* makeTestAdapter(wrapper); const exited = @@ -331,6 +338,19 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { const turn = yield* adapter.sendTurn({ threadId, input: "retry now", attachments: [] }); assert.equal(turn.threadId, threadId); yield* adapter.stopSession(threadId); + const requests = yield* Effect.promise(() => readJsonLines(requestLogPath)); + assert.equal(requests.filter((request) => request.method === "session/new").length, 1); + assert.deepStrictEqual(session.resumeCursor, { + schemaVersion: 1, + sessionId: "mock-session-1", + }); + const resumes = requests.filter((request) => request.method === "session/load"); + assert.equal(resumes.length, 1); + assert.deepStrictEqual(resumes[0]?.params, { + sessionId: "mock-session-1", + cwd: process.cwd(), + mcpServers: [], + }); }), );