authorgravatar for saphid@gmail.comAlex Southwell <saphid@gmail.com> 2026-09-30 10:09:03+10:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-29 21:09:03-03:00
log050cfad04f673fb092630614b8f7e82000b43ea5
treebc540f71e910ece36e2fbf85e7783487f11683d8
parent55ec55b1fb2e65f2736ff8420ae0373210c1911e
signature Signed by PGP key B5690EEEBB952194

fix(grok): recover from crashed provider sessions (#10607)

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: SAPHID <saphid@users.noreply.github.com>

3 files changed, 99 insertions(+), 4 deletions(-)

apps/server/scripts/acp-mock-agent.ts+6
......@@ -621,6 +621,12 @@ const program = Effect.gen(function* () {
621621 Effect.gen(function* () {
622622 const requestedSessionId = String(request.sessionId ?? sessionId);
623623 promptCount += 1;
624 if (
625 process.env.T3_ACP_CRASH_PROMPT === "1" &&
626 request.prompt.some((part) => part.type === "text" && part.text === "crash now")
627 ) {
628 return yield* Effect.sync(() => process.exit(23));
629 }
624630
625631 if (completeFirstPromptOnCancel && promptCount === 1) {
626632 yield* agent.client.sessionUpdate({
apps/server/src/provider/Layers/GrokAdapter.test.ts+60
......@@ -363,6 +363,66 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
363363 }),
364364 );
365365
366 it.effect("retires a crashed process so a deliberate retry can resume", () =>
367 Effect.gen(function* () {
368 const threadId = ThreadId.make("grok-crash-recovery");
369 const tempDir = yield* Effect.promise(() =>
370 NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "grok-crash-recovery-")),
371 );
372 const requestLogPath = NodePath.join(tempDir, "requests.ndjson");
373 const wrapper = yield* Effect.promise(() =>
374 makeMockGrokWrapper({
375 T3_ACP_CRASH_PROMPT: "1",
376 T3_ACP_REQUEST_LOG_PATH: requestLogPath,
377 }),
378 );
379 const adapter = yield* makeTestAdapter(wrapper);
380 const exited =
381 yield* Deferred.make<Extract<ProviderRuntimeEvent, { type: "session.exited" }>>();
382 const events = yield* Stream.runForEach(adapter.streamEvents, (event) =>
383 event.type === "session.exited"
384 ? Deferred.succeed(exited, event).pipe(Effect.asVoid)
385 : Effect.void,
386 ).pipe(Effect.forkChild);
387 const input = {
388 threadId,
389 provider: ProviderDriverKind.make("grok"),
390 cwd: process.cwd(),
391 runtimeMode: "full-access" as const,
392 };
393 const session = yield* adapter.startSession(input);
394 const failure = yield* Effect.flip(
395 adapter.sendTurn({ threadId, input: "crash now", attachments: [] }),
396 );
397 assert.isDefined(failure);
398 assert.isFalse(yield* adapter.hasSession(threadId));
399 const retryDuringTeardown = yield* Effect.flip(
400 adapter.sendTurn({ threadId, input: "retry during teardown", attachments: [] }),
401 );
402 assert.equal(retryDuringTeardown._tag, "ProviderAdapterSessionNotFoundError");
403 assert.equal((yield* Deferred.await(exited)).payload.exitKind, "error");
404 assert.deepStrictEqual(yield* adapter.listSessions(), []);
405 yield* Fiber.interrupt(events);
406 yield* adapter.startSession({ ...input, resumeCursor: session.resumeCursor });
407 const turn = yield* adapter.sendTurn({ threadId, input: "retry now", attachments: [] });
408 assert.equal(turn.threadId, threadId);
409 yield* adapter.stopSession(threadId);
410 const requests = yield* Effect.promise(() => readJsonLines(requestLogPath));
411 assert.equal(requests.filter((request) => request.method === "session/new").length, 1);
412 assert.deepStrictEqual(session.resumeCursor, {
413 schemaVersion: 1,
414 sessionId: "mock-session-1",
415 });
416 const resumes = requests.filter((request) => request.method === "session/load");
417 assert.equal(resumes.length, 1);
418 assert.deepStrictEqual(resumes[0]?.params, {
419 sessionId: "mock-session-1",
420 cwd: process.cwd(),
421 mcpServers: [],
422 });
423 }),
424 );
425
366426 it.effect("starts a session and maps mock ACP prompt flow to runtime events", () =>
367427 Effect.gen(function* () {
368428 const threadId = ThreadId.make("grok-mock-thread");
apps/server/src/provider/Layers/GrokAdapter.ts+33-4
......@@ -175,6 +175,7 @@ interface GrokSessionContext {
175175 currentModelId: string | undefined;
176176 currentReasoningEffort: string | undefined;
177177 stopped: boolean;
178 terminated: boolean;
178179 /** Live monitor/shell identities and their originating turns. */
179180 readonly backgroundTasks: Map<string, GrokBackgroundTaskRecord>;
180181}
......@@ -345,6 +346,7 @@ export function grokPromptSettlementBelongsToContext(input: {
345346
346347export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapterLiveOptions) {
347348 return Effect.gen(function* () {
349 const ownerScope = yield* Effect.scope;
348350 const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("grok");
349351 const fileSystem = yield* FileSystem.FileSystem;
350352 const path = yield* Path.Path;
......@@ -922,7 +924,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
922924 threadId: ThreadId,
923925 ): Effect.Effect<GrokSessionContext, ProviderAdapterSessionNotFoundError> => {
924926 const ctx = sessions.get(threadId);
925 if (!ctx || ctx.stopped) {
927 if (!ctx || ctx.stopped || ctx.terminated) {
926928 return Effect.fail(
927929 new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId }),
928930 );
......@@ -946,7 +948,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
946948 ...(yield* makeEventStamp()),
947949 provider: PROVIDER,
948950 threadId: ctx.threadId,
949 payload: { exitKind: "graceful" },
951 payload: { exitKind: ctx.terminated ? "error" : "graceful" },
950952 });
951953 });
952954
......@@ -1316,12 +1318,35 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
13161318 ? normalizeGrokReasoningEffort(requestedStartReasoningEffort)
13171319 : currentStartReasoningEffort,
13181320 stopped: false,
1321 terminated: false,
13191322 backgroundTasks: new Map(),
13201323 };
13211324
13221325 const nf = yield* Stream.runDrain(
13231326 Stream.mapEffect(acp.getEvents(), (event) =>
13241327 Effect.gen(function* () {
1328 if (event._tag === "ConnectionTerminated") {
1329 ctx.terminated = true;
1330 yield* withThreadLock(
1331 ctx.threadId,
1332 Effect.gen(function* () {
1333 if (sessions.get(ctx.threadId) !== ctx) return;
1334 if (ctx.activeTurnId) {
1335 yield* settlePromptInFlight(
1336 ctx.threadId,
1337 ctx.activeTurnId,
1338 ctx.acpSessionId,
1339 {
1340 errorMessage: "Grok connection terminated.",
1341 settleAllPrompts: true,
1342 },
1343 );
1344 }
1345 yield* stopSessionInternal(ctx);
1346 }),
1347 ).pipe(Effect.forkIn(ownerScope));
1348 return;
1349 }
13251350 if (event._tag === "EventStreamBarrier") {
13261351 yield* Deferred.succeed(event.acknowledge, undefined);
13271352 return;
......@@ -2165,12 +2190,16 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
21652190 );
21662191
21672192 const listSessions: GrokAdapterShape["listSessions"] = () =>
2168 Effect.sync(() => Array.from(sessions.values(), (c) => ({ ...c.session })));
2193 Effect.sync(() =>
2194 Array.from(sessions.values())
2195 .filter((c) => !c.terminated)
2196 .map((c) => ({ ...c.session })),
2197 );
21692198
21702199 const hasSession: GrokAdapterShape["hasSession"] = (threadId) =>
21712200 Effect.sync(() => {
21722201 const c = sessions.get(threadId);
2173 return c !== undefined && !c.stopped;
2202 return c !== undefined && !c.stopped && !c.terminated;
21742203 });
21752204
21762205 const stopAll: GrokAdapterShape["stopAll"] = () =>