diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index bc7828dd854..355f731599b 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -36,10 +36,12 @@ const emitStaleXAiPromptCompleteBeforeSecondHang = const emitOverlappingXAiPromptCompleteOutOfOrder = process.env.T3_ACP_EMIT_OVERLAPPING_XAI_PROMPT_COMPLETE_OUT_OF_ORDER === "1"; const failPrompt = process.env.T3_ACP_FAIL_PROMPT === "1"; +const failFirstPrompt = process.env.T3_ACP_FAIL_FIRST_PROMPT === "1"; const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1"; const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1"; const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT; const promptDelayMs = Number(process.env.T3_ACP_PROMPT_DELAY_MS ?? "0"); +const setConfigOptionDelayMs = Number(process.env.T3_ACP_SET_CONFIG_OPTION_DELAY_MS ?? "0"); const permissionOptionIds = { allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once", allowAlways: process.env.T3_ACP_ALLOW_ALWAYS_OPTION_ID ?? "allow-always", @@ -398,6 +400,12 @@ const program = Effect.gen(function* () { yield* agent.handleSetSessionConfigOption((request) => Effect.gen(function* () { + // Cursor selects its model through set_config_option, so this widens the + // sendTurn preparation window: tests can land a concurrent sendTurn + // between the in-flight counter increment and the turn.started emission. + if (Number.isFinite(setConfigOptionDelayMs) && setConfigOptionDelayMs > 0) { + yield* Effect.sleep(`${setConfigOptionDelayMs} millis`); + } if (exitOnSetConfigOption) { return yield* Effect.sync(() => { process.exit(7); @@ -461,7 +469,7 @@ const program = Effect.gen(function* () { yield* Effect.sleep(`${promptDelayMs} millis`); } - if (failPrompt) { + if (failPrompt || (failFirstPrompt && promptCount === 1)) { return yield* AcpError.AcpRequestError.internalError("Mock prompt failure"); } diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 17aeff2d0e3..f45129532c4 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -35,7 +35,11 @@ import * as TestClock from "effect/testing/TestClock"; import { attachmentRelativePath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; -import { ProviderAdapterProcessError, ProviderAdapterValidationError } from "../Errors.ts"; +import { + ProviderAdapterProcessError, + ProviderAdapterRequestError, + ProviderAdapterValidationError, +} from "../Errors.ts"; import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts"; const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); @@ -754,6 +758,94 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("does not wedge the turn when an attachment has an unsupported mime type", () => { + const baseDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "claude-attachments-")); + const harness = makeHarness({ + cwd: "/tmp/project-claude-unsupported-attachment", + baseDir, + }); + return Effect.gen(function* () { + yield* Effect.addFinalizer(() => + Effect.sync(() => + NodeFS.rmSync(baseDir, { + recursive: true, + force: true, + }), + ), + ); + + const adapter = yield* ClaudeAdapter; + const { attachmentsDir } = yield* ServerConfig; + + const attachment = { + type: "image" as const, + id: "thread-claude-attachment-87654321-4321-4321-4321-cba987654321", + name: "photo.heic", + mimeType: "image/heic", + sizeBytes: 4, + }; + const attachmentPath = NodePath.join(attachmentsDir, attachmentRelativePath(attachment)); + NodeFS.mkdirSync(NodePath.dirname(attachmentPath), { recursive: true }); + NodeFS.writeFileSync(attachmentPath, Uint8Array.from([1, 2, 3, 4])); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + // Collect exactly the 3 session-startup events plus the single + // turn.started event expected from the *second* (valid) sendTurn + // below. If the unsupported-attachment turn also emitted + // turn.started, this fiber would collect that extra event instead + // and the later assertions on event identity/count would fail. + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 4).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const failure = yield* adapter + .sendTurn({ + threadId: session.threadId, + input: "What's in this image?", + attachments: [attachment], + }) + .pipe(Effect.flip); + + assert.instanceOf(failure, ProviderAdapterRequestError); + assert.match(failure.detail, /Unsupported Claude image attachment type 'image\/heic'/u); + + // The failed validation must leave the session exactly as it was + // before sendTurn was called: still "ready", no active turn. + const sessionsAfterFailure = yield* adapter.listSessions(); + const sessionAfterFailure = sessionsAfterFailure.find( + (candidate) => candidate.threadId === session.threadId, + ); + assert.isDefined(sessionAfterFailure); + assert.equal(sessionAfterFailure?.status, "ready"); + assert.isUndefined(sessionAfterFailure?.activeTurnId); + + // A subsequent, valid turn must still work normally and be the + // only source of a turn.started event. + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "Never mind, just say hi.", + attachments: [], + }); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const turnStartedEvents = runtimeEvents.filter((event) => event.type === "turn.started"); + assert.equal(turnStartedEvents.length, 1); + const turnStarted = turnStartedEvents[0]; + if (turnStarted?.type === "turn.started") { + assert.equal(String(turnStarted.turnId), String(turn.turnId)); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("maps Claude stream/runtime messages to canonical provider runtime events", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index f6e63eeffad..de38ac1ca67 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -3693,6 +3693,19 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } const turnId = steeringTurnState?.turnId ?? TurnId.make(yield* randomUUIDv4); + + // Validate/construct the user message BEFORE any turn-state mutation or + // event emission below. buildUserMessageEffect can fail (e.g. an + // unsupported attachment mime type) and must not leave the session + // marked "running" with a turn.started already emitted — that combo + // races the async recovery path and can wedge the turn forever. See + // ProviderCommandReactor's recoverTurnStartFailure. + const message = yield* buildUserMessageEffect(input, { + fileSystem, + attachmentsDir: serverConfig.attachmentsDir, + boundInstanceId, + }); + if (steeringTurnState === null) { const turnState: ClaudeTurnState = { turnId, @@ -3726,12 +3739,6 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); } - const message = yield* buildUserMessageEffect(input, { - fileSystem, - attachmentsDir: serverConfig.attachmentsDir, - boundInstanceId, - }); - yield* Queue.offer(context.promptQueue, { type: "message", message, diff --git a/apps/server/src/provider/Layers/CursorAdapter.test.ts b/apps/server/src/provider/Layers/CursorAdapter.test.ts index 491f718a977..2619a168c6f 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.test.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.test.ts @@ -27,6 +27,7 @@ import { import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; +import { ProviderAdapterRequestError } from "../Errors.ts"; import type { CursorAdapterShape } from "../Services/CursorAdapter.ts"; import { makeCursorAdapter } from "./CursorAdapter.ts"; const decodeCursorSettings = Schema.decodeSync(CursorSettings); @@ -250,6 +251,70 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => { }), ); + it.effect("does not wedge the turn when an attachment fails to resolve", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-invalid-attachment-thread"); + + const wrapperPath = yield* Effect.promise(() => makeMockAgentWrapper()); + yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } }); + + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 4).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("cursor"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }); + + const failure = yield* adapter + .sendTurn({ + threadId: session.threadId, + input: "What's in this image?", + attachments: [ + { + type: "image", + id: "../escape-attachments-dir", + name: "photo.png", + mimeType: "image/png", + sizeBytes: 4, + }, + ], + }) + .pipe(Effect.flip); + + assert.instanceOf(failure, ProviderAdapterRequestError); + assert.match(failure.detail, /Invalid attachment id/u); + + const sessionsAfterFailure = yield* adapter.listSessions(); + const sessionAfterFailure = sessionsAfterFailure.find( + (candidate) => candidate.threadId === threadId, + ); + assert.isDefined(sessionAfterFailure); + assert.isUndefined(sessionAfterFailure?.activeTurnId); + + // A subsequent, valid turn must still work and be the only source + // of a turn.started event. + yield* adapter.sendTurn({ + threadId, + input: "hello mock", + attachments: [], + }); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const turnStartedEvents = runtimeEvents.filter((event) => event.type === "turn.started"); + assert.equal(turnStartedEvents.length, 1); + + yield* adapter.stopSession(threadId); + }), + ); + it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => Effect.gen(function* () { const adapter = yield* CursorAdapter; @@ -326,6 +391,306 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => { }), ); + it.effect("steers into the reserved turn when sendTurn overlaps preparation", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-steer-during-prep-thread"); + + // A model change forces a real session/set_config_option round-trip + // (same-value selections are skipped runtime-side), and the mock delays + // that request — holding the first prompt inside its preparation window + // (after the in-flight increment, before turn.started) long enough for + // the second sendTurn to land inside it. + const wrapperPath = yield* Effect.promise(() => + makeMockAgentWrapper({ T3_ACP_SET_CONFIG_OPTION_DELAY_MS: "1000" }), + ); + yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } }); + + const runtimeEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("cursor"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }); + + const firstTurnFiber = yield* adapter + .sendTurn({ + threadId, + input: "first prompt", + attachments: [], + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "composer-2" }, + }) + .pipe(Effect.forkChild); + // Forked with no intervening await: this sendTurn must observe the + // first one's reservation, not mint a second turn id. + const secondTurnFiber = yield* adapter + .sendTurn({ + threadId, + input: "second prompt", + attachments: [], + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }) + .pipe(Effect.forkChild); + + const firstTurn = yield* Fiber.join(firstTurnFiber); + const secondTurn = yield* Fiber.join(secondTurnFiber); + assert.equal(String(secondTurn.turnId), String(firstTurn.turnId)); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const turnStartedEvents = runtimeEvents.filter((event) => event.type === "turn.started"); + const turnCompletedEvents = runtimeEvents.filter((event) => event.type === "turn.completed"); + + // One turn boundary for the merged run — a second turn.started would + // open a turn that can never settle. + assert.equal(turnStartedEvents.length, 1); + assert.equal(String(turnStartedEvents[0]?.turnId), String(firstTurn.turnId)); + assert.equal(turnCompletedEvents.length, 1); + assert.equal(String(turnCompletedEvents[0]?.turnId), String(firstTurn.turnId)); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("starts and settles the steered turn when the reserving prompt fails preparation", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-reserver-fails-prep-thread"); + + const wrapperPath = yield* Effect.promise(() => + makeMockAgentWrapper({ T3_ACP_SET_CONFIG_OPTION_DELAY_MS: "1000" }), + ); + yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } }); + + const runtimeEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("cursor"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }); + + // The reserving prompt survives the delayed model change, then dies in + // attachment validation — after a steer has already joined its turn. + const failingTurnFiber = yield* adapter + .sendTurn({ + threadId, + input: "What's in this image?", + attachments: [ + { + type: "image", + id: "../escape-attachments-dir", + name: "photo.png", + mimeType: "image/png", + sizeBytes: 4, + }, + ], + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "composer-2" }, + }) + .pipe(Effect.flip, Effect.forkChild); + const steeredTurnFiber = yield* adapter + .sendTurn({ + threadId, + input: "hello mock", + attachments: [], + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }) + .pipe(Effect.forkChild); + + const failure = yield* Fiber.join(failingTurnFiber); + assert.instanceOf(failure, ProviderAdapterRequestError); + assert.match(failure.detail, /Invalid attachment id/u); + const steeredTurn = yield* Fiber.join(steeredTurnFiber); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const turnStartedEvents = runtimeEvents.filter((event) => event.type === "turn.started"); + const turnCompletedEvents = runtimeEvents.filter((event) => event.type === "turn.completed"); + + // The surviving prompt must still announce its turn: completing a turn + // that never emitted turn.started leaves the UI idle without a stop + // affordance while the agent is still working. + assert.equal(turnStartedEvents.length, 1); + assert.equal(String(turnStartedEvents[0]?.turnId), String(steeredTurn.turnId)); + assert.equal(turnCompletedEvents.length, 1); + assert.equal(String(turnCompletedEvents[0]?.turnId), String(steeredTurn.turnId)); + + // The settled reservation must not wedge the session: a fresh sendTurn + // opens a fresh turn. + const nextTurn = yield* adapter.sendTurn({ + threadId, + input: "hello again", + attachments: [], + }); + assert.notEqual(String(nextTurn.turnId), String(steeredTurn.turnId)); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("settles with the recorded result when the last holder is interrupted", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-interrupt-last-holder-thread"); + + // The reserving prompt gets stuck in its delayed model change while the + // steered prompt joins, announces the turn, and resolves — recording a + // stop reason without settling (two prompts still counted). The stuck + // reserver is then fiber-interrupted: the drain must settle with the + // recorded result, not skip on the interrupt and leave the announced + // turn open forever. + const wrapperPath = yield* Effect.promise(() => + makeMockAgentWrapper({ T3_ACP_SET_CONFIG_OPTION_DELAY_MS: "30000" }), + ); + yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } }); + + const runtimeEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("cursor"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }); + + const stuckReserverFiber = yield* adapter + .sendTurn({ + threadId, + input: "first prompt", + attachments: [], + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "composer-2" }, + }) + .pipe(Effect.forkChild); + const steeredTurn = yield* adapter.sendTurn({ + threadId, + input: "second prompt", + attachments: [], + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }); + + // The steered prompt has fully drained; the reserver still holds its + // in-flight count inside the delayed configuration step. + yield* Fiber.interrupt(stuckReserverFiber); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const turnStartedEvents = runtimeEvents.filter((event) => event.type === "turn.started"); + const turnCompletedEvents = runtimeEvents.filter((event) => event.type === "turn.completed"); + + assert.equal(turnStartedEvents.length, 1); + assert.equal(String(turnStartedEvents[0]?.turnId), String(steeredTurn.turnId)); + assert.equal(turnCompletedEvents.length, 1); + assert.equal(String(turnCompletedEvents[0]?.turnId), String(steeredTurn.turnId)); + const completion = turnCompletedEvents[0]; + if (completion?.type === "turn.completed") { + assert.equal(completion.payload.state, "completed"); + } + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("settles the announced turn as failed when its only prompt fails", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-prompt-fails-after-announce-thread"); + + const wrapperPath = yield* Effect.promise(() => + makeMockAgentWrapper({ T3_ACP_FAIL_FIRST_PROMPT: "1" }), + ); + yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } }); + + let completedCount = 0; + const runtimeEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => { + if (event.type === "turn.completed") { + completedCount += 1; + } + return completedCount === 2; + }), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("cursor"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }); + + // The turn is announced, then its only prompt fails. The error must + // reach the caller AND the announced turn must settle as failed: + // reporting success would mask the error and flip the session to + // ready, while emitting nothing would leave the turn.started + // unanswered and wedge the thread in "working". + const failure = yield* adapter + .sendTurn({ + threadId, + input: "hello mock", + attachments: [], + }) + .pipe(Effect.flip); + assert.instanceOf(failure, ProviderAdapterRequestError); + + const nextTurn = yield* adapter.sendTurn({ + threadId, + input: "hello again", + attachments: [], + }); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const turnStartedEvents = runtimeEvents.filter((event) => event.type === "turn.started"); + const turnCompletedEvents = runtimeEvents.filter((event) => event.type === "turn.completed"); + + assert.equal(turnStartedEvents.length, 2); + assert.equal(turnCompletedEvents.length, 2); + + // The failed turn settles as failed — after its turn.started, so + // ingestion deterministically ends in the error state. + const failedCompletion = turnCompletedEvents[0]; + assert.equal(String(failedCompletion?.turnId), String(turnStartedEvents[0]?.turnId)); + assert.notEqual(String(failedCompletion?.turnId), String(nextTurn.turnId)); + if (failedCompletion?.type === "turn.completed") { + assert.equal(failedCompletion.payload.state, "failed"); + assert.isDefined(failedCompletion.payload.errorMessage); + } + + // The follow-up turn is untainted by the failed one. + const successCompletion = turnCompletedEvents[1]; + assert.equal(String(successCompletion?.turnId), String(nextTurn.turnId)); + if (successCompletion?.type === "turn.completed") { + assert.equal(successCompletion.payload.state, "completed"); + } + + yield* adapter.stopSession(threadId); + }), + ); + it.effect("closes the ACP child process when a session stops", () => Effect.gen(function* () { const adapter = yield* CursorAdapter; diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index 4dd38519c5a..dd9641ebea2 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -21,6 +21,7 @@ import { type ThreadId, TurnId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Crypto from "effect/Crypto"; import * as Deferred from "effect/Deferred"; @@ -133,6 +134,15 @@ interface CursorSessionContext { readonly turns: Array<{ id: TurnId; items: Array }>; lastPlanFingerprint: string | undefined; activeTurnId: TurnId | undefined; + /** Whether the active turn has emitted turn.started. Whichever prompt + * survives preparation first announces the turn, so a steered prompt is + * not silenced when the reserving prompt fails before announcing. */ + activeTurnStarted: boolean; + /** Whether the active turn has emitted turn.completed. */ + activeTurnSettled: boolean; + /** Stop reason of the most recent prompt resolution for the active turn, + * kept so a drain-time settle can report it. */ + activeTurnLastStopReason: string | null | undefined; /** 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. */ @@ -778,6 +788,9 @@ export function makeCursorAdapter( turns: [], lastPlanFingerprint: undefined, activeTurnId: undefined, + activeTurnStarted: false, + activeTurnSettled: false, + activeTurnLastStopReason: undefined, promptsInFlight: 0, stopped: false, }; @@ -917,8 +930,19 @@ export function makeCursorAdapter( 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; the matching - // decrement is the `ensuring` below. + // decrement is the `ensuring` below. The turn id is reserved in the + // same synchronous step: a concurrent sendTurn arriving while this + // one awaits session configuration or attachment I/O must observe + // this turn id and steer into it, not mint a second one. The + // reservation is adapter-internal; the session snapshot and the + // turn.started emission stay after validation. ctx.promptsInFlight += 1; + ctx.activeTurnId = turnId; + if (steeringTurnId === undefined) { + ctx.activeTurnStarted = false; + ctx.activeTurnSettled = false; + ctx.activeTurnLastStopReason = undefined; + } return yield* Effect.gen(function* () { const turnModelSelection = @@ -939,27 +963,11 @@ export function makeCursorAdapter( mapError: ({ cause, method }) => mapAcpToAdapterError(PROVIDER, input.threadId, method, cause), }); - ctx.activeTurnId = turnId; - if (steeringTurnId === undefined) { - ctx.lastPlanFingerprint = undefined; - } - ctx.session = { - ...ctx.session, - activeTurnId: turnId, - updatedAt: yield* nowIso, - }; - - if (steeringTurnId === undefined) { - yield* offerRuntimeEvent({ - type: "turn.started", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId, - payload: { model: resolvedModel }, - }); - } - + // Validate/build the prompt content BEFORE any turn-state + // mutation or turn.started emission below. Attachment validation + // can fail (e.g. an unresolvable attachment id) and must not + // leave the session with an active turn id and an already-emitted + // turn.started — that combination wedges the turn forever. const promptParts: Array = []; if (input.input?.trim()) { promptParts.push({ type: "text", text: input.input.trim() }); @@ -1004,6 +1012,31 @@ export function makeCursorAdapter( }); } + if (steeringTurnId === undefined) { + ctx.lastPlanFingerprint = undefined; + } + ctx.session = { + ...ctx.session, + activeTurnId: turnId, + updatedAt: yield* nowIso, + }; + + // Whichever prompt survives preparation first announces the turn. + // Guarding on the steering check instead would silence a steered + // prompt whose reserving prompt failed validation before emitting + // turn.started — the turn would then complete without ever starting. + if (!ctx.activeTurnStarted) { + ctx.activeTurnStarted = true; + yield* offerRuntimeEvent({ + type: "turn.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: { model: resolvedModel }, + }); + } + const result = yield* ctx.acp .prompt({ prompt: promptParts, @@ -1030,7 +1063,9 @@ export function makeCursorAdapter( // Only the last remaining prompt settles the turn — a steer- // superseded prompt resolving (usually cancelled) while another is // in flight or pending must leave the merged turn running. + ctx.activeTurnLastStopReason = result.stopReason ?? null; if (ctx.promptsInFlight === 1) { + ctx.activeTurnSettled = true; yield* offerRuntimeEvent({ type: "turn.completed", ...(yield* makeEventStamp()), @@ -1050,10 +1085,79 @@ export function makeCursorAdapter( resumeCursor: ctx.session.resumeCursor, }; }).pipe( - Effect.ensuring( - Effect.sync(() => { + Effect.onExit((exit) => + Effect.gen(function* () { ctx.promptsInFlight = Math.max(0, ctx.promptsInFlight - 1); - }), + if (ctx.promptsInFlight > 0 || ctx.activeTurnId !== turnId) { + return; + } + if (!ctx.activeTurnStarted) { + // The reservation never announced a turn (every prompt failed + // preparation); drop it so the next sendTurn opens a fresh + // turn instead of steering into a dead one. + ctx.activeTurnId = undefined; + return; + } + if (ctx.activeTurnSettled) { + return; + } + // The last prompt drained without the turn settling: either it + // exited after another prompt of the merged turn resolved while + // both were still counted, or every prompt failed after the + // turn was announced. Settle with the last known result, or as + // failed when no prompt ever resolved — reporting success there + // would mask the error and flip the session to ready, while + // emitting nothing would leave the turn.started unanswered and + // wedge the thread in "working". This emission trails the + // turn.started already in the queue, so ingestion ends in the + // error state regardless of how the caller's turn-start failure + // recovery is interleaved. + if (ctx.activeTurnLastStopReason === undefined) { + // An interrupted drain with no recorded result is session + // teardown, not a turn outcome; leave settling to the session + // lifecycle events. A recorded sibling result, by contrast, + // IS the turn's outcome and is settled below no matter how + // this last holder exited. + if (Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)) { + return; + } + ctx.activeTurnSettled = true; + const failureMessage = Exit.isFailure(exit) ? Cause.squash(exit.cause) : undefined; + const errorMessage = + failureMessage instanceof Error && failureMessage.message.trim().length > 0 + ? failureMessage.message.trim() + : "Cursor turn failed before any prompt resolved."; + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: { + state: "failed", + stopReason: null, + errorMessage, + }, + }); + return; + } + ctx.activeTurnSettled = true; + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: { + state: ctx.activeTurnLastStopReason === "cancelled" ? "cancelled" : "completed", + stopReason: ctx.activeTurnLastStopReason ?? null, + }, + }); + // The only failure above is event-stamp id generation + // (crypto.randomUUIDv4); a finalizer cannot surface it as a + // typed error, and swallowing it would silently drop the + // settling turn.completed. + }).pipe(Effect.orDie), ), ); });