diff --git a/apps/mobile/src/features/threads/PendingUserInputCard.tsx b/apps/mobile/src/features/threads/PendingUserInputCard.tsx index 5700d1b79e44..a9edaa7d0fb3 100644 --- a/apps/mobile/src/features/threads/PendingUserInputCard.tsx +++ b/apps/mobile/src/features/threads/PendingUserInputCard.tsx @@ -36,6 +36,8 @@ export interface PendingUserInputCardProps { readonly onToggleCollapsed: () => void; /** Renders a stop control on the collapsed bar, which replaces the composer. */ readonly onStopThread?: () => void; + /** Optimistic Stop feedback while the interrupt settles (#8618). */ + readonly isStoppingThread?: boolean; /** * 0 collapsed → 1 expanded. Slides the iOS overlay card down behind the * collapsed bar (inside a clipping window) on the UI thread; the host @@ -187,10 +189,11 @@ export function PendingUserInputCard(props: PendingUserInputCardProps) { {props.onStopThread ? ( ) : null} diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index 174e4d5cd2c0..699e0055f636 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -124,6 +124,8 @@ export interface ThreadComposerProps { readonly onNativePasteImages: (uris: ReadonlyArray) => Promise; readonly onRemoveDraftImage: (imageId: string) => void; readonly onStopThread: () => void; + /** Optimistic Stop feedback: disables Stop while the interrupt settles (#8618). */ + readonly isStoppingThread: boolean; readonly onSendMessage: () => Promise; readonly onUpdateModelSelection: (modelSelection: ModelSelection) => void; readonly onUpdateRuntimeMode: (runtimeMode: RuntimeMode) => void; @@ -716,9 +718,10 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer /> {showStopAction ? ( ) : ( @@ -807,9 +810,10 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer /> {showStopAction ? ( ) : voicePresentation.showsSend ? ( diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 15cb6593380d..ef63e0750e15 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -127,6 +127,8 @@ export interface ThreadDetailScreenProps { readonly onNativePasteImages: (uris: ReadonlyArray) => Promise; readonly onRemoveDraftImage: (imageId: string) => void; readonly onStopThread: () => void; + /** Optimistic Stop feedback: set on press, cleared when work ends (#8618). */ + readonly isStoppingThread: boolean; readonly onSendMessage: () => Promise; readonly onReconnectEnvironment: () => void; readonly onUpdateThreadModelSelection: (modelSelection: ModelSelection) => void; @@ -780,6 +782,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread collapsed={userInputCollapsed} onToggleCollapsed={handleToggleUserInputCollapsed} onStopThread={props.onStopThread} + isStoppingThread={props.isStoppingThread} cardProgress={userInputCardProgress} cardCoverage={userInputCardCoverage} onInputFocusChange={handleOwnedInputFocusChange} @@ -820,6 +823,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread onNativePasteImages={props.onNativePasteImages} onRemoveDraftImage={props.onRemoveDraftImage} onStopThread={props.onStopThread} + isStoppingThread={props.isStoppingThread} onSendMessage={handleSendMessage} onReconnectEnvironment={props.onReconnectEnvironment} onUpdateModelSelection={props.onUpdateThreadModelSelection} diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 79e898eaa1c9..ddf72ad4c1dd 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -214,6 +214,18 @@ function ThreadRouteContent( const gitActions = useSelectedThreadGitActions(); const requests = useSelectedThreadRequests(); const interruptThreadTurn = useAtomCommand(threadEnvironment.interruptTurn, "thread interrupt"); + const [isStoppingThread, setIsStoppingThread] = useState(false); + const selectedThreadSessionStatus = selectedThread?.session?.status ?? null; + const selectedThreadIdentity = selectedThread + ? scopedThreadKey(selectedThread.environmentId, selectedThread.id) + : null; + useEffect(() => { + // "Stopping" holds until the session leaves running/starting; per-thread + // so a switch never leaks the state onto another thread. + if (selectedThreadSessionStatus !== "running" && selectedThreadSessionStatus !== "starting") { + setIsStoppingThread(false); + } + }, [selectedThreadIdentity, selectedThreadSessionStatus]); const navigation = useNavigation(); const params = props.route.params; const environmentIdRaw = firstRouteParam(params.environmentId); @@ -486,7 +498,11 @@ function ThreadRouteContent( ) { return; } - return interruptThreadTurn({ + // Optimistic stopping feedback for remote links: holds until the session + // leaves running/starting (reset effect below); a failed interrupt clears + // it immediately so Stop never sticks (#8618). + setIsStoppingThread(true); + void interruptThreadTurn({ environmentId: selectedThread.environmentId, input: { threadId: selectedThread.id, @@ -494,6 +510,10 @@ function ThreadRouteContent( ? { turnId: selectedThread.session.activeTurnId } : {}), }, + }).then((result) => { + if (result._tag === "Failure") { + setIsStoppingThread(false); + } }); }, [interruptThreadTurn, selectedThread]); @@ -799,6 +819,7 @@ function ThreadRouteContent( onRemoveDraftImage={composer.onRemoveDraftImage} serverConfig={serverConfig} onStopThread={handleStopThread} + isStoppingThread={isStoppingThread} onSendMessage={composer.onSendMessage} onReconnectEnvironment={handleReconnectEnvironment} onUpdateThreadModelSelection={composer.onUpdateModelSelection} diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 504fa7c52542..3c54019423d4 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1839,6 +1839,99 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect( + "marks the session-pinned turn interrupted when the interrupt request omits turnId", + () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const now = "2026-01-01T00:00:00.000Z"; + const threadId = ThreadId.make("thread-interrupt-no-turn-id"); + const turnId = TurnId.make("turn-no-turn-id-1"); + + yield* eventStore.append({ + type: "thread.created", + eventId: EventId.make("evt-nti1"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-nti1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-nti1"), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-no-turn-id"), + title: "Interrupt without turn id", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.session-set", + eventId: EventId.make("evt-nti2"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:00:01.000Z", + commandId: CommandId.make("cmd-nti2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-nti2"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "running", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: turnId, + lastError: null, + updatedAt: "2026-01-01T00:00:01.000Z", + }, + }, + }); + + // Remote snapshots lag, so Stop can persist an interrupt request + // without a turnId — the pinned turn must still be settled (#8618). + yield* eventStore.append({ + type: "thread.turn-interrupt-requested", + eventId: EventId.make("evt-nti3"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:00:02.000Z", + commandId: CommandId.make("cmd-nti3"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-nti3"), + metadata: {}, + payload: { + threadId, + createdAt: "2026-01-01T00:00:02.000Z", + }, + }); + + yield* projectionPipeline.bootstrap; + + const rows = yield* sql<{ + readonly state: string; + readonly completedAt: string | null; + }>` + SELECT state, completed_at AS "completedAt" + FROM projection_turns + WHERE thread_id = ${threadId} AND turn_id = ${turnId} + `; + assert.deepEqual(rows, [{ state: "interrupted", completedAt: "2026-01-01T00:00:02.000Z" }]); + }), + ); + it.effect("settles a superseded running turn when a new turn becomes active", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 3de33474d205..1e8c9a82ed43 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1472,12 +1472,25 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } case "thread.turn-interrupt-requested": { - if (event.payload.turnId === undefined) { + // Remote snapshots lag, so Stop often arrives without a turnId + // (see buildThreadTurnInterruptInput). Fall back to the session's + // pinned turn instead of dropping the interrupt on the floor — a + // turnId-less interrupt must still settle the running turn (#8618). + let effectiveTurnId = event.payload.turnId; + if (effectiveTurnId === undefined) { + const session = yield* projectionThreadSessionRepository.getByThreadId({ + threadId: event.payload.threadId, + }); + if (Option.isSome(session) && session.value.activeTurnId !== null) { + effectiveTurnId = session.value.activeTurnId; + } + } + if (effectiveTurnId === undefined) { return; } const existingTurn = yield* projectionTurnRepository.getByTurnId({ threadId: event.payload.threadId, - turnId: event.payload.turnId, + turnId: effectiveTurnId, }); if (Option.isSome(existingTurn)) { yield* projectionTurnRepository.upsertByTurnId({ @@ -1490,7 +1503,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } yield* projectionTurnRepository.upsertByTurnId({ - turnId: event.payload.turnId, + turnId: effectiveTurnId, threadId: event.payload.threadId, pendingMessageId: null, sourceProposedPlanThreadId: null, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index cad80f1d3bca..fb2abdfe11d3 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -2668,6 +2668,169 @@ describe("ProviderCommandReactor", () => { }); }); + effectIt.effect("releases a running session whose turn already ended when Stop repeats", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const now = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-zombie"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: asTurnId("turn-1"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + + // First Stop settles the turn in projection, but the provider ack + // leaves the session pinned at running — the #4713 zombie. + yield* harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-turn-interrupt-zombie-1"), + threadId: ThreadId.make("thread-1"), + turnId: asTurnId("turn-1"), + createdAt: now, + }); + + yield* Effect.promise(() => + waitFor(async () => { + const thread = (await harness.readModel()).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + return thread?.latestTurn?.state === "interrupted"; + }), + ); + + // Repeat Stop arrives without a turnId (remote snapshot lag). It must + // release the wedged session instead of being accepted with no effect. + yield* harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-turn-interrupt-zombie-2"), + threadId: ThreadId.make("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + }); + + yield* Effect.promise(() => + waitFor(async () => { + const thread = (await harness.readModel()).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + return thread?.session?.status === "stopped"; + }), + ); + + const thread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + expect(thread?.session).toMatchObject({ status: "stopped", activeTurnId: null }); + expect(harness.stopSession).toHaveBeenCalledWith({ threadId: ThreadId.make("thread-1") }); + }), + ); + + effectIt.effect("escalates a repeat Stop when a turn-less session stays running", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const first = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-escalate"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: first, + }, + createdAt: first, + }); + + yield* harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-turn-interrupt-escalate-1"), + threadId: ThreadId.make("thread-1"), + createdAt: first, + }); + + yield* Effect.promise(() => waitFor(() => harness.interruptTurn.mock.calls.length === 1)); + // The provider ignored the abort: session still running, no escalation yet. + expect(harness.stopSession).not.toHaveBeenCalled(); + + // A repeat Stop past the grace period escalates to a full session stop. + yield* harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-turn-interrupt-escalate-2"), + threadId: ThreadId.make("thread-1"), + createdAt: "2026-01-01T00:00:06.000Z", + }); + + yield* Effect.promise(() => + waitFor(async () => { + const thread = (await harness.readModel()).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + return thread?.session?.status === "stopped"; + }), + ); + + expect(harness.stopSession).toHaveBeenCalledWith({ threadId: ThreadId.make("thread-1") }); + const thread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + expect(thread?.session).toMatchObject({ status: "stopped", activeTurnId: null }); + }), + ); + + effectIt.effect("does not escalate a repeat Stop inside the grace period", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const first = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-no-escalate"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: first, + }, + createdAt: first, + }); + + yield* harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-turn-interrupt-no-escalate-1"), + threadId: ThreadId.make("thread-1"), + createdAt: first, + }); + + yield* harness.engine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-turn-interrupt-no-escalate-2"), + threadId: ThreadId.make("thread-1"), + createdAt: "2026-01-01T00:00:02.000Z", + }); + + yield* Effect.promise(() => waitFor(() => harness.interruptTurn.mock.calls.length === 2)); + expect(harness.stopSession).not.toHaveBeenCalled(); + }), + ); + effectIt.effect( "stops a running session and records the failure when provider interrupt fails", () => diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 57edb60ff715..0380740c242c 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -7,6 +7,7 @@ import { ProviderDriverKind, type ProjectId, type OrchestrationSession, + type OrchestrationLatestTurn, ThreadId, type ProviderSession, type RuntimeMode, @@ -1243,6 +1244,149 @@ const make = Effect.gen(function* () { .pipe(Effect.catchCause(recoverTurnStartFailure), Effect.forkScoped); }); + /** + * Grace period before a repeat Stop escalates to a full session stop. The + * provider settles the turn asynchronously after a successful interrupt, so + * the first Stop must not tear the session down — but when the same turn is + * still pinned running well after a previous Stop, the provider ignored the + * abort and the repeat Stop has to release it (#8618, #4713). + */ + const INTERRUPT_ESCALATION_GRACE_MS = 5_000; + const recentTurnInterrupts = new Map(); + + const releaseWedgedSession = Effect.fnUntraced(function* (input: { + readonly threadId: ThreadId; + readonly turnId: TurnId | null; + readonly reason: string; + readonly createdAt: string; + }) { + // Best-effort kill of a possibly-wedged provider session (e.g. a CLI + // child that ignored the abort and still holds the turn open). + yield* providerService.stopSession({ threadId: input.threadId }).pipe( + Effect.catchCause((stopCause) => + Cause.hasInterruptsOnly(stopCause) + ? Effect.interrupt + : Effect.logWarning( + "provider command reactor failed to stop wedged session after interrupt", + { + threadId: input.threadId, + cause: Cause.pretty(stopCause), + }, + ), + ), + ); + const latestThread = yield* resolveThread(input.threadId); + const latestSession = latestThread?.session; + if (!latestSession || latestSession.status !== "running") { + return; + } + if ( + input.turnId !== null && + latestSession.activeTurnId !== null && + latestSession.activeTurnId !== input.turnId + ) { + // A new turn started while stopping; leave it alone. + return; + } + yield* setThreadSession({ + threadId: input.threadId, + session: { + ...latestSession, + status: "stopped", + activeTurnId: null, + lastError: input.reason, + updatedAt: input.createdAt, + }, + createdAt: input.createdAt, + }); + yield* appendProviderFailureActivity({ + threadId: input.threadId, + kind: "provider.turn.interrupt.failed", + summary: "Provider turn interrupt timed out", + detail: input.reason, + turnId: input.turnId, + createdAt: input.createdAt, + }); + }); + + /** + * Releases sessions a successful interrupt left behind. Returns true when + * the session was released and the normal interrupt path must be skipped: + * + * - the pinned turn already reached a terminal state but the session still + * claims running (the #4713 zombie: every further Stop is accepted but + * has nothing to act on), or + * - the same turn is still pinned running well after a previous Stop, so + * the provider ignored the abort and this repeat Stop escalates. + * + * Otherwise records this request so a later repeat Stop can escalate, and + * returns false. + */ + const maybeReleaseWedgedSession = Effect.fnUntraced(function* (input: { + readonly threadId: ThreadId; + readonly session: OrchestrationSession; + readonly latestTurn: OrchestrationLatestTurn | null; + readonly createdAt: string; + }) { + if (input.session.status !== "running") { + return false; + } + const activeTurnId = input.session.activeTurnId; + const latestTurn = input.latestTurn; + // The pinned turn already reached a terminal state but the session still + // claims running — release it now (#4713, #8802). The completedAt guard + // matters: this very event's own projection also settles the turn with + // completedAt === createdAt, and that healthy in-flight interrupt must + // proceed to the provider instead of being mistaken for a zombie. + if ( + latestTurn !== null && + (activeTurnId === null || latestTurn.turnId === activeTurnId) && + latestTurn.state !== "running" && + latestTurn.completedAt !== null && + latestTurn.completedAt < input.createdAt + ) { + recentTurnInterrupts.delete(input.threadId); + yield* releaseWedgedSession({ + threadId: input.threadId, + turnId: activeTurnId, + reason: + `Turn ${latestTurn.turnId} already ${latestTurn.state} but the session stayed ` + + `running; Stop released it.`, + createdAt: input.createdAt, + }); + return true; + } + const requestedAtMs = Date.parse(input.createdAt); + const previous = recentTurnInterrupts.get(input.threadId); + // The same session is still pinned running well after a previous Stop + // (or has no pinned turn at all) — the provider ignored the abort, so + // this repeat Stop escalates to a full session stop (#8618). + if ( + previous !== undefined && + previous.turnId === activeTurnId && + Number.isFinite(requestedAtMs) && + requestedAtMs - previous.requestedAtMs >= INTERRUPT_ESCALATION_GRACE_MS + ) { + recentTurnInterrupts.delete(input.threadId); + yield* releaseWedgedSession({ + threadId: input.threadId, + turnId: activeTurnId, + reason: + `Stop did not settle the session within ` + + `${INTERRUPT_ESCALATION_GRACE_MS / 1000}s; the session was stopped.`, + createdAt: input.createdAt, + }); + return true; + } + // Remember this request so a later repeat Stop can escalate. + if (Number.isFinite(requestedAtMs)) { + recentTurnInterrupts.set(input.threadId, { turnId: activeTurnId, requestedAtMs }); + } else { + recentTurnInterrupts.delete(input.threadId); + } + return false; + }); + const processTurnInterruptRequested = Effect.fn("processTurnInterruptRequested")(function* ( event: Extract, ) { @@ -1262,6 +1406,21 @@ const make = Effect.gen(function* () { }); } + // A successful interrupt settles the turn asynchronously, but a wedged + // provider can leave the session pinned at running forever while every + // further Stop is accepted with no effect (#4713, #8802, #8618). Release + // such sessions here instead of waiting for a session-set that never + // comes. + const released = yield* maybeReleaseWedgedSession({ + threadId: event.payload.threadId, + session, + latestTurn: thread.latestTurn, + createdAt: event.payload.createdAt, + }); + if (released) { + return; + } + const recoverInterruptFailure = (cause: Cause.Cause) => { if (Cause.hasInterruptsOnly(cause)) { return Effect.interrupt; @@ -1486,6 +1645,7 @@ const make = Effect.gen(function* () { return; } case "thread.turn-start-requested": + recentTurnInterrupts.delete(event.payload.threadId); yield* processTurnStartRequested(event); return; case "thread.turn-interrupt-requested": @@ -1498,9 +1658,11 @@ const make = Effect.gen(function* () { yield* processUserInputResponseRequested(event); return; case "thread.session-stop-requested": + recentTurnInterrupts.delete(event.payload.threadId); yield* processSessionStopRequested(event); return; case "thread.settled": { + recentTurnInterrupts.delete(event.payload.threadId); const thread = yield* projectionSnapshotQuery.getThreadShellById(event.payload.threadId); if ( Option.isNone(thread) || diff --git a/apps/server/src/provider/opencodeRuntime.ts b/apps/server/src/provider/opencodeRuntime.ts index afd806e5666e..85ba2a21cbc2 100644 --- a/apps/server/src/provider/opencodeRuntime.ts +++ b/apps/server/src/provider/opencodeRuntime.ts @@ -35,6 +35,7 @@ import * as NetService from "@t3tools/shared/Net"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { compareSemverVersions, parseSemver } from "@t3tools/shared/semver"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import { sanitizeTerminalValue, stripTerminalEscapes } from "@t3tools/shared/stripTerminalEscapes"; const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); const OPENCODE_EMPTY_CONFIG_CONTENT = "{}"; @@ -300,7 +301,7 @@ export function parseModelsCliOutput(stdout: string): { string, { id: string; name: string; models: { [key: string]: Model } } >(); - const lines = stdout.split("\n"); + const lines = stripTerminalEscapes(stdout).split("\n"); let currentSlug: string | null = null; const jsonLines: Array = []; @@ -353,7 +354,7 @@ export function parseModelsCliOutput(stdout: string): { /** @internal */ export function parseAgentListCliOutput(stdout: string): ReadonlyArray { const agents: Array = []; - const lines = stdout.split("\n"); + const lines = stripTerminalEscapes(stdout).split("\n"); let currentHeader: { name: string; mode: string } | null = null; const blockLines: Array = []; @@ -395,7 +396,8 @@ export function parseAgentListCliOutput(stdout: string): ReadonlyArray { /** @internal */ export function parseSkillsCliOutput(stdout: string): ReadonlyArray { - const result = decodeOpenCodeSkillsCliOutputExit(stdout); + const clean = stripTerminalEscapes(stdout); + const result = decodeOpenCodeSkillsCliOutputExit(clean); return Exit.isSuccess(result) ? result.value : []; } diff --git a/apps/server/src/textGeneration/OpenCodeTextGeneration.ts b/apps/server/src/textGeneration/OpenCodeTextGeneration.ts index e0e960422b18..545e63613cb7 100644 --- a/apps/server/src/textGeneration/OpenCodeTextGeneration.ts +++ b/apps/server/src/textGeneration/OpenCodeTextGeneration.ts @@ -11,6 +11,7 @@ import { import { sanitizeBranchFragment, sanitizeFeatureBranchName } from "@t3tools/shared/git"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { extractJsonObject } from "@t3tools/shared/schemaJson"; +import { sanitizeTerminalValue } from "@t3tools/shared/stripTerminalEscapes"; import * as ServerConfig from "../config.ts"; import { resolveAttachmentPath } from "../attachmentStore.ts"; @@ -230,8 +231,10 @@ export const makeOpenCodeTextGeneration = Effect.fn("makeOpenCodeTextGeneration" cwd: input.cwd, }); } - const selectedAgent = getModelSelectionStringOptionValue(input.modelSelection, "agent"); - const selectedVariant = getModelSelectionStringOptionValue(input.modelSelection, "variant"); + const rawAgent = getModelSelectionStringOptionValue(input.modelSelection, "agent"); + const rawVariant = getModelSelectionStringOptionValue(input.modelSelection, "variant"); + const selectedAgent = rawAgent ? sanitizeTerminalValue(rawAgent) : undefined; + const selectedVariant = rawVariant ? sanitizeTerminalValue(rawVariant) : undefined; const promptContext = { operation: input.operation, cwd: input.cwd, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index e7f1a23bbc47..4fcca5cc94d3 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -2525,6 +2525,19 @@ function ChatViewContent(props: ChatViewProps) { threadError, }); const isWorking = phase === "running" || isSendBusy || isConnecting || isRevertingCheckpoint; + // Optimistic stopping state for remote: mirrors isStoppingBackgroundWork so Stop + // gives immediate feedback even though session stays "running" until provider settles. + // See fix-remote-stop-no-feedback-stuck-thinking-7h3k9p2m (issue #8618). + const [isStoppingTurn, setIsStoppingTurn] = useState(false); + useEffect(() => { + if (!isWorking) { + setIsStoppingTurn(false); + } + }, [isWorking]); + useEffect(() => { + // Per-thread: switching threads must not leak Stopping... to B + setIsStoppingTurn(false); + }, [activeThreadId]); const activeWorkStartedAt = deriveActiveWorkStartedAt( activeLatestTurn, activeThread?.session ?? null, @@ -6524,17 +6537,23 @@ function ChatViewContent(props: ChatViewProps) { const onInterrupt = async () => { if (!activeThread) return; + setIsStoppingTurn(true); const result = await interruptThreadTurn({ environmentId, input: buildThreadTurnInterruptInput(activeThread), }); - if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { - const error = squashAtomCommandFailure(result); - setThreadError( - activeThread.id, - error instanceof Error ? error.message : "Failed to interrupt the current turn.", - ); + if (result._tag === "Failure") { + setIsStoppingTurn(false); + if (!isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + setThreadError( + activeThread.id, + error instanceof Error ? error.message : "Failed to interrupt the current turn.", + ); + } } + // Success clears via isWorking effect when session leaves running; keep Stopping... + // visible during relay RTT so remote click is not silent. }; const onRespondToApproval = useCallback( @@ -7592,6 +7611,7 @@ function ChatViewContent(props: ChatViewProps) { phase={phase} isConnecting={isConnecting} isSendBusy={isSendBusy} + isStoppingTurn={isStoppingTurn} sendDisabledReason={ feedbackUploading ? "Sending feedback" diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 349faeac0728..d9ce1d4bf412 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -521,6 +521,8 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions( isComplete: boolean; } | null; isRunning: boolean; + /** Optimistic Stop feedback while the session still reports running. */ + isStopping: boolean; showPlanFollowUpPrompt: boolean; promptHasText: boolean; isSendBusy: boolean; @@ -552,6 +554,7 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions( compact={props.compact} pendingAction={props.pendingAction} isRunning={props.isRunning} + isStopping={props.isStopping} showPlanFollowUpPrompt={props.showPlanFollowUpPrompt} promptHasText={props.promptHasText} isSendBusy={props.isSendBusy} @@ -649,6 +652,8 @@ export interface ChatComposerProps { phase: SessionPhase; isConnecting: boolean; isSendBusy: boolean; + /** Optimistic Stop feedback: set on click, cleared when work ends (#8618). */ + isStoppingTurn: boolean; sendDisabledReason: string | null; isPreparingWorktree: boolean; bannerItems: readonly ComposerBannerStackItem[]; @@ -767,6 +772,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) phase, isConnecting, isSendBusy, + isStoppingTurn, sendDisabledReason: externalSendDisabledReason, isPreparingWorktree, environmentUnavailable, @@ -4379,6 +4385,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) activeThreadModelDisplayName={activeThreadModelDisplayName} pendingAction={pendingPrimaryAction} isRunning={phase === "running"} + isStopping={isStoppingTurn} showPlanFollowUpPrompt={ pendingUserInputs.length === 0 && showPlanFollowUpPrompt } diff --git a/apps/web/src/components/chat/ComposerPrimaryActions.test.tsx b/apps/web/src/components/chat/ComposerPrimaryActions.test.tsx index 92f24c833db8..f23f97ac3dba 100644 --- a/apps/web/src/components/chat/ComposerPrimaryActions.test.tsx +++ b/apps/web/src/components/chat/ComposerPrimaryActions.test.tsx @@ -44,12 +44,17 @@ function renderPendingActions(isRunning: boolean) { ); } -function renderRunningActions(showSendWhileRunning: boolean, hasSendableContent: boolean) { +function renderRunningActions( + showSendWhileRunning: boolean, + hasSendableContent: boolean, + isStopping = false, +) { return renderToStaticMarkup( createElement(ComposerPrimaryActions, { compact: true, pendingAction: null, isRunning: true, + isStopping, showPlanFollowUpPrompt: false, promptHasText: hasSendableContent, isSendBusy: false, @@ -236,4 +241,12 @@ describe("ComposerPrimaryActions", () => { expect(markup).toContain('aria-label="Stop generation"'); expect(markup).not.toContain('aria-label="Send message"'); }); + + it("disables stop and announces Stopping generation while the interrupt settles", () => { + const markup = renderRunningActions(false, false, true); + + expect(markup).toContain('aria-label="Stopping generation"'); + expect(markup).toContain("disabled"); + expect(markup).not.toContain('aria-label="Stop generation"'); + }); }); diff --git a/apps/web/src/components/chat/ComposerPrimaryActions.tsx b/apps/web/src/components/chat/ComposerPrimaryActions.tsx index 2a27796d92a5..f59f9e2fb38c 100644 --- a/apps/web/src/components/chat/ComposerPrimaryActions.tsx +++ b/apps/web/src/components/chat/ComposerPrimaryActions.tsx @@ -19,6 +19,8 @@ interface ComposerPrimaryActionsProps { compact: boolean; pendingAction: PendingActionState | null; isRunning: boolean; + /** Optimistic Stop feedback: set on click, cleared when work ends (#8618). */ + isStopping?: boolean; showPlanFollowUpPrompt: boolean; promptHasText: boolean; isSendBusy: boolean; @@ -62,6 +64,7 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({ compact, pendingAction, isRunning, + isStopping = false, showPlanFollowUpPrompt, promptHasText, isSendBusy, @@ -89,7 +92,7 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({ ); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index c540644289df..c9eadc7d94d7 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -178,7 +178,14 @@ export function createThreadEnvironmentAtoms( label: "environment-data:commands:thread:interrupt-turn", execute: (input: InterruptThreadTurnInput) => interruptThreadTurn(input), scheduler, - concurrency, + // Stop is pressed repeatedly while a turn winds down (especially on + // high-latency remote links). singleFlight shares the in-flight request + // per thread instead of serializing every click behind it (#8618). + concurrency: { + mode: "singleFlight" as const, + key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => + JSON.stringify([environmentId, input.threadId]), + }, }), respondToApproval: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:respond-to-approval", diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 401980997663..ade41938a0c3 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -705,6 +705,67 @@ describe("applyThreadDetailEvent", () => { }); }); + describe("thread.turn-interrupt-requested", () => { + it("falls back to the session-pinned turn when the payload omits turnId", () => { + const threadWithRunningTurn: OrchestrationThread = { + ...baseThread, + latestTurn: { + turnId: TurnId.make("turn-1"), + state: "running", + requestedAt: "2026-04-01T07:00:00.000Z", + startedAt: "2026-04-01T07:00:00.000Z", + completedAt: null, + assistantMessageId: MessageId.make("msg-3"), + }, + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: TurnId.make("turn-1"), + lastError: null, + updatedAt: "2026-04-01T07:00:00.000Z", + }, + }; + + const result = applyThreadDetailEvent(threadWithRunningTurn, { + ...baseEventFields, + sequence: 10, + occurredAt: "2026-04-01T08:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.turn-interrupt-requested", + payload: { + threadId: ThreadId.make("thread-1"), + createdAt: "2026-04-01T08:00:00.000Z", + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.latestTurn?.state).toBe("interrupted"); + expect(result.thread.latestTurn?.completedAt).toBe("2026-04-01T08:00:00.000Z"); + } + }); + + it("returns unchanged without a turnId and without a pinned turn", () => { + const result = applyThreadDetailEvent(baseThread, { + ...baseEventFields, + sequence: 10, + occurredAt: "2026-04-01T08:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.turn-interrupt-requested", + payload: { + threadId: ThreadId.make("thread-1"), + createdAt: "2026-04-01T08:00:00.000Z", + }, + }); + + expect(result.kind).toBe("unchanged"); + }); + }); + describe("thread.session-stop-requested", () => { it("marks session as stopped", () => { const threadWithSession: OrchestrationThread = { diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index 690c74bdea0a..2ab97c9991b4 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -278,11 +278,15 @@ export function applyThreadDetailEvent( }; case "thread.turn-interrupt-requested": { - if (event.payload.turnId === undefined) { + // Remote snapshots lag, so Stop often arrives without a turnId (see + // buildThreadTurnInterruptInput). Fall back to the session's pinned + // turn instead of ignoring the interrupt (#8618). + const effectiveTurnId = event.payload.turnId ?? thread.session?.activeTurnId ?? null; + if (effectiveTurnId === null) { return { kind: "unchanged" }; } const latestTurn = thread.latestTurn; - if (latestTurn === null || latestTurn.turnId !== event.payload.turnId) { + if (latestTurn === null || latestTurn.turnId !== effectiveTurnId) { return { kind: "unchanged" }; } return { diff --git a/packages/shared/package.json b/packages/shared/package.json index fda7a91b1a2f..ad46c501088d 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -250,6 +250,10 @@ "./nodeSqliteClient": { "types": "./src/nodeSqliteClient.ts", "import": "./src/nodeSqliteClient.ts" + }, + "./stripTerminalEscapes": { + "types": "./src/stripTerminalEscapes.ts", + "import": "./src/stripTerminalEscapes.ts" } }, "scripts": { diff --git a/packages/shared/src/model.ts b/packages/shared/src/model.ts index cad7c8e8db86..1c00f6c418f2 100644 --- a/packages/shared/src/model.ts +++ b/packages/shared/src/model.ts @@ -8,6 +8,8 @@ import { type ProviderOptionSelection, } from "@t3tools/contracts"; +import { sanitizeTerminalValue } from "./stripTerminalEscapes.ts"; + const DEFAULT_PROVIDER_DRIVER_KIND = ProviderDriverKind.make("codex"); export interface SelectableModelOption { @@ -44,7 +46,9 @@ export function getProviderOptionStringSelectionValue( id: string, ): string | undefined { const value = getProviderOptionSelectionValue(selections, id); - return typeof value === "string" ? value : undefined; + if (typeof value !== "string") return undefined; + const sanitized = sanitizeTerminalValue(value); + return sanitized.length > 0 ? sanitized : undefined; } export function getProviderOptionBooleanSelectionValue( @@ -244,7 +248,7 @@ export function normalizeCustomModelSlug(model: string | null | undefined): stri return null; } - return model.trim() || null; + return sanitizeTerminalValue(model) || null; } export function resolveSelectableModel( @@ -290,7 +294,8 @@ export function resolveSelectableModel( /** Trim a string, returning null for empty/missing values. */ export function trimOrNull(value: T | null | undefined): T | null { if (typeof value !== "string") return null; - const trimmed = value.trim() as T; + const sanitized = sanitizeTerminalValue(value); + const trimmed = sanitized.trim() as T; return trimmed || null; } diff --git a/packages/shared/src/stripTerminalEscapes.ts b/packages/shared/src/stripTerminalEscapes.ts new file mode 100644 index 000000000000..0c00ccc55d6d --- /dev/null +++ b/packages/shared/src/stripTerminalEscapes.ts @@ -0,0 +1,38 @@ +/** + * Strip terminal escape sequences from captured CLI stdout. + * + * OpenCode's CLI (and potentially other provider CLIs) can emit OSC title + * sequences (`ESC ]0; BEL` / `ESC \`) and ANSI CSI color codes directly + * to stdout, even when stdout is a pipe. When T3 Code captures that output + * via `ChildProcessSpawner`, those bytes pollute structured parsing — e.g. + * `opencode agent list` becomes `\x1b]0;t3code: ready\x07build (primary)` + * instead of `build (primary)`, causing the agent inventory to store a + * polluted id that later fails with `Agent not found`. + * + * This is defensive for any provider CLI; the regexes are intentionally + * permissive and also handle Ghostty/Zsh title integrations that can leak + * through `shell: true` spawns. + */ +const OSC_RE = /\x1b\].*?(?:\x07|\x1b\\)/g; +const CSI_RE = /\x1b\[[0-9;?]*[ -/]*[@-~]/g; +const CHARSET_RE = /\x1b[()][A-Za-z0-9]/g; +const SINGLE_ESC_RE = /\x1b[@-Z\\-_]/g; + +export function stripTerminalEscapes(input: string): string { + if (!input || input.indexOf("\x1b") === -1) { + return input; + } + return input + .replace(OSC_RE, "") + .replace(CSI_RE, "") + .replace(CHARSET_RE, "") + .replace(SINGLE_ESC_RE, ""); +} + +/** + * Strip escapes and also trim the result. Useful for single-value fields + * like agent/variant names that should never contain control bytes. + */ +export function sanitizeTerminalValue(input: string): string { + return stripTerminalEscapes(input).trim(); +}