From 5e7df51d19417dfa1519ca308f8705bdbff350ba Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 10 Jun 2026 21:53:45 -0700 Subject: [PATCH 1/3] Handle steered turns without closing active sessions - Keep running turns open across interim assistant messages - Treat session status changes as the authoritative turn boundary - Allow pending turn starts to replace superseded active turns --- .../Layers/ProjectionPipeline.test.ts | 221 ++++++++++++++++++ .../Layers/ProjectionPipeline.ts | 119 +++++++++- .../Layers/ProviderRuntimeIngestion.test.ts | 148 +++++++++--- .../Layers/ProviderRuntimeIngestion.ts | 18 +- .../src/orchestration/projector.test.ts | 85 +++++-- apps/server/src/orchestration/projector.ts | 74 ++++-- .../src/provider/Layers/ClaudeAdapter.test.ts | 69 ++++++ .../src/provider/Layers/ClaudeAdapter.ts | 79 ++++--- .../provider/Layers/OpenCodeAdapter.test.ts | 79 +++++++ .../src/provider/Layers/OpenCodeAdapter.ts | 83 ++++--- .../chat/MessagesTimeline.logic.test.ts | 98 +++++++- .../components/chat/MessagesTimeline.logic.ts | 56 ++++- .../src/components/chat/MessagesTimeline.tsx | 2 +- apps/web/src/session-logic.ts | 6 +- apps/web/src/store.test.ts | 33 ++- apps/web/src/store.ts | 94 ++++++-- .../src/threadDetailReducer.test.ts | 89 +++++++ .../client-runtime/src/threadDetailReducer.ts | 75 +++++- 18 files changed, 1234 insertions(+), 194 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 369eea0f7a0..d0b6987c3b8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1232,6 +1232,227 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("keeps the turn running across interim assistant messages until the session ends", () => + 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-turn-lifecycle"); + const turnId = TurnId.make("turn-lifecycle-1"); + + yield* eventStore.append({ + type: "thread.created", + eventId: EventId.make("evt-tl1"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-tl1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-tl1"), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-turn-lifecycle"), + title: "Turn lifecycle", + modelSelection: { + instanceId: ProviderInstanceId.make("claude"), + model: "claude-opus", + }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.session-set", + eventId: EventId.make("evt-tl2"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:00:01.000Z", + commandId: CommandId.make("cmd-tl2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-tl2"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "running", + providerName: "claude", + runtimeMode: "full-access", + activeTurnId: turnId, + lastError: null, + updatedAt: "2026-01-01T00:00:01.000Z", + }, + }, + }); + + // Interim assistant message completes mid-turn (commentary between + // tool calls) — the turn must stay running and unsettled. + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.make("evt-tl3"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:00:05.000Z", + commandId: CommandId.make("cmd-tl3"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-tl3"), + metadata: {}, + payload: { + threadId, + messageId: MessageId.make("message-tl-interim"), + role: "assistant", + text: "interim commentary", + turnId, + streaming: false, + createdAt: "2026-01-01T00:00:05.000Z", + updatedAt: "2026-01-01T00:00:05.000Z", + }, + }); + + yield* projectionPipeline.bootstrap; + + const runningRows = 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(runningRows, [{ state: "running", completedAt: null }]); + + // The session leaving "running" is the turn-end signal. + yield* eventStore.append({ + type: "thread.session-set", + eventId: EventId.make("evt-tl4"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:01:00.000Z", + commandId: CommandId.make("cmd-tl4"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-tl4"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "ready", + providerName: "claude", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: "2026-01-01T00:01:00.000Z", + }, + }, + }); + + yield* projectionPipeline.bootstrap; + + const settledRows = 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(settledRows, [ + { state: "completed", completedAt: "2026-01-01T00:01:00.000Z" }, + ]); + }), + ); + + it.effect("settles a superseded running turn when a new turn becomes active", () => + 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-turn-supersede"); + const oldTurnId = TurnId.make("turn-superseded"); + const newTurnId = TurnId.make("turn-steer"); + + yield* eventStore.append({ + type: "thread.created", + eventId: EventId.make("evt-ts1"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-ts1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-ts1"), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-turn-supersede"), + title: "Turn supersede", + modelSelection: { + instanceId: ProviderInstanceId.make("opencode"), + model: "big-pickle", + }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + const appendRunningSessionSet = (eventId: string, turnId: TurnId, updatedAt: string) => + eventStore.append({ + type: "thread.session-set", + eventId: EventId.make(eventId), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: updatedAt, + commandId: CommandId.make(`cmd-${eventId}`), + causationEventId: null, + correlationId: CorrelationId.make(`cmd-${eventId}`), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "running", + providerName: "opencode", + runtimeMode: "full-access", + activeTurnId: turnId, + lastError: null, + updatedAt, + }, + }, + }); + + yield* appendRunningSessionSet("evt-ts2", oldTurnId, "2026-01-01T00:00:01.000Z"); + // A steer: a new turn becomes active without the provider ever + // completing the previous one. + yield* appendRunningSessionSet("evt-ts3", newTurnId, "2026-01-01T00:00:30.000Z"); + + yield* projectionPipeline.bootstrap; + + const rows = yield* sql<{ + readonly turnId: string; + readonly state: string; + readonly completedAt: string | null; + }>` + SELECT turn_id AS "turnId", state, completed_at AS "completedAt" + FROM projection_turns + WHERE thread_id = ${threadId} + ORDER BY requested_at + `; + assert.deepEqual(rows, [ + { turnId: oldTurnId, state: "completed", completedAt: "2026-01-01T00:00:30.000Z" }, + { turnId: newTurnId, state: "running", completedAt: null }, + ]); + }), + ); + it.effect("keeps accumulated assistant text when completion payload text is empty", () => 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 4a48de19d39..c789d3a7437 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -2,6 +2,7 @@ import { ApprovalRequestId, type ChatAttachment, type OrchestrationEvent, + type OrchestrationSessionStatus, ThreadId, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -69,6 +70,29 @@ export const ORCHESTRATION_PROJECTOR_NAMES = { type ProjectorName = (typeof ORCHESTRATION_PROJECTOR_NAMES)[keyof typeof ORCHESTRATION_PROJECTOR_NAMES]; +/** + * Turn state to settle still-running turns with when their session leaves the + * "running" status, or null while the session is (re)starting or running and + * turns must stay unsettled. + */ +function settledTurnStateForSessionStatus( + status: OrchestrationSessionStatus, +): "completed" | "interrupted" | "error" | null { + switch (status) { + case "idle": + case "ready": + return "completed"; + case "error": + return "error"; + case "interrupted": + case "stopped": + return "interrupted"; + case "starting": + case "running": + return null; + } +} + interface ProjectorDefinition { readonly name: ProjectorName; readonly apply: ( @@ -999,9 +1023,60 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti case "thread.session-set": { const turnId = event.payload.session.activeTurnId; if (turnId === null || event.payload.session.status !== "running") { + // Leaving the "running" session status is the turn-end signal: + // settle still-running turns so their duration reflects the whole + // turn rather than the last assistant message. + const settledTurnState = settledTurnStateForSessionStatus( + event.payload.session.status, + ); + if (settledTurnState === null) { + return; + } + const existingTurns = yield* projectionTurnRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + yield* Effect.forEach( + existingTurns.filter((turn) => turn.turnId !== null && turn.state === "running"), + (turn) => + turn.turnId === null + ? Effect.void + : projectionTurnRepository.upsertByTurnId({ + ...turn, + turnId: turn.turnId, + state: settledTurnState, + // A running turn's completedAt can only hold a mid-turn + // placeholder checkpoint timestamp — the session leaving + // "running" is the authoritative turn end. + completedAt: event.payload.session.updatedAt, + }), + { concurrency: 1 }, + ); return; } + // A new active turn supersedes any still-running turn on the same + // thread — steering can open a new turn without the provider ever + // completing the previous one. + const otherRunningTurns = yield* projectionTurnRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + yield* Effect.forEach( + otherRunningTurns.filter( + (turn) => + turn.turnId !== null && turn.turnId !== turnId && turn.state === "running", + ), + (turn) => + turn.turnId === null + ? Effect.void + : projectionTurnRepository.upsertByTurnId({ + ...turn, + turnId: turn.turnId, + state: "completed", + completedAt: event.payload.session.updatedAt, + }), + { concurrency: 1 }, + ); + const existingTurn = yield* projectionTurnRepository.getByTurnId({ threadId: event.payload.threadId, turnId, @@ -1080,6 +1155,19 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti if (event.payload.turnId === null || event.payload.role !== "assistant") { return; } + // A completed assistant message only settles the turn once the + // session is no longer running it — providers may emit several + // assistant messages per turn (commentary between tool calls), and + // the turn must stay unsettled until the provider reports turn end + // (projected as thread.session-set leaving the "running" status). + const session = yield* projectionThreadSessionRepository.getByThreadId({ + threadId: event.payload.threadId, + }); + const turnStillRunning = + Option.isSome(session) && + session.value.status === "running" && + session.value.activeTurnId === event.payload.turnId; + const settlesTurn = !event.payload.streaming && !turnStillRunning; const existingTurn = yield* projectionTurnRepository.getByTurnId({ threadId: event.payload.threadId, turnId: event.payload.turnId, @@ -1088,16 +1176,16 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti yield* projectionTurnRepository.upsertByTurnId({ ...existingTurn.value, assistantMessageId: event.payload.messageId, - state: event.payload.streaming - ? existingTurn.value.state - : existingTurn.value.state === "interrupted" + state: settlesTurn + ? existingTurn.value.state === "interrupted" ? "interrupted" : existingTurn.value.state === "error" ? "error" - : "completed", - completedAt: event.payload.streaming - ? existingTurn.value.completedAt - : (existingTurn.value.completedAt ?? event.payload.updatedAt), + : "completed" + : existingTurn.value.state, + completedAt: settlesTurn + ? (existingTurn.value.completedAt ?? event.payload.updatedAt) + : existingTurn.value.completedAt, startedAt: existingTurn.value.startedAt ?? event.payload.createdAt, requestedAt: existingTurn.value.requestedAt ?? event.payload.createdAt, }); @@ -1110,10 +1198,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti sourceProposedPlanThreadId: null, sourceProposedPlanId: null, assistantMessageId: event.payload.messageId, - state: event.payload.streaming ? "running" : "completed", + state: settlesTurn ? "completed" : "running", requestedAt: event.payload.createdAt, startedAt: event.payload.createdAt, - completedAt: event.payload.streaming ? null : event.payload.updatedAt, + completedAt: settlesTurn ? event.payload.updatedAt : null, checkpointTurnCount: null, checkpointRef: null, checkpointStatus: null, @@ -1160,6 +1248,15 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } case "thread.turn-diff-completed": { + // Mid-turn diff updates produce placeholder checkpoints; record the + // checkpoint, but don't settle a turn its session is still running. + const session = yield* projectionThreadSessionRepository.getByThreadId({ + threadId: event.payload.threadId, + }); + const turnStillRunning = + Option.isSome(session) && + session.value.status === "running" && + session.value.activeTurnId === event.payload.turnId; const existingTurn = yield* projectionTurnRepository.getByTurnId({ threadId: event.payload.threadId, turnId: event.payload.turnId, @@ -1175,7 +1272,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti yield* projectionTurnRepository.upsertByTurnId({ ...existingTurn.value, assistantMessageId: event.payload.assistantMessageId, - state: nextState, + state: turnStillRunning ? existingTurn.value.state : nextState, checkpointTurnCount: event.payload.checkpointTurnCount, checkpointRef: event.payload.checkpointRef, checkpointStatus: event.payload.status, @@ -1193,7 +1290,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti sourceProposedPlanThreadId: null, sourceProposedPlanId: null, assistantMessageId: event.payload.assistantMessageId, - state: nextState, + state: turnStillRunning ? "running" : nextState, requestedAt: event.payload.completedAt, startedAt: event.payload.completedAt, completedAt: event.payload.completedAt, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index dceb5944864..64955590235 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1117,39 +1117,39 @@ describe("ProviderRuntimeIngestion", () => { const createdAt = "2026-01-01T00:00:00.000Z"; await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.create", - commandId: CommandId.make("cmd-thread-create-plan-source-guarded"), - threadId: sourceThreadId, - projectId: asProjectId("project-1"), - title: "Plan Source", - modelSelection: { - instanceId: ProviderInstanceId.make("codex"), - model: "gpt-5-codex", - }, - interactionMode: "plan", - runtimeMode: "approval-required", - branch: null, - worktreePath: null, - createdAt, - }), - ); - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.session.set", - commandId: CommandId.make("cmd-session-set-plan-source-guarded"), - threadId: sourceThreadId, - session: { + Effect.andThen( + harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-plan-source-guarded"), threadId: sourceThreadId, - status: "ready", - providerName: "codex", + projectId: asProjectId("project-1"), + title: "Plan Source", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: "plan", runtimeMode: "approval-required", - activeTurnId: null, - updatedAt: createdAt, - lastError: null, - }, - createdAt, - }), + branch: null, + worktreePath: null, + createdAt, + }), + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-plan-source-guarded"), + threadId: sourceThreadId, + session: { + threadId: sourceThreadId, + status: "ready", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + updatedAt: createdAt, + lastError: null, + }, + createdAt, + }), + ), ); harness.setProviderSession({ provider: ProviderDriverKind.make("codex"), @@ -1260,6 +1260,92 @@ describe("ProviderRuntimeIngestion", () => { expect(targetThreadAfterRejectedStart?.session?.activeTurnId).toBe(activeTurnId); }); + it("accepts a conflicting turn.started for a pending turn start when the provider expects that turn", async () => { + // Steering a running turn: the server requests a new turn while the old + // one is still active, and providers like opencode open the new turn + // without ever completing the superseded one. The new turn.started must + // replace the active turn instead of being rejected as stale. + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const oldTurnId = asTurnId("turn-steered-over"); + const newTurnId = asTurnId("turn-from-steer"); + const createdAt = "2026-01-01T00:00:00.000Z"; + + harness.setProviderSession({ + provider: ProviderDriverKind.make("codex"), + status: "running", + runtimeMode: "approval-required", + threadId, + createdAt, + updatedAt: createdAt, + activeTurnId: oldTurnId, + }); + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-steered-over"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId, + turnId: oldTurnId, + }); + await waitForThread( + harness.readModel, + (thread) => + thread.session?.status === "running" && thread.session?.activeTurnId === oldTurnId, + 2_000, + threadId, + ); + + // The steer: a user-requested turn start while the old turn still runs. + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-turn-start-steer"), + threadId, + message: { + messageId: asMessageId("msg-steer"), + role: "user", + text: "actually, do 15 instead", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + }), + ); + + // The provider session tracks the new turn before emitting turn.started + // (sendTurn updates the session first). + harness.setProviderSession({ + provider: ProviderDriverKind.make("codex"), + status: "running", + runtimeMode: "approval-required", + threadId, + createdAt, + updatedAt: createdAt, + activeTurnId: newTurnId, + }); + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-from-steer"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId, + turnId: newTurnId, + }); + + const threadAfterSteer = await waitForThread( + harness.readModel, + (thread) => + thread.session?.status === "running" && thread.session?.activeTurnId === newTurnId, + 2_000, + threadId, + ); + expect(threadAfterSteer.session?.activeTurnId).toBe(newTurnId); + expect(threadAfterSteer.latestTurn?.turnId).toBe(newTurnId); + expect(threadAfterSteer.latestTurn?.state).toBe("running"); + }); + it("does not mark the source proposed plan implemented for an unrelated turn.started when no thread active turn is tracked", async () => { const harness = await createHarness(); const sourceThreadId = asThreadId("thread-plan"); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index c5c155fd24a..3e5978f4846 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1226,6 +1226,22 @@ const make = Effect.gen(function* () { activeTurnId !== null && eventTurnId !== undefined && !sameId(activeTurnId, eventTurnId); const missingTurnForActiveTurn = activeTurnId !== null && eventTurnId === undefined; + // A turn.started that conflicts with the active turn is legitimate when + // the server itself has a turn start pending for this thread AND the + // provider session already tracks the event's turn as its active turn: + // steering a running turn makes some providers (e.g. opencode) open a + // new turn without ever completing the superseded one. A stale + // turn.started for some other turn id still gets rejected. + const conflictingTurnStartIsPendingTurnStart = + event.type === "turn.started" && conflictsWithActiveTurn + ? sameId(yield* getExpectedProviderTurnIdForThread(thread.id), eventTurnId) && + Option.isSome( + yield* projectionTurnRepository.getPendingTurnStartByThreadId({ + threadId: thread.id, + }), + ) + : false; + const shouldApplyThreadLifecycle = (() => { if (!STRICT_PROVIDER_LIFECYCLE_GUARD) { return true; @@ -1237,7 +1253,7 @@ const make = Effect.gen(function* () { case "thread.started": return true; case "turn.started": - return !conflictsWithActiveTurn; + return !conflictsWithActiveTurn || conflictingTurnStartIsPendingTurnStart; case "turn.completed": if (conflictsWithActiveTurn || missingTurnForActiveTurn) { return false; diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 804c1e8a6f8..fadd5078026 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -267,37 +267,76 @@ describe("orchestration projector", () => { ), ); - const afterRunning = await Effect.runPromise( - projectEvent( - afterCreate, - makeEvent({ - sequence: 2, - type: "thread.session-set", - aggregateKind: "thread", - aggregateId: "thread-1", - occurredAt: startedAt, - commandId: "cmd-running", - payload: { - threadId: "thread-1", - session: { + const settledAt = "2026-02-23T08:01:00.000Z"; + const [afterRunning, afterReady] = await Effect.runPromise( + Effect.flatMap( + projectEvent( + afterCreate, + makeEvent({ + sequence: 2, + type: "thread.session-set", + aggregateKind: "thread", + aggregateId: "thread-1", + occurredAt: startedAt, + commandId: "cmd-running", + payload: { threadId: "thread-1", - status: "running", - providerName: "codex", - providerSessionId: "session-1", - providerThreadId: "provider-thread-1", - runtimeMode: "approval-required", - activeTurnId: "turn-1", - lastError: null, - updatedAt: startedAt, + session: { + threadId: "thread-1", + status: "running", + providerName: "codex", + providerSessionId: "session-1", + providerThreadId: "provider-thread-1", + runtimeMode: "approval-required", + activeTurnId: "turn-1", + lastError: null, + updatedAt: startedAt, + }, }, - }, - }), + }), + ), + (running) => + Effect.map( + projectEvent( + running, + makeEvent({ + sequence: 3, + type: "thread.session-set", + aggregateKind: "thread", + aggregateId: "thread-1", + occurredAt: settledAt, + commandId: "cmd-ready", + payload: { + threadId: "thread-1", + session: { + threadId: "thread-1", + status: "ready", + providerName: "codex", + providerSessionId: "session-1", + providerThreadId: "provider-thread-1", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: settledAt, + }, + }, + }), + ), + (ready) => [running, ready] as const, + ), ), ); const thread = afterRunning.threads[0]; expect(thread?.latestTurn?.turnId).toBe("turn-1"); expect(thread?.session?.status).toBe("running"); + + // Leaving the "running" session status settles the running turn with the + // session timestamp as the turn end. + const settledThread = afterReady.threads[0]; + expect(settledThread?.latestTurn?.turnId).toBe("turn-1"); + expect(settledThread?.latestTurn?.state).toBe("completed"); + expect(settledThread?.latestTurn?.completedAt).toBe(settledAt); }); it("updates canonical thread runtime mode from thread.runtime-mode-set", async () => { diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 0c92f965433..fc6ab8f6fcf 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -38,6 +38,29 @@ function checkpointStatusToLatestTurnState(status: "ready" | "missing" | "error" return "completed" as const; } +/** + * Turn state to settle a still-running latest turn with when its session + * leaves the "running" status, or null while the session is (re)starting or + * running and the turn must stay unsettled. + */ +function settledTurnStateForSessionStatus( + status: OrchestrationSession["status"], +): "completed" | "interrupted" | "error" | null { + switch (status) { + case "idle": + case "ready": + return "completed"; + case "error": + return "error"; + case "interrupted": + case "stopped": + return "interrupted"; + case "starting": + case "running": + return null; + } +} + function updateThread( threads: ReadonlyArray, threadId: ThreadId, @@ -438,6 +461,9 @@ export function projectEvent( "session", ); + // Leaving the "running" session status is the turn-end signal: settle + // a still-running latest turn so its duration reflects the whole turn. + const settledTurnState = settledTurnStateForSessionStatus(session.status); return { ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { @@ -461,7 +487,18 @@ export function projectEvent( ? thread.latestTurn.assistantMessageId : null, } - : thread.latestTurn, + : thread.latestTurn !== null && + thread.latestTurn.state === "running" && + settledTurnState !== null + ? { + ...thread.latestTurn, + state: settledTurnState, + // A running turn's completedAt can only hold a mid-turn + // placeholder checkpoint timestamp — the session leaving + // "running" is the authoritative turn end. + completedAt: session.updatedAt, + } + : thread.latestTurn, updatedAt: event.occurredAt, }), }; @@ -544,24 +581,31 @@ export function projectEvent( .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount) .slice(-MAX_THREAD_CHECKPOINTS); + // Mid-turn diff updates produce placeholder checkpoints; record the + // checkpoint, but don't settle a turn its session is still running. + const turnStillRunning = + thread.session?.status === "running" && thread.session.activeTurnId === payload.turnId; + return { ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { checkpoints, - latestTurn: { - turnId: payload.turnId, - state: checkpointStatusToLatestTurnState(payload.status), - requestedAt: - thread.latestTurn?.turnId === payload.turnId - ? thread.latestTurn.requestedAt - : payload.completedAt, - startedAt: - thread.latestTurn?.turnId === payload.turnId - ? (thread.latestTurn.startedAt ?? payload.completedAt) - : payload.completedAt, - completedAt: payload.completedAt, - assistantMessageId: payload.assistantMessageId, - }, + latestTurn: turnStillRunning + ? thread.latestTurn + : { + turnId: payload.turnId, + state: checkpointStatusToLatestTurnState(payload.status), + requestedAt: + thread.latestTurn?.turnId === payload.turnId + ? thread.latestTurn.requestedAt + : payload.completedAt, + startedAt: + thread.latestTurn?.turnId === payload.turnId + ? (thread.latestTurn.startedAt ?? payload.completedAt) + : payload.completedAt, + completedAt: payload.completedAt, + assistantMessageId: payload.assistantMessageId, + }, updatedAt: event.occurredAt, }), }; diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index dfaff75b8c1..502ede7d40a 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -890,6 +890,75 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const runtimeEventsFiber = yield* Stream.takeUntil( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runCollect, Effect.forkChild); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "run 5 commands", + attachments: [], + }); + + // Steer: a second sendTurn while the turn is still running continues + // the same turn — the message is queued into the live agent loop. + const steeredTurn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "actually run 15", + attachments: [], + }); + assert.equal(String(steeredTurn.turnId), String(turn.turnId)); + + harness.query.emit({ + type: "assistant", + session_id: "sdk-session-steer", + uuid: "assistant-steer-1", + parent_tool_use_id: null, + message: { + id: "assistant-message-steer-1", + content: [{ type: "text", text: "Adjusting to 15." }], + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-steer", + uuid: "result-steer-1", + } as unknown as SDKMessage); + + 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 whole run: the steer produced no + // turn.completed/turn.started pair. + assert.equal(turnStartedEvents.length, 1); + assert.equal(String(turnStartedEvents[0]?.turnId), String(turn.turnId)); + assert.equal(turnCompletedEvents.length, 1); + assert.equal(String(turnCompletedEvents[0]?.turnId), String(turn.turnId)); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("maps Claude reasoning deltas, streamed tool inputs, and tool results", () => { 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 3970b20fe23..38b77c69262 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -122,6 +122,13 @@ interface ClaudeResumeState { interface ClaudeTurnState { readonly turnId: TurnId; readonly startedAt: string; + /** + * True for turns auto-started by assistant output arriving without an + * active turn (background agent/subagent responses between user prompts). + * Synthetic turns are auto-closed by the next sendTurn; real turns are + * steered instead (the queued message continues the same turn). + */ + readonly synthetic?: boolean; readonly items: Array; readonly assistantTextBlocks: Map; readonly assistantTextBlockOrder: Array; @@ -2486,6 +2493,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( context.turnState = { turnId, startedAt, + synthetic: true, items: [], assistantTextBlocks: new Map(), assistantTextBlockOrder: [], @@ -3629,9 +3637,14 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ? input.modelSelection : undefined; - if (context.turnState) { - // Auto-close a stale synthetic turn (from background agent responses - // between user prompts) to prevent blocking the user's next turn. + // A sendTurn while a real turn is running is a steer: the message is + // queued into the live SDK agent loop and the work continues as the same + // turn — no synthetic turn boundary. Stale synthetic turns (from + // background agent responses between user prompts) are auto-closed + // instead, so they don't block the user's next turn. + const steeringTurnState = + context.turnState && context.turnState.synthetic !== true ? context.turnState : null; + if (context.turnState && steeringTurnState === null) { yield* completeTurn(context, "completed"); } @@ -3666,37 +3679,39 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); } - const turnId = TurnId.make(yield* randomUUIDv4); - const turnState: ClaudeTurnState = { - turnId, - startedAt: yield* nowIso, - items: [], - assistantTextBlocks: new Map(), - assistantTextBlockOrder: [], - capturedProposedPlanKeys: new Set(), - nextSyntheticAssistantBlockIndex: -1, - }; + const turnId = steeringTurnState?.turnId ?? TurnId.make(yield* randomUUIDv4); + if (steeringTurnState === null) { + const turnState: ClaudeTurnState = { + turnId, + startedAt: yield* nowIso, + items: [], + assistantTextBlocks: new Map(), + assistantTextBlockOrder: [], + capturedProposedPlanKeys: new Set(), + nextSyntheticAssistantBlockIndex: -1, + }; - const updatedAt = yield* nowIso; - context.turnState = turnState; - context.session = { - ...context.session, - status: "running", - activeTurnId: turnId, - updatedAt, - }; + const updatedAt = yield* nowIso; + context.turnState = turnState; + context.session = { + ...context.session, + status: "running", + activeTurnId: turnId, + updatedAt, + }; - const turnStartedStamp = yield* makeEventStamp(); - yield* offerRuntimeEvent({ - type: "turn.started", - eventId: turnStartedStamp.eventId, - provider: PROVIDER, - createdAt: turnStartedStamp.createdAt, - threadId: context.session.threadId, - turnId, - payload: modelSelection?.model ? { model: modelSelection.model } : {}, - providerRefs: {}, - }); + const turnStartedStamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "turn.started", + eventId: turnStartedStamp.eventId, + provider: PROVIDER, + createdAt: turnStartedStamp.createdAt, + threadId: context.session.threadId, + turnId, + payload: modelSelection?.model ? { model: modelSelection.model } : {}, + providerRefs: {}, + }); + } const message = yield* buildUserMessageEffect(input, { fileSystem, diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 5c0badc2de6..3f483d8fd7e 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -395,6 +395,85 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-steer"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId, + input: "run 5 commands", + modelSelection: { + instanceId: ProviderInstanceId.make("opencode"), + model: "openai/gpt-5", + }, + }); + + // Steer: OpenCode queues the prompt into the busy session, so the + // active turn id is reused instead of opening a new turn. + const steeredTurn = yield* adapter.sendTurn({ + threadId, + input: "actually run 15", + modelSelection: { + instanceId: ProviderInstanceId.make("opencode"), + model: "openai/gpt-5", + }, + }); + assert.equal(String(steeredTurn.turnId), String(turn.turnId)); + + const sessions = yield* adapter.listSessions(); + const session = sessions.find((entry) => entry.threadId === threadId); + assert.equal(session?.status, "running"); + assert.equal(String(session?.activeTurnId), String(turn.turnId)); + assert.equal(runtimeMock.state.promptCalls.length, 2); + }), + ); + + it.effect("keeps the running turn when a steer prompt fails", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-steer-failure"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId, + input: "run 5 commands", + modelSelection: { + instanceId: ProviderInstanceId.make("opencode"), + model: "openai/gpt-5", + }, + }); + + runtimeMock.state.promptAsyncError = new Error("steer failed"); + const error = yield* adapter + .sendTurn({ + threadId, + input: "actually run 15", + modelSelection: { + instanceId: ProviderInstanceId.make("opencode"), + model: "openai/gpt-5", + }, + }) + .pipe(Effect.flip); + + // The original turn keeps running — only the steer prompt failed. + assert.equal(error._tag, "ProviderAdapterRequestError"); + const sessions = yield* adapter.listSessions(); + const session = sessions.find((entry) => entry.threadId === threadId); + assert.equal(session?.status, "running"); + assert.equal(String(session?.activeTurnId), String(turn.turnId)); + }), + ); + it.effect("passes agent and variant options for the adapter's bound custom instance id", () => { const instanceId = ProviderInstanceId.make("opencode_zen"); const adapterLayer = Layer.effect( diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 34a67199125..54444ce586d 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -1151,7 +1151,11 @@ export function makeOpenCodeAdapter( const sendTurn: OpenCodeAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) { const context = ensureSessionContext(sessions, input.threadId); - const turnId = TurnId.make(`opencode-turn-${yield* randomUUIDv4}`); + // A sendTurn while a turn is active is a steer: OpenCode queues the + // prompt into the busy session and the work continues as one turn, so + // the active turn id is reused instead of opening a new turn. + const steeringTurnId = context.activeTurnId; + const turnId = steeringTurnId ?? TurnId.make(`opencode-turn-${yield* randomUUIDv4}`); const modelSelection = input.modelSelection ?? (context.session.model @@ -1206,14 +1210,16 @@ export function makeOpenCodeAdapter( { clearLastError: true }, ); - yield* emit({ - ...(yield* buildEventBase({ threadId: input.threadId, turnId })), - type: "turn.started", - payload: { - model: modelSelection?.model ?? context.session.model, - ...(variant ? { effort: variant } : {}), - }, - }); + if (steeringTurnId === undefined) { + yield* emit({ + ...(yield* buildEventBase({ threadId: input.threadId, turnId })), + type: "turn.started", + payload: { + model: modelSelection?.model ?? context.session.model, + ...(variant ? { effort: variant } : {}), + }, + }); + } yield* runOpenCodeSdk("session.promptAsync", () => context.client.session.promptAsync({ @@ -1225,35 +1231,38 @@ export function makeOpenCodeAdapter( }), ).pipe( Effect.mapError(toRequestError), - // On failure: clear active-turn state, flip the session back to ready - // with lastError set, emit turn.aborted, then let the typed error - // propagate. We don't need to rebuild the error here — `toRequestError` - // already produced the right shape. + // On failure of a fresh turn: clear active-turn state, flip the + // session back to ready with lastError set, emit turn.aborted, then + // let the typed error propagate. We don't need to rebuild the error + // here — `toRequestError` already produced the right shape. A failed + // steer leaves the still-running original turn untouched. Effect.tapError((requestError) => - Effect.gen(function* () { - context.activeTurnId = undefined; - context.activeAgent = undefined; - context.activeVariant = undefined; - yield* updateProviderSession( - context, - { - status: "ready", - model: modelSelection?.model ?? context.session.model, - lastError: requestError.detail, - }, - { clearActiveTurnId: true }, - ); - yield* emit({ - ...(yield* buildEventBase({ - threadId: input.threadId, - turnId, - })), - type: "turn.aborted", - payload: { - reason: requestError.detail, - }, - }); - }), + steeringTurnId !== undefined + ? Effect.void + : Effect.gen(function* () { + context.activeTurnId = undefined; + context.activeAgent = undefined; + context.activeVariant = undefined; + yield* updateProviderSession( + context, + { + status: "ready", + model: modelSelection?.model ?? context.session.model, + lastError: requestError.detail, + }, + { clearActiveTurnId: true }, + ); + yield* emit({ + ...(yield* buildEventBase({ + threadId: input.threadId, + turnId, + })), + type: "turn.aborted", + payload: { + reason: requestError.detail, + }, + }); + }), ), ); diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index c2e21cbd803..032f8635698 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -451,8 +451,8 @@ describe("deriveMessagesTimelineRows", () => { ); expect(foldRow?.turnId).toBe("turn-1"); expect(foldRow?.expanded).toBe(false); - // First fold entry (00:00:05) → terminal message completedAt (00:00:22). - expect(foldRow?.label).toBe("Worked for 17s"); + // User message boundary (00:00:00) → terminal message completedAt (00:00:22). + expect(foldRow?.label).toBe("Worked for 22s"); expect(collapsedRows.map((row) => row.id)).toEqual([ "user-entry", "turn-fold:turn-1", @@ -480,6 +480,100 @@ describe("deriveMessagesTimelineRows", () => { ).toBeDefined(); }); + it("derives a sane duration for a steer-superseded turn with one instant commentary message", () => { + // A steer ends the previous turn early: its only message completes the + // instant it is created, and trailing work entries land after it. The + // fold duration must span from the user message that started the turn to + // the last entry, not message createdAt → message completedAt (~0ms). + const rows = deriveMessagesTimelineRows({ + timelineEntries: [ + { + id: "user-entry", + kind: "message", + createdAt: "2026-01-01T00:00:00Z", + message: { + id: "user-1" as never, + role: "user" as const, + text: "do it once more", + turnId: null, + createdAt: "2026-01-01T00:00:00Z", + streaming: false, + }, + }, + { + id: "assistant-commentary-entry", + kind: "message", + createdAt: "2026-01-01T00:00:09Z", + message: { + id: "assistant-commentary" as never, + role: "assistant" as const, + text: "Kicking off call 1.", + turnId: "turn-1" as never, + createdAt: "2026-01-01T00:00:09Z", + completedAt: "2026-01-01T00:00:09Z", + streaming: false, + }, + }, + { + id: "work-entry-1", + kind: "work", + createdAt: "2026-01-01T00:00:12Z", + entry: { + id: "work-1", + createdAt: "2026-01-01T00:00:12Z", + turnId: "turn-1" as never, + label: "Ran command", + tone: "tool" as const, + }, + }, + { + id: "steer-user-entry", + kind: "message", + createdAt: "2026-01-01T00:00:14Z", + message: { + id: "user-2" as never, + role: "user" as const, + text: "actually do 15", + turnId: null, + createdAt: "2026-01-01T00:00:14Z", + streaming: false, + }, + }, + { + id: "assistant-next-turn-entry", + kind: "message", + createdAt: "2026-01-01T00:00:17Z", + message: { + id: "assistant-next" as never, + role: "assistant" as const, + text: "One down — adjusting.", + turnId: "turn-2" as never, + createdAt: "2026-01-01T00:00:17Z", + streaming: true, + }, + }, + ], + latestTurn: { + turnId: "turn-2" as never, + state: "running", + startedAt: "2026-01-01T00:00:14Z", + completedAt: null, + }, + isWorking: true, + activeTurnStartedAt: "2026-01-01T00:00:14Z", + turnDiffSummaryByAssistantMessageId: new Map(), + revertTurnCountByUserMessageId: new Map(), + }); + + const foldRow = rows.find( + (row): row is Extract<(typeof rows)[number], { kind: "turn-fold" }> => + row.kind === "turn-fold", + ); + // User message (00:00:00) → trailing work entry (00:00:12). + expect(foldRow?.turnId).toBe("turn-1"); + expect(foldRow?.label).toBe("Worked for 12s"); + }); + it("uses latest-turn timings and the stopped label for an interrupted latest turn", () => { const rows = deriveMessagesTimelineRows({ timelineEntries: [ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 411f1d856cc..416b37e4f51 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -12,6 +12,16 @@ function computeElapsedMs(startIso: string, endIso: string): number | null { return Math.max(0, end - start); } +function maxIsoTimestamp(a: string | null, b: string | null): string | null { + if (a === null) return b; + if (b === null) return a; + const aMs = Date.parse(a); + const bMs = Date.parse(b); + if (!Number.isFinite(aMs)) return b; + if (!Number.isFinite(bMs)) return a; + return bMs > aMs ? b : a; +} + export interface TimelineDurationMessage { id: string; role: "user" | "assistant" | "system"; @@ -167,10 +177,22 @@ function deriveTurnFolds(input: { entries: Array; terminalEntry: Extract | null; hasStreamingMessage: boolean; + /** + * The user message that kicked the turn off. Entry timestamps alone + * undercount the duration (the first entry appears only once the + * provider starts producing output), and a turn cut short by a steer may + * hold a single instantaneous commentary message. + */ + startBoundary: string | null; } const groupsByTurnId = new Map(); + let pendingUserBoundary: string | null = null; for (const entry of input.timelineEntries) { + if (entry.kind === "message" && entry.message.role === "user") { + pendingUserBoundary = entry.message.createdAt; + continue; + } const turnId = entry.kind === "message" && entry.message.role === "assistant" ? (entry.message.turnId ?? null) @@ -180,11 +202,20 @@ function deriveTurnFolds(input: { if (!turnId) { continue; } - const group = groupsByTurnId.get(turnId) ?? { - entries: [], - terminalEntry: null, - hasStreamingMessage: false, - }; + let group = groupsByTurnId.get(turnId); + if (!group) { + group = { + entries: [], + terminalEntry: null, + hasStreamingMessage: false, + // Each user boundary starts at most one turn; a second turn after the + // same user message (e.g. a steer-superseded continuation) falls back + // to its own first entry. + startBoundary: pendingUserBoundary, + }; + pendingUserBoundary = null; + groupsByTurnId.set(turnId, group); + } group.entries.push(entry); if (entry.kind === "message") { if (input.terminalAssistantMessageIds.has(entry.message.id)) { @@ -194,7 +225,6 @@ function deriveTurnFolds(input: { group.hasStreamingMessage = true; } } - groupsByTurnId.set(turnId, group); } const foldsByAnchorEntryId = new Map(); @@ -223,17 +253,21 @@ function deriveTurnFolds(input: { const isLatestInterruptedTurn = input.latestTurn?.turnId === turnId && input.latestTurn.state === "interrupted"; + // A turn cut short by a steer leaves trailing work entries behind its + // terminal message — take whichever ended last. + const lastEntryEnd = + lastEntry.kind === "message" + ? (lastEntry.message.completedAt ?? lastEntry.createdAt) + : lastEntry.createdAt; const elapsedMs = input.latestTurn?.turnId === turnId && input.latestTurn.startedAt && input.latestTurn.completedAt ? computeElapsedMs(input.latestTurn.startedAt, input.latestTurn.completedAt) : computeElapsedMs( - firstEntry.createdAt, - group.terminalEntry?.message.completedAt ?? - (lastEntry.kind === "message" - ? (lastEntry.message.completedAt ?? lastEntry.createdAt) - : lastEntry.createdAt), + group.startBoundary ?? firstEntry.createdAt, + maxIsoTimestamp(group.terminalEntry?.message.completedAt ?? null, lastEntryEnd) ?? + lastEntryEnd, ); const duration = elapsedMs !== null ? formatDuration(elapsedMs) : null; const label = isLatestInterruptedTurn diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 3a129fbfbe1..7e4a402d910 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -368,7 +368,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ maintainScrollAtEndThreshold={0.1} maintainVisibleContentPosition onScroll={handleScroll} - className="h-full overflow-x-hidden overscroll-y-contain px-3 sm:px-5" + className="scrollbar-gutter-both h-full overflow-x-hidden overscroll-y-contain px-3 sm:px-5" ListHeaderComponent={TIMELINE_LIST_HEADER} ListFooterComponent={TIMELINE_LIST_FOOTER} /> diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 676ecdc96dd..63ab93c48ee 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -264,7 +264,11 @@ export function workEntryIndicatesToolNeutralStatus(entry: WorkLogEntry): boolea export function formatDuration(durationMs: number): string { if (!Number.isFinite(durationMs) || durationMs < 0) return "0ms"; if (durationMs < 1_000) return `${Math.max(1, Math.round(durationMs))}ms`; - if (durationMs < 10_000) return `${(durationMs / 1_000).toFixed(1)}s`; + if (durationMs < 10_000) { + const tenths = Math.round(durationMs / 100) / 10; + // 9.95s+ rounds up to the next bucket — render "10s", not "10.0s". + return tenths >= 10 ? "10s" : `${tenths.toFixed(1)}s`; + } if (durationMs < 60_000) return `${Math.round(durationMs / 1_000)}s`; const minutes = Math.floor(durationMs / 60_000); const seconds = Math.round((durationMs % 60_000) / 1_000); diff --git a/apps/web/src/store.test.ts b/apps/web/src/store.test.ts index cad78ab9d35..2fe06d518d2 100644 --- a/apps/web/src/store.test.ts +++ b/apps/web/src/store.test.ts @@ -776,9 +776,40 @@ describe("incremental orchestration updates", () => { localEnvironmentId, ); + // A completed assistant message must not settle the turn while the + // session is still running it — providers emit interim assistant + // messages between tool calls. expect(threadsOf(next)[0]?.session?.status).toBe("running"); - expect(threadsOf(next)[0]?.latestTurn?.state).toBe("completed"); + expect(threadsOf(next)[0]?.latestTurn?.state).toBe("running"); + expect(threadsOf(next)[0]?.latestTurn?.completedAt).toBeNull(); expect(threadsOf(next)[0]?.messages).toHaveLength(1); + + const settled = applyOrchestrationEvents( + next, + [ + makeEvent( + "thread.session-set", + { + threadId: thread.id, + session: { + threadId: thread.id, + status: "ready", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: "2026-02-27T00:00:04.000Z", + }, + }, + { sequence: 4 }, + ), + ], + localEnvironmentId, + ); + + // Leaving the running session status is the turn-end signal. + expect(threadsOf(settled)[0]?.latestTurn?.state).toBe("completed"); + expect(threadsOf(settled)[0]?.latestTurn?.completedAt).toBe("2026-02-27T00:00:04.000Z"); }); it("does not regress latestTurn when an older turn diff completes late", () => { diff --git a/apps/web/src/store.ts b/apps/web/src/store.ts index 7d995b5ea75..12b621e3900 100644 --- a/apps/web/src/store.ts +++ b/apps/web/src/store.ts @@ -885,6 +885,29 @@ function buildLatestTurn(params: { }; } +/** + * Turn state to settle a still-running latest turn with when its session + * leaves the "running" status, or null while the session is (re)starting or + * running and the turn must stay unsettled. + */ +function settledTurnStateForSessionStatus( + status: OrchestrationSessionStatus, +): "completed" | "interrupted" | "error" | null { + switch (status) { + case "idle": + case "ready": + return "completed"; + case "error": + return "error"; + case "interrupted": + case "stopped": + return "interrupted"; + case "starting": + case "running": + return null; + } +} + function rebindTurnDiffSummariesForAssistantMessage( turnDiffSummaries: ReadonlyArray, turnId: TurnId, @@ -1419,6 +1442,15 @@ function applyEnvironmentOrchestrationEvent( event.payload.messageId, ) : thread.turnDiffSummaries; + // A completed assistant message only settles the turn once the + // session is no longer running it — providers may emit several + // assistant messages per turn (commentary between tool calls), and + // the turn must stay unsettled until the provider reports turn end. + const turnStillRunning = + event.payload.turnId !== null && + thread.session?.orchestrationStatus === "running" && + thread.session.activeTurnId === event.payload.turnId; + const settlesTurn = !event.payload.streaming && !turnStillRunning; const latestTurn: Thread["latestTurn"] = event.payload.role === "assistant" && event.payload.turnId !== null && @@ -1426,13 +1458,13 @@ function applyEnvironmentOrchestrationEvent( ? buildLatestTurn({ previous: thread.latestTurn, turnId: event.payload.turnId, - state: event.payload.streaming - ? "running" - : thread.latestTurn?.state === "interrupted" + state: settlesTurn + ? thread.latestTurn?.state === "interrupted" ? "interrupted" : thread.latestTurn?.state === "error" ? "error" - : "completed", + : "completed" + : "running", requestedAt: thread.latestTurn?.turnId === event.payload.turnId ? thread.latestTurn.requestedAt @@ -1442,11 +1474,11 @@ function applyEnvironmentOrchestrationEvent( ? (thread.latestTurn.startedAt ?? event.payload.createdAt) : event.payload.createdAt, sourceProposedPlan: thread.pendingSourceProposedPlan, - completedAt: event.payload.streaming - ? thread.latestTurn?.turnId === event.payload.turnId + completedAt: settlesTurn + ? event.payload.updatedAt + : thread.latestTurn?.turnId === event.payload.turnId ? (thread.latestTurn.completedAt ?? null) - : null - : event.payload.updatedAt, + : null, assistantMessageId: event.payload.messageId, }) : thread.latestTurn; @@ -1460,11 +1492,12 @@ function applyEnvironmentOrchestrationEvent( }); case "thread.session-set": - return updateThreadState(state, event.payload.threadId, (thread) => ({ - ...thread, - session: mapSession(event.payload.session), - error: sanitizeThreadErrorMessage(event.payload.session.lastError), - latestTurn: + return updateThreadState(state, event.payload.threadId, (thread) => { + // Leaving the "running" session status is the turn-end signal: + // settle a still-running latest turn so its duration reflects the + // whole turn, not the last assistant message. + const settledTurnState = settledTurnStateForSessionStatus(event.payload.session.status); + const latestTurn: Thread["latestTurn"] = event.payload.session.status === "running" && event.payload.session.activeTurnId !== null ? buildLatestTurn({ previous: thread.latestTurn, @@ -1485,9 +1518,30 @@ function applyEnvironmentOrchestrationEvent( : null, sourceProposedPlan: thread.pendingSourceProposedPlan, }) - : thread.latestTurn, - updatedAt: event.occurredAt, - })); + : thread.latestTurn !== null && + thread.latestTurn.state === "running" && + settledTurnState !== null + ? buildLatestTurn({ + previous: thread.latestTurn, + turnId: thread.latestTurn.turnId, + state: settledTurnState, + requestedAt: thread.latestTurn.requestedAt, + startedAt: thread.latestTurn.startedAt, + // A running turn's completedAt can only hold a mid-turn + // placeholder checkpoint timestamp — the session leaving + // "running" is the authoritative turn end. + completedAt: event.payload.session.updatedAt, + assistantMessageId: thread.latestTurn.assistantMessageId, + }) + : thread.latestTurn; + return { + ...thread, + session: mapSession(event.payload.session), + error: sanitizeThreadErrorMessage(event.payload.session.lastError), + latestTurn, + updatedAt: event.occurredAt, + }; + }); case "thread.session-stop-requested": return updateThreadState(state, event.payload.threadId, (thread) => @@ -1552,8 +1606,14 @@ function applyEnvironmentOrchestrationEvent( (right.checkpointTurnCount ?? Number.MAX_SAFE_INTEGER), ) .slice(-MAX_THREAD_CHECKPOINTS); + // Mid-turn diff updates produce placeholder checkpoints; record the + // diff summary, but don't settle a turn its session is still running. + const turnStillRunning = + thread.session?.orchestrationStatus === "running" && + thread.session.activeTurnId === event.payload.turnId; const latestTurn = - thread.latestTurn === null || thread.latestTurn.turnId === event.payload.turnId + !turnStillRunning && + (thread.latestTurn === null || thread.latestTurn.turnId === event.payload.turnId) ? buildLatestTurn({ previous: thread.latestTurn, turnId: event.payload.turnId, diff --git a/packages/client-runtime/src/threadDetailReducer.test.ts b/packages/client-runtime/src/threadDetailReducer.test.ts index 56a2d64dcb1..f2af7284083 100644 --- a/packages/client-runtime/src/threadDetailReducer.test.ts +++ b/packages/client-runtime/src/threadDetailReducer.test.ts @@ -288,9 +288,98 @@ describe("applyThreadDetailEvent", () => { expect(result.thread.latestTurn?.assistantMessageId).toBe("msg-3"); } }); + + it("keeps latestTurn running for interim assistant messages while the session runs the turn", () => { + const threadWithRunningSession: OrchestrationThread = { + ...baseThread, + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "claude", + runtimeMode: "full-access", + activeTurnId: TurnId.make("turn-1"), + lastError: null, + updatedAt: "2026-04-01T06:59:00.000Z", + }, + latestTurn: { + turnId: TurnId.make("turn-1"), + state: "running", + requestedAt: "2026-04-01T06:59:00.000Z", + startedAt: "2026-04-01T06:59:00.000Z", + completedAt: null, + assistantMessageId: null, + }, + }; + + const result = applyThreadDetailEvent(threadWithRunningSession, { + ...baseEventFields, + sequence: 8, + occurredAt: "2026-04-01T07:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.message-sent", + payload: { + threadId: ThreadId.make("thread-1"), + messageId: MessageId.make("msg-3"), + role: "assistant", + text: "Interim commentary between tool calls.", + turnId: TurnId.make("turn-1"), + streaming: false, + createdAt: "2026-04-01T07:00:00.000Z", + updatedAt: "2026-04-01T07:00:00.000Z", + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.latestTurn?.state).toBe("running"); + expect(result.thread.latestTurn?.completedAt).toBeNull(); + } + }); }); describe("thread.session-set", () => { + it("settles a running latestTurn when the session leaves the running status", () => { + 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"), + }, + }; + + const result = applyThreadDetailEvent(threadWithRunningTurn, { + ...baseEventFields, + sequence: 9, + occurredAt: "2026-04-01T08:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.session-set", + payload: { + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "ready", + providerName: "claude", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: "2026-04-01T08:00:00.000Z", + }, + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.latestTurn?.state).toBe("completed"); + expect(result.thread.latestTurn?.completedAt).toBe("2026-04-01T08:00:00.000Z"); + } + }); + it("updates session and latestTurn for a running session", () => { const result = applyThreadDetailEvent(baseThread, { ...baseEventFields, diff --git a/packages/client-runtime/src/threadDetailReducer.ts b/packages/client-runtime/src/threadDetailReducer.ts index 125f39428a6..53bad5785b9 100644 --- a/packages/client-runtime/src/threadDetailReducer.ts +++ b/packages/client-runtime/src/threadDetailReducer.ts @@ -7,6 +7,7 @@ import type { OrchestrationEvent, OrchestrationLatestTurn, OrchestrationMessage, + OrchestrationSession, OrchestrationThread, OrchestrationThreadActivity, TurnId, @@ -232,20 +233,29 @@ export function applyThreadDetailEvent( : Arr.append(thread.messages, message); const cappedMessages = Arr.takeRight(messages, limits.maxMessages); - // Update latestTurn for assistant messages bound to a turn. + // Update latestTurn for assistant messages bound to a turn. A completed + // assistant message only settles the turn once the session is no longer + // running it — providers may emit several assistant messages per turn + // (commentary between tool calls), and the turn must stay unsettled + // until the provider reports turn end. + const turnStillRunning = + event.payload.turnId !== null && + thread.session?.status === "running" && + thread.session.activeTurnId === event.payload.turnId; + const settlesTurn = !event.payload.streaming && !turnStillRunning; const latestTurn: OrchestrationThread["latestTurn"] = event.payload.role === "assistant" && event.payload.turnId !== null && (thread.latestTurn === null || thread.latestTurn.turnId === event.payload.turnId) ? { turnId: event.payload.turnId, - state: event.payload.streaming - ? "running" - : thread.latestTurn?.state === "interrupted" + state: settlesTurn + ? thread.latestTurn?.state === "interrupted" ? "interrupted" : thread.latestTurn?.state === "error" ? "error" - : "completed", + : "completed" + : "running", requestedAt: thread.latestTurn?.turnId === event.payload.turnId ? thread.latestTurn.requestedAt @@ -254,11 +264,11 @@ export function applyThreadDetailEvent( thread.latestTurn?.turnId === event.payload.turnId ? (thread.latestTurn.startedAt ?? event.payload.createdAt) : event.payload.createdAt, - completedAt: event.payload.streaming - ? thread.latestTurn?.turnId === event.payload.turnId + completedAt: settlesTurn + ? event.payload.updatedAt + : thread.latestTurn?.turnId === event.payload.turnId ? (thread.latestTurn.completedAt ?? null) - : null - : event.payload.updatedAt, + : null, assistantMessageId: event.payload.messageId, } : thread.latestTurn; @@ -287,6 +297,9 @@ export function applyThreadDetailEvent( // ── Session ───────────────────────────────────────────────────── case "thread.session-set": { + // Leaving the "running" session status is the turn-end signal: settle a + // still-running latest turn so its duration reflects the whole turn. + const settledTurnState = settledTurnStateForSessionStatus(event.payload.session.status); const latestTurn: OrchestrationLatestTurn | null = event.payload.session.status === "running" && event.payload.session.activeTurnId !== null ? { @@ -306,7 +319,18 @@ export function applyThreadDetailEvent( ? thread.latestTurn.assistantMessageId : null, } - : thread.latestTurn; + : thread.latestTurn !== null && + thread.latestTurn.state === "running" && + settledTurnState !== null + ? { + ...thread.latestTurn, + state: settledTurnState, + // A running turn's completedAt can only hold a mid-turn + // placeholder checkpoint timestamp — the session leaving + // "running" is the authoritative turn end. + completedAt: event.payload.session.updatedAt, + } + : thread.latestTurn; return { kind: "updated", @@ -380,8 +404,14 @@ export function applyThreadDetailEvent( Arr.takeRight(limits.maxCheckpoints), ); + // Mid-turn diff updates produce placeholder checkpoints; record the + // checkpoint, but don't settle a turn its session is still running. + const diffTurnStillRunning = + thread.session?.status === "running" && + thread.session.activeTurnId === event.payload.turnId; const latestTurn = - thread.latestTurn === null || thread.latestTurn.turnId === event.payload.turnId + !diffTurnStillRunning && + (thread.latestTurn === null || thread.latestTurn.turnId === event.payload.turnId) ? { turnId: event.payload.turnId, state: checkpointStatusToTurnState(event.payload.status), @@ -482,6 +512,29 @@ export function applyThreadDetailEvent( // ── Helpers ────────────────────────────────────────────────────────── +/** + * Turn state to settle a still-running latest turn with when its session + * leaves the "running" status, or null while the session is (re)starting or + * running and the turn must stay unsettled. + */ +function settledTurnStateForSessionStatus( + status: OrchestrationSession["status"], +): "completed" | "interrupted" | "error" | null { + switch (status) { + case "idle": + case "ready": + return "completed"; + case "error": + return "error"; + case "interrupted": + case "stopped": + return "interrupted"; + case "starting": + case "running": + return null; + } +} + function checkpointStatusToTurnState( status: "ready" | "missing" | "error", ): OrchestrationLatestTurn["state"] { From 3b80ae3ebb86ab329a8c6c3c52f5d318919ac4ee Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 10 Jun 2026 22:48:25 -0700 Subject: [PATCH 2/3] Handle steered in-flight turns in ACP adapters - Reuse the active turn when a new sendTurn arrives mid-prompt - Prevent superseded prompt completions from closing the merged turn - Add Cursor coverage for steering and delay the mock agent prompt --- apps/server/scripts/acp-mock-agent.ts | 5 + .../Layers/ProjectionPipeline.ts | 7 +- .../src/provider/Layers/ClaudeAdapter.test.ts | 4 +- .../src/provider/Layers/CursorAdapter.test.ts | 77 +++++ .../src/provider/Layers/CursorAdapter.ts | 246 +++++++------ .../server/src/provider/Layers/GrokAdapter.ts | 322 ++++++++++-------- 6 files changed, 414 insertions(+), 247 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 1a38e489fb2..2b5da74eef0 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -22,6 +22,7 @@ const emitXAiAskUserQuestion = process.env.T3_ACP_EMIT_XAI_ASK_USER_QUESTION === 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 permissionOptionIds = { allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once", allowAlways: process.env.T3_ACP_ALLOW_ALWAYS_OPTION_ID ?? "allow-always", @@ -364,6 +365,10 @@ const program = Effect.gen(function* () { Effect.gen(function* () { const requestedSessionId = String(request.sessionId ?? sessionId); + if (Number.isFinite(promptDelayMs) && promptDelayMs > 0) { + yield* Effect.sleep(`${promptDelayMs} millis`); + } + if (emitInterleavedAssistantToolCalls) { const toolCallId = "tool-call-1"; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index c789d3a7437..8fedddcfbd8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1026,9 +1026,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti // Leaving the "running" session status is the turn-end signal: // settle still-running turns so their duration reflects the whole // turn rather than the last assistant message. - const settledTurnState = settledTurnStateForSessionStatus( - event.payload.session.status, - ); + const settledTurnState = settledTurnStateForSessionStatus(event.payload.session.status); if (settledTurnState === null) { return; } @@ -1062,8 +1060,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti }); yield* Effect.forEach( otherRunningTurns.filter( - (turn) => - turn.turnId !== null && turn.turnId !== turnId && turn.state === "running", + (turn) => turn.turnId !== null && turn.turnId !== turnId && turn.state === "running", ), (turn) => turn.turnId === null diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 502ede7d40a..916c9d077dd 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -943,9 +943,7 @@ describe("ClaudeAdapterLive", () => { 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", - ); + const turnCompletedEvents = runtimeEvents.filter((event) => event.type === "turn.completed"); // One turn boundary for the whole run: the steer produced no // turn.completed/turn.started pair. diff --git a/apps/server/src/provider/Layers/CursorAdapter.test.ts b/apps/server/src/provider/Layers/CursorAdapter.test.ts index b1e54948bae..c71c6964459 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.test.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.test.ts @@ -13,6 +13,7 @@ import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { createModelSelection } from "@t3tools/shared/model"; import { @@ -232,6 +233,82 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => { }), ); + it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-steer-thread"); + + // Keep the first prompt in flight long enough for the steer to land. + const wrapperPath = yield* Effect.promise(() => + makeMockAgentWrapper({ T3_ACP_PROMPT_DELAY_MS: "1500" }), + ); + 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: "run 5 commands", + attachments: [], + }) + .pipe(Effect.forkChild); + + // Poll until the first prompt is in flight — sendTurn binds the active + // turn id before prompting. The mock agent runs on the real clock, so + // each TestClock.adjust just provides the scheduler hops for its stdio + // responses to land. + yield* Effect.gen(function* () { + for (let attempt = 0; attempt < 200; attempt += 1) { + const sessions = yield* adapter.listSessions(); + const session = sessions.find((entry) => entry.threadId === threadId); + if (session?.activeTurnId !== undefined) { + return; + } + yield* TestClock.adjust("10 millis"); + } + throw new Error("Timed out waiting for the first prompt to be in flight."); + }); + + // Steer: a second sendTurn while the first prompt is still in flight + // continues the same turn. + const steeredTurn = yield* adapter.sendTurn({ + threadId, + input: "actually run 15", + attachments: [], + }); + const firstTurn = yield* Fiber.join(firstTurnFiber); + assert.equal(String(steeredTurn.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 whole run: the superseded first prompt + // resolving must not settle the merged turn. + 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("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 016feeb79a4..cdb3c224b97 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -132,6 +132,10 @@ interface CursorSessionContext { readonly turns: Array<{ id: TurnId; items: Array }>; lastPlanFingerprint: string | undefined; activeTurnId: TurnId | 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. */ + promptsInFlight: number; stopped: boolean; } @@ -754,6 +758,7 @@ export function makeCursorAdapter( turns: [], lastPlanFingerprint: undefined, activeTurnId: undefined, + promptsInFlight: 0, stopped: false, }; @@ -882,121 +887,152 @@ export function makeCursorAdapter( const sendTurn: CursorAdapterShape["sendTurn"] = (input) => Effect.gen(function* () { const ctx = yield* requireSession(input.threadId); - const turnId = TurnId.make(yield* randomUUIDv4); - const turnModelSelection = - input.modelSelection?.instanceId === boundInstanceId ? input.modelSelection : undefined; - const model = turnModelSelection?.model ?? ctx.session.model; - const resolvedModel = resolveCursorAcpBaseModelId(model); - yield* applyRequestedSessionConfiguration({ - runtime: ctx.acp, - runtimeMode: ctx.session.runtimeMode, - interactionMode: input.interactionMode, - modelSelection: - model === undefined - ? undefined - : { - model, - options: turnModelSelection?.options, - }, - mapError: ({ cause, method }) => - mapAcpToAdapterError(PROVIDER, input.threadId, method, cause), - }); - ctx.activeTurnId = turnId; - ctx.lastPlanFingerprint = undefined; - ctx.session = { - ...ctx.session, - activeTurnId: turnId, - updatedAt: yield* nowIso, - }; - - yield* offerRuntimeEvent({ - type: "turn.started", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId, - payload: { model: resolvedModel }, - }); + // 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. + 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; the matching + // decrement is the `ensuring` below. + ctx.promptsInFlight += 1; + + return yield* Effect.gen(function* () { + const turnModelSelection = + input.modelSelection?.instanceId === boundInstanceId ? input.modelSelection : undefined; + const model = turnModelSelection?.model ?? ctx.session.model; + const resolvedModel = resolveCursorAcpBaseModelId(model); + yield* applyRequestedSessionConfiguration({ + runtime: ctx.acp, + runtimeMode: ctx.session.runtimeMode, + interactionMode: input.interactionMode, + modelSelection: + model === undefined + ? undefined + : { + model, + options: turnModelSelection?.options, + }, + 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, + }; - const promptParts: Array = []; - if (input.input?.trim()) { - promptParts.push({ type: "text", text: input.input.trim() }); - } - if (input.attachments && input.attachments.length > 0) { - for (const attachment of input.attachments) { - const attachmentPath = resolveAttachmentPath({ - attachmentsDir: serverConfig.attachmentsDir, - attachment, + if (steeringTurnId === undefined) { + yield* offerRuntimeEvent({ + type: "turn.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: { model: resolvedModel }, }); - if (!attachmentPath) { - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: `Invalid attachment id '${attachment.id}'.`, + } + + const promptParts: Array = []; + if (input.input?.trim()) { + promptParts.push({ type: "text", text: input.input.trim() }); + } + if (input.attachments && input.attachments.length > 0) { + for (const attachment of input.attachments) { + const attachmentPath = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }); + if (!attachmentPath) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: `Invalid attachment id '${attachment.id}'.`, + }); + } + const bytes = yield* fileSystem.readFile(attachmentPath).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: cause.message, + cause, + }), + ), + ); + promptParts.push({ + type: "image", + data: Buffer.from(bytes).toString("base64"), + mimeType: attachment.mimeType, }); } - const bytes = yield* fileSystem.readFile(attachmentPath).pipe( - Effect.mapError( - (cause) => - new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: cause.message, - cause, - }), - ), - ); - promptParts.push({ - type: "image", - data: Buffer.from(bytes).toString("base64"), - mimeType: attachment.mimeType, - }); } - } - if (promptParts.length === 0) { - return yield* new ProviderAdapterValidationError({ - provider: PROVIDER, - operation: "sendTurn", - issue: "Turn requires non-empty text or attachments.", - }); - } + if (promptParts.length === 0) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Turn requires non-empty text or attachments.", + }); + } - const result = yield* ctx.acp - .prompt({ - prompt: promptParts, - }) - .pipe( - Effect.mapError((error) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), - ), - ); + const result = yield* ctx.acp + .prompt({ + prompt: promptParts, + }) + .pipe( + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), + ), + ); - ctx.turns.push({ id: turnId, items: [{ prompt: promptParts, result }] }); - ctx.session = { - ...ctx.session, - activeTurnId: turnId, - updatedAt: yield* nowIso, - model: resolvedModel, - }; + const turnRecord = ctx.turns.find((turn) => turn.id === turnId); + if (turnRecord) { + turnRecord.items.push({ prompt: promptParts, result }); + } else { + ctx.turns.push({ id: turnId, items: [{ prompt: promptParts, result }] }); + } + ctx.session = { + ...ctx.session, + activeTurnId: turnId, + updatedAt: yield* nowIso, + model: resolvedModel, + }; - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId, - payload: { - state: result.stopReason === "cancelled" ? "cancelled" : "completed", - stopReason: result.stopReason ?? null, - }, - }); + // 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. + if (ctx.promptsInFlight === 1) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: { + state: result.stopReason === "cancelled" ? "cancelled" : "completed", + stopReason: result.stopReason ?? null, + }, + }); + } - return { - threadId: input.threadId, - turnId, - resumeCursor: ctx.session.resumeCursor, - }; + return { + threadId: input.threadId, + turnId, + resumeCursor: ctx.session.resumeCursor, + }; + }).pipe( + Effect.ensuring( + Effect.sync(() => { + ctx.promptsInFlight = Math.max(0, ctx.promptsInFlight - 1); + }), + ), + ); }); const interruptTurn: CursorAdapterShape["interruptTurn"] = (threadId) => diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index ed4097ecda3..0f1007f261b 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -107,6 +107,10 @@ interface GrokSessionContext { turns: Array<{ id: TurnId; items: Array }>; lastPlanFingerprint: string | undefined; activeTurnId: TurnId | 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. */ + promptsInFlight: number; currentModelId: string | undefined; stopped: boolean; } @@ -555,6 +559,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte turns: [], lastPlanFingerprint: undefined, activeTurnId: undefined, + promptsInFlight: 0, currentModelId: boundModelId, stopped: false, }; @@ -664,150 +669,199 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte input.threadId, Effect.gen(function* () { const ctx = yield* requireSession(input.threadId); - const turnId = TurnId.make(yield* randomUUIDv4); - const turnModelSelection = - input.modelSelection?.instanceId === boundInstanceId - ? input.modelSelection + // 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. + 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; + + return yield* Effect.gen(function* () { + const turnModelSelection = + input.modelSelection?.instanceId === boundInstanceId + ? input.modelSelection + : undefined; + const requestedTurnModelId = turnModelSelection?.model + ? resolveGrokAcpBaseModelId(turnModelSelection.model) : undefined; - const requestedTurnModelId = turnModelSelection?.model - ? resolveGrokAcpBaseModelId(turnModelSelection.model) - : undefined; - const currentModelId = yield* applyGrokAcpModelSelection({ - runtime: ctx.acp, - currentModelId: ctx.currentModelId, - requestedModelId: requestedTurnModelId, - mapError: (cause) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), - }); + const currentModelId = yield* applyGrokAcpModelSelection({ + runtime: ctx.acp, + currentModelId: ctx.currentModelId, + requestedModelId: requestedTurnModelId, + mapError: (cause) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), + }); - const text = input.input?.trim(); - const imagePromptParts = yield* Effect.forEach(input.attachments ?? [], (attachment) => - Effect.gen(function* () { - const attachmentPath = resolveAttachmentPath({ - attachmentsDir: serverConfig.attachmentsDir, - attachment, - }); - if (!attachmentPath) { - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: `Invalid attachment id '${attachment.id}'.`, - }); - } - const bytes = yield* fileSystem.readFile(attachmentPath).pipe( - Effect.mapError( - (cause) => - new ProviderAdapterRequestError({ + const text = input.input?.trim(); + const imagePromptParts = yield* Effect.forEach( + input.attachments ?? [], + (attachment) => + Effect.gen(function* () { + const attachmentPath = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }); + if (!attachmentPath) { + return yield* new ProviderAdapterRequestError({ provider: PROVIDER, method: "session/prompt", - detail: cause.message, - cause, - }), - ), - ); - return { - type: "image", - data: Buffer.from(bytes).toString("base64"), - mimeType: attachment.mimeType, - } satisfies EffectAcpSchema.ContentBlock; - }), - ); - const promptParts: Array = [ - ...(text ? [{ type: "text" as const, text }] : []), - ...imagePromptParts, - ]; - - if (promptParts.length === 0) { - return yield* new ProviderAdapterValidationError({ - provider: PROVIDER, - operation: "sendTurn", - issue: "Turn requires non-empty text or attachments.", - }); - } - - ctx.currentModelId = currentModelId; - const displayModel = currentModelId - ? resolveGrokAcpBaseModelId(currentModelId) - : undefined; - ctx.activeTurnId = turnId; - ctx.lastPlanFingerprint = undefined; - ctx.session = { - ...ctx.session, - activeTurnId: turnId, - updatedAt: yield* nowIso, - ...(displayModel ? { model: displayModel } : {}), - }; - - yield* offerRuntimeEvent({ - type: "turn.started", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId, - payload: displayModel ? { model: displayModel } : {}, - }); + detail: `Invalid attachment id '${attachment.id}'.`, + }); + } + const bytes = yield* fileSystem.readFile(attachmentPath).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: cause.message, + cause, + }), + ), + ); + return { + type: "image", + data: Buffer.from(bytes).toString("base64"), + mimeType: attachment.mimeType, + } satisfies EffectAcpSchema.ContentBlock; + }), + ); + const promptParts: Array = [ + ...(text ? [{ type: "text" as const, text }] : []), + ...imagePromptParts, + ]; + + if (promptParts.length === 0) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Turn requires non-empty text or attachments.", + }); + } - return { - acp: ctx.acp, - acpSessionId: ctx.acpSessionId, - displayModel, - promptParts, - turnId, - }; + ctx.currentModelId = currentModelId; + const displayModel = currentModelId + ? resolveGrokAcpBaseModelId(currentModelId) + : undefined; + ctx.activeTurnId = turnId; + if (steeringTurnId === undefined) { + ctx.lastPlanFingerprint = undefined; + } + ctx.session = { + ...ctx.session, + activeTurnId: turnId, + updatedAt: yield* nowIso, + ...(displayModel ? { model: displayModel } : {}), + }; + + if (steeringTurnId === undefined) { + yield* offerRuntimeEvent({ + type: "turn.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: displayModel ? { model: displayModel } : {}, + }); + } + + return { + acp: ctx.acp, + acpSessionId: ctx.acpSessionId, + displayModel, + promptParts, + turnId, + }; + }).pipe( + Effect.tapCause(() => + Effect.sync(() => { + ctx.promptsInFlight = Math.max(0, ctx.promptsInFlight - 1); + }), + ), + ); }), ); - const result = yield* prepared.acp - .prompt({ - prompt: prepared.promptParts, - }) - .pipe( - Effect.mapError((error) => - mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), - ), - ); + return yield* Effect.gen(function* () { + const result = yield* prepared.acp + .prompt({ + prompt: prepared.promptParts, + }) + .pipe( + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), + ), + ); - return yield* withThreadLock( - input.threadId, - Effect.gen(function* () { - const ctx = yield* requireSession(input.threadId); - if (ctx.acpSessionId !== prepared.acpSessionId) { - return yield* new ProviderAdapterRequestError({ - provider: PROVIDER, - method: "session/prompt", - detail: "Grok session changed before the turn completed.", - }); - } - - ctx.turns = [ - ...ctx.turns, - { id: prepared.turnId, items: [{ prompt: prepared.promptParts, result }] }, - ]; - ctx.session = { - ...ctx.session, - activeTurnId: prepared.turnId, - updatedAt: yield* nowIso, - ...(prepared.displayModel ? { model: prepared.displayModel } : {}), - }; - - yield* offerRuntimeEvent({ - type: "turn.completed", - ...(yield* makeEventStamp()), - provider: PROVIDER, - threadId: input.threadId, - turnId: prepared.turnId, - payload: { - state: result.stopReason === "cancelled" ? "cancelled" : "completed", - stopReason: result.stopReason ?? null, - }, - }); + return yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(input.threadId); + if (ctx.acpSessionId !== prepared.acpSessionId) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: "Grok session changed before the turn completed.", + }); + } - return { - threadId: input.threadId, - turnId: prepared.turnId, - resumeCursor: ctx.session.resumeCursor, - }; - }), + const existingTurnRecord = ctx.turns.find((turn) => turn.id === prepared.turnId); + ctx.turns = existingTurnRecord + ? ctx.turns.map((turn) => + turn.id === prepared.turnId + ? { + ...turn, + items: [...turn.items, { prompt: prepared.promptParts, result }], + } + : turn, + ) + : [ + ...ctx.turns, + { id: prepared.turnId, items: [{ prompt: prepared.promptParts, result }] }, + ]; + ctx.session = { + ...ctx.session, + activeTurnId: prepared.turnId, + updatedAt: yield* nowIso, + ...(prepared.displayModel ? { model: prepared.displayModel } : {}), + }; + + // 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. + if (ctx.promptsInFlight === 1) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId: prepared.turnId, + payload: { + state: result.stopReason === "cancelled" ? "cancelled" : "completed", + stopReason: result.stopReason ?? null, + }, + }); + } + + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + }), + ); + }).pipe( + Effect.ensuring( + Effect.sync(() => { + const liveCtx = sessions.get(input.threadId); + if (liveCtx) { + liveCtx.promptsInFlight = Math.max(0, liveCtx.promptsInFlight - 1); + } + }), + ), ); }); From 4f09fb2cbaa0fe732339aa3d980afd0430824f5f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 10 Jun 2026 22:53:31 -0700 Subject: [PATCH 3/3] Pin @expo/metro-config to the patched version A fresh resolve (release-smoke regenerates the lockfile from scratch) started picking the just-published @expo/metro-config@56.0.14, orphaning the 56.0.13 patch and failing with ERR_PNPM_UNUSED_PATCH. Pin the override to the patched version so unlocked resolution stays deterministic. Co-Authored-By: Claude Fable 5 --- pnpm-lock.yaml | 3 ++- pnpm-workspace.yaml | 4 ++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a3e7b69e624..96d563a2fd5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -50,6 +50,7 @@ overrides: '@effect/vitest': 4.0.0-beta.78 '@effect/vitest>@vitest/runner': '-' '@effect/vitest>vitest': '-' + '@expo/metro-config': 56.0.13 '@types/node': 24.12.4 effect: 4.0.0-beta.78 vite: npm:@voidzero-dev/vite-plus-core@0.1.24 @@ -12620,7 +12621,7 @@ snapshots: '@babel/plugin-transform-private-methods': 7.29.7(@babel/core@7.29.7) '@babel/plugin-transform-private-property-in-object': 7.29.7(@babel/core@7.29.7) '@babel/plugin-transform-react-display-name': 7.29.7(@babel/core@7.29.7) - '@babel/plugin-transform-react-jsx': 7.29.7(@babel/core@7.29.7) + '@babel/plugin-transform-react-jsx': 7.28.6(@babel/core@7.29.7) '@babel/plugin-transform-react-jsx-self': 7.29.7(@babel/core@7.29.7) '@babel/plugin-transform-react-jsx-source': 7.29.7(@babel/core@7.29.7) '@babel/plugin-transform-regenerator': 7.29.7(@babel/core@7.29.7) diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index ccd810fdbf7..4ab589fffbd 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -56,6 +56,10 @@ overrides: "@effect/vitest": "catalog:" "@effect/vitest>@vitest/runner": "-" "@effect/vitest>vitest": "-" + # Pinned to the version the patch in patchedDependencies targets — a fresh + # resolve (e.g. release-smoke regenerating the lockfile) must not drift to a + # newer release and orphan the patch (ERR_PNPM_UNUSED_PATCH). + "@expo/metro-config": "56.0.13" "@types/node": "catalog:" effect: "catalog:" vite: "catalog:"