diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 6cb71660a74c..ab81c37b85fb 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -26,7 +26,11 @@ import { } from "@t3tools/contracts"; import { ServerConfig } from "../../config.ts"; -import { grokPromptSettlementBelongsToContext, makeGrokAdapter } from "./GrokAdapter.ts"; +import { + grokPromptSettlementBelongsToContext, + grokTurnCompletionForPromptEpoch, + makeGrokAdapter, +} from "./GrokAdapter.ts"; const decodeGrokSettings = Schema.decodeSync(GrokSettings); const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); @@ -122,6 +126,58 @@ it("requires a settlement to match the live Grok turn", () => { ); }); +it("emits the current epoch result when the cancelled prompt drains first", () => { + const completed = { completedStopReason: "end_turn" as const }; + const cancelled = { completedStopReason: "cancelled" as const }; + const afterSuperseded = grokTurnCompletionForPromptEpoch({ + promptEpoch: 1, + discardBeforeEpoch: 2, + remainingPrompts: 1, + stored: undefined, + incoming: cancelled, + emitTurnCompletion: false, + }); + assert.isUndefined(afterSuperseded.stored); + assert.isUndefined(afterSuperseded.emit); + + const afterCurrent = grokTurnCompletionForPromptEpoch({ + promptEpoch: 2, + discardBeforeEpoch: 2, + remainingPrompts: 0, + stored: afterSuperseded.stored, + incoming: completed, + emitTurnCompletion: true, + }); + assert.deepEqual(afterCurrent.stored, completed); + assert.deepEqual(afterCurrent.emit, completed); +}); + +it("emits the current epoch result when the cancelled prompt drains last", () => { + const completed = { completedStopReason: "end_turn" as const }; + const cancelled = { completedStopReason: "cancelled" as const }; + const afterCurrent = grokTurnCompletionForPromptEpoch({ + promptEpoch: 2, + discardBeforeEpoch: 2, + remainingPrompts: 1, + stored: undefined, + incoming: completed, + emitTurnCompletion: true, + }); + assert.deepEqual(afterCurrent.stored, completed); + assert.isUndefined(afterCurrent.emit); + + const afterSuperseded = grokTurnCompletionForPromptEpoch({ + promptEpoch: 1, + discardBeforeEpoch: 2, + remainingPrompts: 0, + stored: afterCurrent.stored, + incoming: cancelled, + emitTurnCompletion: false, + }); + assert.deepEqual(afterSuperseded.stored, completed); + assert.deepEqual(afterSuperseded.emit, completed); +}); + it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { it.effect("starts a session and maps mock ACP prompt flow to runtime events", () => Effect.gen(function* () { @@ -1264,4 +1320,247 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { // hang until the suite timeout instead of failing here. }).pipe(TestClock.withLive), ); + + it.effect("cancels an in-flight prompt when a mid-turn sendTurn steers", () => + Effect.gen(function* () { + const threadId = ThreadId.make("grok-steer-cancels-in-flight"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "grok-acp-steer-")), + ); + const requestLogPath = NodePath.join(tempDir, "requests.ndjson"); + const wrapperPath = yield* Effect.promise(() => + makeMockGrokWrapper({ + T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1", + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const firstTurnStarted = yield* Deferred.make(); + const turnCompleted = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.gen(function* () { + runtimeEvents.push(event); + if (String(event.threadId) !== String(threadId)) { + return; + } + if (event.type === "turn.started" && event.turnId !== undefined) { + yield* Deferred.succeed(firstTurnStarted, event.turnId).pipe(Effect.ignore); + return; + } + if (event.type === "turn.completed") { + yield* Deferred.succeed(turnCompleted, undefined).pipe(Effect.ignore); + } + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("grok"), + cwd: process.cwd(), + runtimeMode: "full-access", + }); + + const firstSendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "hang until steered", attachments: [] }) + .pipe(Effect.forkChild); + const firstTurnId = yield* Deferred.await(firstTurnStarted).pipe(Effect.timeout("2 seconds")); + yield* waitForFileContent(requestLogPath, 80, '"method":"session/prompt"'); + + const steered = yield* adapter + .sendTurn({ threadId, input: "take this instead", attachments: [] }) + .pipe(Effect.timeout("3 seconds")); + yield* Deferred.await(turnCompleted).pipe(Effect.timeout("3 seconds")); + yield* Fiber.join(firstSendTurnFiber).pipe(Effect.timeout("3 seconds")); + + const requestLog = yield* Effect.promise(() => readJsonLines(requestLogPath)); + const methods = requestLog.flatMap((entry) => + typeof entry.method === "string" ? [entry.method] : [], + ); + const turnStartedEvents = runtimeEvents.filter( + (event) => event.type === "turn.started" && String(event.threadId) === String(threadId), + ); + const turnCompletedEvents = runtimeEvents.filter( + (event): event is Extract => + event.type === "turn.completed" && String(event.threadId) === String(threadId), + ); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + + assert.equal(String(steered.turnId), String(firstTurnId)); + assert.isTrue(methods.includes("session/cancel")); + assert.isAtLeast(methods.filter((method) => method === "session/prompt").length, 2); + assert.lengthOf(turnStartedEvents, 1); + assert.lengthOf(turnCompletedEvents, 1); + assert.equal(turnCompletedEvents[0]?.payload.state, "completed"); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }).pipe(TestClock.withLive), + ); + + it.effect("keeps a steered turn completed when the cancelled prompt settles first", () => + Effect.gen(function* () { + const threadId = ThreadId.make("grok-steer-cancelled-prompt-settles-first"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "grok-acp-steer-old-first-")), + ); + const requestLogPath = NodePath.join(tempDir, "requests.ndjson"); + const wrapperPath = yield* Effect.promise(() => + makeMockGrokWrapper({ + T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1", + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const firstTurnStarted = yield* Deferred.make(); + const turnCompleted = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.gen(function* () { + runtimeEvents.push(event); + if (String(event.threadId) !== String(threadId)) { + return; + } + if (event.type === "turn.started" && event.turnId !== undefined) { + yield* Deferred.succeed(firstTurnStarted, event.turnId).pipe(Effect.ignore); + return; + } + if (event.type === "turn.completed") { + yield* Deferred.succeed(turnCompleted, undefined).pipe(Effect.ignore); + } + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("grok"), + cwd: process.cwd(), + runtimeMode: "full-access", + }); + + const firstSendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "hang until steered", attachments: [] }) + .pipe(Effect.forkChild); + yield* Deferred.await(firstTurnStarted).pipe(Effect.timeout("2 seconds")); + yield* waitForFileContent(requestLogPath, 80, '"method":"session/prompt"'); + + const steered = yield* adapter + .sendTurn({ threadId, input: "take this instead", attachments: [] }) + .pipe(Effect.forkChild); + yield* Fiber.join(firstSendTurnFiber).pipe(Effect.timeout("3 seconds")); + const steeredResult = yield* Fiber.join(steered).pipe(Effect.timeout("3 seconds")); + yield* Deferred.await(turnCompleted).pipe(Effect.timeout("3 seconds")); + + const turnCompletedEvents = runtimeEvents.filter( + (event): event is Extract => + event.type === "turn.completed" && String(event.threadId) === String(threadId), + ); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + + assert.lengthOf(turnCompletedEvents, 1); + assert.equal(String(steeredResult.turnId), String(turnCompletedEvents[0]?.turnId)); + assert.equal(turnCompletedEvents[0]?.payload.state, "completed"); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }).pipe(TestClock.withLive), + ); + + it.effect("keeps the original prompt running when a steer fails during preparation", () => + Effect.gen(function* () { + const threadId = ThreadId.make("grok-failed-steer-keeps-original-prompt"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "grok-acp-failed-steer-")), + ); + const requestLogPath = NodePath.join(tempDir, "requests.ndjson"); + const wrapperPath = yield* Effect.promise(() => + makeMockGrokWrapper({ + T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1", + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const firstTurnStarted = yield* Deferred.make(); + const turnCompleted = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.gen(function* () { + runtimeEvents.push(event); + if (String(event.threadId) !== String(threadId)) { + return; + } + if (event.type === "turn.started" && event.turnId !== undefined) { + yield* Deferred.succeed(firstTurnStarted, event.turnId).pipe(Effect.ignore); + return; + } + if (event.type === "turn.completed") { + yield* Deferred.succeed(turnCompleted, undefined).pipe(Effect.ignore); + } + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("grok"), + cwd: process.cwd(), + runtimeMode: "full-access", + }); + + const firstSendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "hang until a failed steer", attachments: [] }) + .pipe(Effect.forkChild); + const firstTurnId = yield* Deferred.await(firstTurnStarted).pipe(Effect.timeout("2 seconds")); + + const steerError = yield* Effect.flip( + adapter.sendTurn({ + threadId, + input: " ", + attachments: [], + }), + ); + yield* waitForFileContent(requestLogPath, 80, '"method":"session/prompt"'); + + const sessionsAfterFailedSteer = yield* adapter.listSessions(); + const sessionAfterFailedSteer = sessionsAfterFailedSteer.find( + (session) => session.threadId === threadId, + ); + const completedBeforeInterrupt = runtimeEvents.filter( + (event): event is Extract => + event.type === "turn.completed" && String(event.threadId) === String(threadId), + ); + + yield* adapter.interruptTurn(threadId, firstTurnId).pipe(Effect.timeout("2 seconds")); + yield* Fiber.join(firstSendTurnFiber).pipe(Effect.timeout("3 seconds")); + yield* Deferred.await(turnCompleted).pipe(Effect.timeout("3 seconds")); + + const turnCompletedEvents = runtimeEvents.filter( + (event): event is Extract => + event.type === "turn.completed" && String(event.threadId) === String(threadId), + ); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + + assert.equal(steerError._tag, "ProviderAdapterValidationError"); + assert.equal(sessionAfterFailedSteer?.status, "running"); + assert.equal(String(sessionAfterFailedSteer?.activeTurnId), String(firstTurnId)); + assert.lengthOf(completedBeforeInterrupt, 0); + assert.lengthOf(turnCompletedEvents, 1); + assert.equal(String(turnCompletedEvents[0]?.turnId), String(firstTurnId)); + assert.equal(turnCompletedEvents[0]?.payload.state, "cancelled"); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }).pipe(TestClock.withLive), + ); }); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 1747e0f12800..20c2c3388987 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -116,8 +116,18 @@ interface GrokSessionContext { interruptedTurnIds: Set; /** Number of sendTurn prompts currently in flight or being prepared. * >0 means a turn is actively running, so a new sendTurn is a steer that - * continues it, and only the last remaining prompt settles the turn. */ + * cancels the in-flight prompt and continues the same turn. The current + * epoch owns the terminal state; a superseded prompt may only flush that + * stored result when it is last to drain. */ promptsInFlight: number; + /** Monotonic id assigned to each sendTurn. Steers discard older epochs. */ + promptEpoch: number; + /** Prompt epochs below this value must not start an ACP session/prompt. */ + discardBeforeEpoch: number; + /** Current-epoch terminal result, emitted when the last in-flight prompt drains. */ + pendingTurnCompletion: GrokTurnTerminal | undefined; + /** Serializes cancel-then-prompt so a steer cannot miss or hit the wrong RPC. */ + readonly promptLifecycle: Semaphore.Semaphore; currentModelId: string | undefined; stopped: boolean; } @@ -226,6 +236,40 @@ export function grokPromptSettlementBelongsToContext(input: { ); } +export type GrokTurnTerminal = { + readonly errorMessage?: string; + readonly completedStopReason?: EffectAcpSchema.StopReason | null; +}; + +/** + * Choose the merged turn's terminal result from one prompt settlement. + * The current epoch (`promptEpoch >= discardBeforeEpoch`) owns the outcome. + * A superseded prompt never overwrites that result; if it drains last it + * only flushes the stored current-epoch result. + */ +export function grokTurnCompletionForPromptEpoch(input: { + readonly promptEpoch: number; + readonly discardBeforeEpoch: number; + readonly remainingPrompts: number; + readonly stored: GrokTurnTerminal | undefined; + readonly incoming: GrokTurnTerminal | undefined; + readonly emitTurnCompletion: boolean; +}): { + readonly stored: GrokTurnTerminal | undefined; + readonly emit: GrokTurnTerminal | undefined; +} { + const superseded = input.promptEpoch < input.discardBeforeEpoch; + const stored = + !superseded && input.emitTurnCompletion && input.incoming !== undefined + ? input.incoming + : input.stored; + // The final drain may belong to a superseded prompt whose own settlement is + // suppressed. It must still flush a terminal result stored by the current + // epoch, or the merged turn remains running forever. + const emit = input.remainingPrompts === 0 ? stored : undefined; + return { stored, emit }; +} + export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapterLiveOptions) { return Effect.gen(function* () { const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("grok"); @@ -306,6 +350,8 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte readonly emitTurnCompletion?: boolean; /** Interrupt/cancel: drop every outstanding prompt slot and settle once. */ readonly settleAllPrompts?: boolean; + /** sendTurn epoch that produced this settlement. */ + readonly promptEpoch?: number; }, ) => Effect.gen(function* () { @@ -313,6 +359,8 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte if (!liveCtx) { return; } + const promptEpoch = options?.promptEpoch; + const superseded = promptEpoch !== undefined && promptEpoch < liveCtx.discardBeforeEpoch; const settlementBelongsToLiveContext = grokPromptSettlementBelongsToContext({ liveAcpSessionId: liveCtx.acpSessionId, expectedAcpSessionId, @@ -324,7 +372,10 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte // interruptTurn already consumed every prompt slot for this turn. A // late prompt result must neither emit a second terminal event nor // consume a slot belonging to a newer turn on the same ACP session. + // A superseded steer prompt also never publishes its own cancellation + // as the merged turn's outcome. if ( + superseded || liveCtx.acpSessionId !== expectedAcpSessionId || liveCtx.interruptedTurnIds.has(turnId) ) { @@ -362,6 +413,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte let settleTurnId = turnId; if (options?.settleAllPrompts) { liveCtx.promptsInFlight = 0; + liveCtx.pendingTurnCompletion = undefined; if (liveCtx.activeTurnId !== turnId && liveCtx.session.activeTurnId !== turnId) { const fallbackTurnId = liveCtx.activeTurnId ?? liveCtx.session.activeTurnId; if (!fallbackTurnId) { @@ -381,15 +433,72 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte } } else { const remainingPrompts = Math.max(0, liveCtx.promptsInFlight - 1); + liveCtx.promptsInFlight = remainingPrompts; + const incoming: GrokTurnTerminal | undefined = + options?.errorMessage !== undefined + ? { errorMessage: options.errorMessage } + : options?.completedStopReason !== undefined + ? { completedStopReason: options.completedStopReason } + : undefined; + const decision = grokTurnCompletionForPromptEpoch({ + promptEpoch: promptEpoch ?? liveCtx.promptEpoch, + discardBeforeEpoch: liveCtx.discardBeforeEpoch, + remainingPrompts, + stored: liveCtx.pendingTurnCompletion, + incoming, + emitTurnCompletion: options?.emitTurnCompletion !== false, + }); + liveCtx.pendingTurnCompletion = decision.stored; if ( remainingPrompts > 0 || liveCtx.activeTurnId !== settleTurnId || liveCtx.session.activeTurnId !== settleTurnId ) { - liveCtx.promptsInFlight = remainingPrompts; return; } - liveCtx.promptsInFlight = remainingPrompts; + const updatedAt = yield* nowIso; + const canEmitTurnCompletion = + liveCtx.session.status === "running" || liveCtx.session.status === "connecting"; + const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session; + liveCtx.activeTurnId = undefined; + liveCtx.pendingTurnCompletion = undefined; + liveCtx.session = { + ...readySession, + status: "ready", + updatedAt, + }; + if (!canEmitTurnCompletion || decision.emit === undefined) { + return; + } + if (decision.emit.errorMessage !== undefined) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId: settleTurnId, + payload: { + state: "failed", + errorMessage: decision.emit.errorMessage, + }, + }); + return; + } + if (decision.emit.completedStopReason !== undefined) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId: settleTurnId, + payload: { + state: + decision.emit.completedStopReason === "cancelled" ? "cancelled" : "completed", + stopReason: decision.emit.completedStopReason, + }, + }); + } + return; } const updatedAt = yield* nowIso; const canEmitTurnCompletion = @@ -784,6 +893,10 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte activeTurnId: undefined, interruptedTurnIds: new Set(), promptsInFlight: 0, + promptEpoch: 0, + discardBeforeEpoch: 0, + pendingTurnCompletion: undefined, + promptLifecycle: yield* Semaphore.make(1), currentModelId: boundModelId, stopped: false, }; @@ -928,15 +1041,18 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte input.threadId, Effect.gen(function* () { const ctx = yield* requireSession(input.threadId); - // A sendTurn while a prompt is in flight is a steer: the agent - // folds the new prompt into the ongoing work, so the active turn - // id is reused instead of opening a new turn. + // A sendTurn while a prompt is in flight is a steer: reuse the + // active turn and cancel the in-flight ACP prompt so Grok takes + // the new instruction immediately, matching Claude/Codex, instead + // of waiting behind serialized session/prompt. const steeringTurnId = ctx.promptsInFlight > 0 ? ctx.activeTurnId : undefined; const turnId = steeringTurnId ?? TurnId.make(yield* randomUUIDv4); // Count this prompt immediately so a superseded in-flight prompt // resolving from here on does not settle the turn; decremented on // preparation failure here, and after the prompt below otherwise. ctx.promptsInFlight += 1; + ctx.promptEpoch += 1; + const promptEpoch = ctx.promptEpoch; // Bind the turn id before cooperative yields so interruptTurn can // settle this prompt even if stop arrives during preparation. ctx.activeTurnId = turnId; @@ -1014,14 +1130,12 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte const displayModel = currentModelId ? resolveGrokAcpBaseModelId(currentModelId) : undefined; - for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { - yield* Effect.yieldNow; - } if (ctx.interruptedTurnIds.has(turnId)) { yield* settlePromptInFlight(input.threadId, turnId, ctx.acpSessionId, { completedStopReason: "cancelled", emitTurnCompletion: false, settleAllPrompts: true, + promptEpoch, }); return yield* new ProviderAdapterRequestError({ provider: PROVIDER, @@ -1049,6 +1163,14 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte turnId, payload: displayModel ? { model: displayModel } : {}, }); + } else { + // Discard the previous epoch only after this replacement is + // ready. A failed steer must not skip the live prompt, which + // settles without a terminal event when emitTurnCompletion is + // false. + yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals); + yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs); + ctx.discardBeforeEpoch = promptEpoch; } return { @@ -1057,6 +1179,9 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte displayModel, promptParts, turnId, + promptEpoch, + promptLifecycle: ctx.promptLifecycle, + steeringTurnId, }; }).pipe( Effect.tapCause(() => @@ -1068,6 +1193,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte yield* settlePromptInFlight(input.threadId, turnId, liveCtx.acpSessionId, { errorMessage: "Grok prompt preparation failed.", emitTurnCompletion: false, + promptEpoch, }); }), ), @@ -1083,27 +1209,91 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte const promptFailureMessageRef = yield* Ref.make(undefined); return yield* Effect.gen(function* () { - const result = yield* prepared.acp - .prompt({ - prompt: prepared.promptParts, - }) - .pipe( - Effect.tap((promptResult) => - Effect.all([ - Ref.set(promptRpcSucceeded, true), - Ref.set(promptResultRef, promptResult), - ]), - ), - Effect.tapError((error) => - Ref.set( - promptFailureMessageRef, - mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error).message, - ).pipe(Effect.andThen(prepared.acp.drainEvents)), - ), - Effect.mapError((error) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), + const promptStart = yield* prepared.promptLifecycle.withPermit( + Effect.gen(function* () { + const liveCtx = sessions.get(input.threadId); + const interrupted = liveCtx?.interruptedTurnIds.has(prepared.turnId) === true; + if ( + !liveCtx || + liveCtx.acpSessionId !== prepared.acpSessionId || + prepared.promptEpoch < liveCtx.discardBeforeEpoch || + interrupted + ) { + return { _tag: "Skipped" as const, interrupted }; + } + if (prepared.steeringTurnId !== undefined) { + yield* Effect.ignore( + liveCtx.acp.cancel.pipe( + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/cancel", error), + ), + ), + ); + } + if (liveCtx.interruptedTurnIds.has(prepared.turnId)) { + return { _tag: "Skipped" as const, interrupted: true }; + } + const dispatched = yield* Deferred.make(); + const fiber = yield* liveCtx.acp + .prompt({ prompt: prepared.promptParts }, { dispatched }) + .pipe(Effect.forkChild); + // Hold the lifecycle permit until the runtime has registered this + // prompt's RPC fiber, so a later steer's session/cancel targets + // this prompt. Fall through if the prompt fails before that point. + yield* Effect.raceFirst( + Deferred.await(dispatched), + Fiber.await(fiber).pipe(Effect.asVoid), + ); + return { _tag: "Started" as const, fiber }; + }), + ); + if (promptStart._tag === "Skipped") { + // Settle after releasing promptLifecycle. Holding both locks + // deadlocks the next sendTurn, which takes the thread lock first. + yield* withThreadLock( + input.threadId, + settlePromptInFlight( + input.threadId, + prepared.turnId, + prepared.acpSessionId, + promptStart.interrupted + ? { + completedStopReason: "cancelled", + settleAllPrompts: true, + promptEpoch: prepared.promptEpoch, + } + : { + emitTurnCompletion: false, + promptEpoch: prepared.promptEpoch, + }, ), ); + yield* Ref.set(promptSettled, true); + const liveCtx = sessions.get(input.threadId); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: liveCtx?.session.resumeCursor, + }; + } + + const result = yield* Fiber.join(promptStart.fiber).pipe( + Effect.tap((promptResult) => + Effect.all([ + Ref.set(promptRpcSucceeded, true), + Ref.set(promptResultRef, promptResult), + ]), + ), + Effect.tapError((error) => + Ref.set( + promptFailureMessageRef, + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error).message, + ).pipe(Effect.andThen(prepared.acp.drainEvents)), + ), + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), + ), + ); return yield* withThreadLock( input.threadId, @@ -1117,6 +1307,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte { errorMessage: "Grok session changed before the turn completed.", settleAllPrompts: true, + promptEpoch: prepared.promptEpoch, }, ); yield* Ref.set(promptSettled, true); @@ -1163,51 +1354,12 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte updatedAt: yield* nowIso, ...(prepared.displayModel ? { model: prepared.displayModel } : {}), }; - const remainingPrompts = Math.max(0, ctx.promptsInFlight - 1); - ctx.promptsInFlight = remainingPrompts; - - // Only the last remaining prompt settles the turn. A steer- - // superseded prompt resolving while another is in flight or - // pending must leave the merged turn running. - if ( - remainingPrompts === 0 && - ctx.activeTurnId === prepared.turnId && - ctx.session.activeTurnId === prepared.turnId - ) { - if (ctx.interruptedTurnIds.has(prepared.turnId)) { - yield* Ref.set(promptSettled, true); - return { - threadId: input.threadId, - turnId: prepared.turnId, - resumeCursor: ctx.session.resumeCursor, - }; - } - const completedAt = yield* nowIso; - const { activeTurnId: _completedTurnId, ...readySession } = ctx.session; - ctx.activeTurnId = undefined; - ctx.session = { - ...readySession, - status: "ready", - updatedAt: completedAt, - ...(prepared.displayModel ? { model: prepared.displayModel } : {}), - }; - const completedStopReason = completedStopReasonFromPromptResponse(result); - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId: prepared.turnId, - payload: { - state: result.stopReason === "cancelled" ? "cancelled" : "completed", - stopReason: completedStopReason, - }, - }); - ctx.interruptedTurnIds.delete(prepared.turnId); - yield* Ref.set(promptSettled, true); - } else if (remainingPrompts > 0) { - yield* Ref.set(promptSettled, true); - } + yield* settlePromptInFlight(input.threadId, prepared.turnId, prepared.acpSessionId, { + completedStopReason: completedStopReasonFromPromptResponse(result), + promptEpoch: prepared.promptEpoch, + }); + ctx.interruptedTurnIds.delete(prepared.turnId); + yield* Ref.set(promptSettled, true); return { threadId: input.threadId, @@ -1240,6 +1392,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte { errorMessage: "Grok session changed before the turn completed.", settleAllPrompts: true, + promptEpoch: prepared.promptEpoch, }, ); return; @@ -1266,6 +1419,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte prepared.acpSessionId, { completedStopReason: completedStopReasonFromPromptResponse(promptResult), + promptEpoch: prepared.promptEpoch, }, ); }), @@ -1278,6 +1432,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte input.threadId, settlePromptInFlight(input.threadId, prepared.turnId, prepared.acpSessionId, { errorMessage: errorMessage ?? "Grok prompt request failed.", + promptEpoch: prepared.promptEpoch, }), ); }).pipe(Effect.catch(() => Effect.void)), diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index 7052182de389..bea27024a50d 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -204,11 +204,14 @@ export class AcpSessionRuntime extends Context.Service< /** Latest configuration options observed from session setup and configuration writes. */ readonly getConfigOptions: Effect.Effect>; /** - * Sends a prompt turn to the active session. + * Sends a prompt turn to the active session. `options.dispatched` settles once the + * `session/prompt` RPC is registered as the active prompt, so a caller that forks this + * effect knows when a later `cancel` will target this prompt. * @see https://agentclientprotocol.com/protocol/schema#session/prompt */ readonly prompt: ( payload: Omit, + options?: { readonly dispatched?: Deferred.Deferred }, ) => Effect.Effect; /** * Sends a real ACP `session/cancel` notification for the active session. @@ -734,7 +737,7 @@ export const make = ( }), getModeState: Ref.get(modeStateRef), getConfigOptions: Ref.get(configOptionsRef), - prompt: (payload) => + prompt: (payload, promptOptions?) => promptSerializationSemaphore.withPermit( Effect.gen(function* () { const started = yield* getStartedState; @@ -755,6 +758,9 @@ export const make = ( acp.agent.prompt(requestPayload), ).pipe(Effect.forkIn(runtimeScope)); yield* Ref.set(activePromptFiberRef, Option.some(promptRpcFiber)); + if (promptOptions?.dispatched) { + yield* Deferred.succeed(promptOptions.dispatched, undefined); + } return yield* Fiber.join(promptRpcFiber).pipe( Effect.catchCause((cause) => Cause.hasInterruptsOnly(cause) @@ -783,9 +789,9 @@ export const make = ( if (Option.isSome(activePromptFiber)) { yield* Fiber.interrupt(activePromptFiber.value).pipe(Effect.ignore); } - yield* acp.agent - .cancel({ sessionId: started.sessionId }) - .pipe(Effect.ignore, Effect.forkIn(runtimeScope)); + // Await the notification write so a replacement session/prompt + // cannot race ahead of session/cancel on the wire. + yield* acp.agent.cancel({ sessionId: started.sessionId }).pipe(Effect.ignore); }), ), ), diff --git a/apps/server/src/provider/acp/XAiAcpExtension.ts b/apps/server/src/provider/acp/XAiAcpExtension.ts index d36a5fcfc895..fd3121938190 100644 --- a/apps/server/src/provider/acp/XAiAcpExtension.ts +++ b/apps/server/src/provider/acp/XAiAcpExtension.ts @@ -228,11 +228,11 @@ export const makeXAiPromptCompletionRuntime = Effect.fn("makeXAiPromptCompletion runtime .start() .pipe(Effect.tap((started) => Ref.set(activeSessionIdRef, started.sessionId))), - prompt: (payload) => + prompt: (payload, promptOptions?) => Effect.gen(function* () { const sessionId = yield* Ref.get(activeSessionIdRef); if (sessionId === undefined) { - return yield* runtime.prompt(payload); + return yield* runtime.prompt(payload, promptOptions); } const promptId = yield* allocatePromptFallbackId; @@ -251,7 +251,7 @@ export const makeXAiPromptCompletionRuntime = Effect.fn("makeXAiPromptCompletion } satisfies Omit; return yield* Effect.raceFirst( - runtime.prompt(requestPayload), + runtime.prompt(requestPayload, promptOptions), Deferred.await(fallback.deferred), ).pipe( Effect.tap((response) => diff --git a/docs/internals/providers.md b/docs/internals/providers.md index fef9eda26686..e843d72aecc9 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -95,6 +95,13 @@ in-session model change reaches ACP `session/set_model`. Cursor and OpenCode still start sessions through `AcpSessionRuntime.start()`. The new `initialize()` method is additive and unused by those adapters. +ACP outbound notifications (`session/cancel` included) encode as JSON-RPC with no `id` or +`headers`. The previous Request encoder emitted `id: ""`, which Grok CLI treats as a malformed +request and drops, so Stop did not stop. Cursor and OpenCode share this protocol path; the mock +agent was previously lenient and hid the bug. `AcpSessionRuntime.cancel` now waits for the cancel +write before returning so a replacement prompt cannot race ahead of it. Grok mid-turn sends cancel +the in-flight prompt and continue the same turn instead of queueing. + ## Raw protocol observation The [ACP protocol](../../packages/effect-acp/src/protocol.ts) and diff --git a/packages/effect-acp/src/protocol.test.ts b/packages/effect-acp/src/protocol.test.ts index 68ca73a07d42..de535fc5052f 100644 --- a/packages/effect-acp/src/protocol.test.ts +++ b/packages/effect-acp/src/protocol.test.ts @@ -340,25 +340,24 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => { yield* transport.notify("session/cancel", { sessionId: "session-1" }); + // A notification must not carry `id` or `headers`. Grok CLI drops frames that do. assert.deepEqual(events, [ { direction: "outgoing", stage: "decoded", payload: { - _tag: "Request", - id: "", + _tag: "Notification", tag: "session/cancel", payload: { sessionId: "session-1", }, - headers: [], }, }, { direction: "outgoing", stage: "raw", payload: - '{"jsonrpc":"2.0","method":"session/cancel","params":{"sessionId":"session-1"},"id":"","headers":[]}\n', + '{"jsonrpc":"2.0","method":"session/cancel","params":{"sessionId":"session-1"}}\n', }, ]); }), @@ -404,11 +403,13 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => { serverRequestMethods: new Set(), }); + // Notifications encode through Schema, so the cause is the schema failure rather + // than the raw TypeError JSON.stringify throws. The ACP error shape is what callers see. const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip); assert.instanceOf(bigintError, AcpError.AcpProtocolParseError); assert.equal(bigintError.operation, "encode-message"); assert.equal(bigintError.method, "x/test"); - assert.instanceOf(bigintError.cause, TypeError); + assert.isDefined(bigintError.cause); assert.equal( bigintError.message, "ACP protocol operation 'encode-message' failed for method 'x/test'.", @@ -420,7 +421,7 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => { assert.instanceOf(circularError, AcpError.AcpProtocolParseError); assert.equal(circularError.operation, "encode-message"); assert.equal(circularError.method, "x/test"); - assert.instanceOf(circularError.cause, TypeError); + assert.isDefined(circularError.cause); const requestError = yield* transport.request("x/request", 1n).pipe( Effect.match({ diff --git a/packages/effect-acp/src/protocol.ts b/packages/effect-acp/src/protocol.ts index 71a6e1238ab6..897e1d3829eb 100644 --- a/packages/effect-acp/src/protocol.ts +++ b/packages/effect-acp/src/protocol.ts @@ -1,6 +1,7 @@ import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; +import * as Exit from "effect/Exit"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -87,6 +88,16 @@ const decodeElicitationComplete = Schema.decodeUnknownEffect( AcpSchema.ElicitationCompleteNotification, ); const parserFactory = RpcSerialization.ndJsonRpc(); +// Outbound JSON-RPC notification: no `id`, so peers never treat it as a request. +const encodeJsonRpcNotification = Schema.encodeUnknownExit( + Schema.fromJsonString( + Schema.Struct({ + jsonrpc: Schema.Literal("2.0"), + method: Schema.String, + params: Schema.Unknown, + }), + ), +); const makeRawQueue = Effect.fn("makeRawQueue")(function* (bufferSize: number | "unbounded" = 0) { if (bufferSize === 0) { @@ -136,6 +147,11 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi const offerOutgoing = Effect.fn("offerOutgoing")(function* ( message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded, ) { + // RpcClient emits `@effect/rpc/Interrupt` when a pending request's fiber is interrupted. + // ACP has no such method; agents log it as an error and cannot act on it, so drop it. + if (message._tag === "Interrupt") { + return; + } yield* logProtocol({ direction: "outgoing", stage: "decoded", @@ -548,17 +564,29 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi supportsSpanPropagation: true, }); + // JSON-RPC notifications carry no `id`. The generic Request encoder emits `id: ""` plus + // `headers`, which real agents (Grok CLI) parse as a malformed request and silently drop. + // That made `session/cancel` a no-op against Grok while the lenient mock agent accepted it. const sendNotification = Effect.fn("sendNotification")(function* ( method: string, payload: unknown, ) { - yield* offerOutgoing({ - _tag: "Request", - id: "", - tag: method, - payload, - headers: [], + yield* logProtocol({ + direction: "outgoing", + stage: "decoded", + payload: { _tag: "Notification", tag: method, payload }, }); + const exit = encodeJsonRpcNotification({ jsonrpc: "2.0", method, params: payload }); + if (Exit.isFailure(exit)) { + return yield* AcpError.AcpProtocolParseError.fromEncodingError( + method, + undefined, + Cause.squash(exit.cause), + ); + } + const encoded = `${exit.value}\n`; + yield* logProtocol({ direction: "outgoing", stage: "raw", payload: encoded }); + yield* Queue.offer(outgoing, encoded); }); const sendRequest = Effect.fn("sendRequest")(function* (method: string, payload: unknown) {