From e5b9b1edc71addb1384d2641f586c1c0b847ac1c Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Fri, 2 Oct 2026 12:46:12 +0200 Subject: [PATCH 1/4] fix(opencode): keep tool-call progress out of assistant answers --- .../TestProviderAdapter.integration.test.ts | 44 ++ .../TestProviderAdapter.integration.ts | 6 +- .../Layers/ProjectionPipeline.test.ts | 85 +++ .../Layers/ProjectionPipeline.ts | 27 +- .../Layers/ProviderRuntimeIngestion.test.ts | 182 ++++++ .../Layers/ProviderRuntimeIngestion.ts | 144 ++++- .../src/orchestration/projector.test.ts | 48 +- apps/server/src/orchestration/projector.ts | 25 +- .../Layers/ProjectionThreadMessages.test.ts | 41 ++ .../Layers/ProjectionThreadMessages.ts | 28 + .../Services/ProjectionThreadMessages.ts | 13 + .../provider/Layers/OpenCodeAdapter.test.ts | 535 +++++++++++++++++- .../src/provider/Layers/OpenCodeAdapter.ts | 270 ++++++++- .../src/state/threadReducer.test.ts | 130 +++++ .../client-runtime/src/state/threadReducer.ts | 30 +- packages/contracts/src/orchestration.ts | 7 +- packages/contracts/src/providerRuntime.ts | 4 + 17 files changed, 1534 insertions(+), 85 deletions(-) create mode 100644 apps/server/integration/TestProviderAdapter.integration.test.ts diff --git a/apps/server/integration/TestProviderAdapter.integration.test.ts b/apps/server/integration/TestProviderAdapter.integration.test.ts new file mode 100644 index 000000000000..7c4acb407cf7 --- /dev/null +++ b/apps/server/integration/TestProviderAdapter.integration.test.ts @@ -0,0 +1,44 @@ +import { EventId, ProviderDriverKind, ThreadId } from "@t3tools/contracts"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { makeTestProviderAdapterHarness } from "./TestProviderAdapter.integration.ts"; + +it.effect("keeps progress deltas out of the synthetic assistant answer", () => + Effect.gen(function* () { + const harness = yield* makeTestProviderAdapterHarness(); + const threadId = ThreadId.make("thread-progress-fixture"); + yield* harness.adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + cwd: "/repo", + }); + const eventBase = { + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId: String(threadId), + type: "content.delta", + } as const; + yield* harness.queueTurnResponse(threadId, { + events: [ + { + ...eventBase, + eventId: EventId.make("progress-delta"), + payload: { streamKind: "assistant_progress_text", delta: "Thinking aloud. " }, + }, + { + ...eventBase, + eventId: EventId.make("answer-delta"), + payload: { streamKind: "assistant_text", delta: "Final answer." }, + }, + ], + }); + yield* harness.adapter.sendTurn({ threadId, input: "Prompt", attachments: [] }); + const snapshot = yield* harness.adapter.readThread(threadId); + assert.deepStrictEqual(snapshot.turns[0]?.items, [ + { type: "userMessage", content: [{ type: "text", text: "Prompt" }] }, + { type: "agentMessage", text: "Final answer." }, + ]); + }), +); diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts index 095cca4e5e74..816c99e9ef6b 100644 --- a/apps/server/integration/TestProviderAdapter.integration.ts +++ b/apps/server/integration/TestProviderAdapter.integration.ts @@ -325,8 +325,10 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter const runtimeEvent = normalizeFixtureEvent(rawEvent); const runtimeType = (runtimeEvent as { type: string }).type; if (runtimeType === "content.delta") { - const payload = runtimeEvent.payload as { delta?: unknown } | undefined; - if (typeof payload?.delta === "string") { + const payload = runtimeEvent.payload as + | { delta?: unknown; streamKind?: unknown } + | undefined; + if (payload?.streamKind === "assistant_text" && typeof payload.delta === "string") { assistantDeltas.push(payload.delta); } } else if (runtimeType === "message.delta") { diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index e8c7154f38e7..0f31bc8bba07 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -67,6 +67,91 @@ const encodeThreadLinkedPullRequest = Schema.encodeSync( Schema.fromJsonString(ThreadLinkedPullRequest), ); +it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-reclassified-anchor-")))( + "reasoning message projection", + (it) => { + it.effect("clears the assistant anchor when progress becomes reasoning", () => + Effect.gen(function* () { + const pipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-reclassified-anchor"); + const turnId = TurnId.make("turn-reclassified-anchor"); + const messageId = MessageId.make("assistant:reclassified-anchor"); + const at = "2026-03-01T08:00:00.000Z"; + const append = (event: Parameters[0]) => + eventStore.append(event).pipe(Effect.flatMap(pipeline.projectEvent)); + const fields = (id: string) => ({ + eventId: EventId.make(`evt-${id}`), + occurredAt: at, + commandId: CommandId.make(`cmd-${id}`), + causationEventId: null, + correlationId: CorrelationId.make(`cmd-${id}`), + metadata: {}, + }); + + yield* append({ + ...fields("anchor-project"), + type: "project.created", + aggregateKind: "project", + aggregateId: ProjectId.make("project-reclassified-anchor"), + payload: { + projectId: ProjectId.make("project-reclassified-anchor"), + title: "Anchor", + workspaceRoot: "/tmp/reclassified-anchor", + defaultModelSelection: null, + scripts: [], + createdAt: at, + updatedAt: at, + }, + }); + yield* append({ + ...fields("anchor-thread"), + type: "thread.created", + aggregateKind: "thread", + aggregateId: threadId, + payload: { + threadId, + projectId: ProjectId.make("project-reclassified-anchor"), + title: "Anchor", + modelSelection: { instanceId: ProviderInstanceId.make("opencode"), model: "mimo" }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: at, + updatedAt: at, + }, + }); + const sendMessage = (role: "assistant" | "reasoning", id: string) => + append({ + ...fields(id), + type: "thread.message-sent", + aggregateKind: "thread", + aggregateId: threadId, + payload: { + threadId, + messageId, + role, + text: role === "assistant" ? "Checking the audit." : "", + turnId, + streaming: role === "assistant", + createdAt: at, + updatedAt: at, + }, + }); + yield* sendMessage("assistant", "anchor-assistant"); + const anchor = sql<{ readonly assistantMessageId: string | null }>` + SELECT assistant_message_id AS "assistantMessageId" + FROM projection_turns WHERE turn_id = ${turnId} + `; + assert.deepEqual(yield* anchor, [{ assistantMessageId: messageId }]); + yield* sendMessage("reasoning", "anchor-reasoning"); + assert.deepEqual(yield* anchor, [{ assistantMessageId: null }]); + }), + ); + }, +); + it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-cursor-batch-")))( "OrchestrationProjectionPipeline cursor batches", (it) => { diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 1acb3c360b15..78458dc8358e 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1582,7 +1582,32 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } case "thread.message-sent": { - if (event.payload.turnId === null || event.payload.role !== "assistant") { + if (event.payload.turnId === null) { + return; + } + if (event.payload.role === "reasoning") { + const existingTurn = yield* projectionTurnRepository.getByTurnId({ + threadId: event.payload.threadId, + turnId: event.payload.turnId, + }); + if ( + Option.isSome(existingTurn) && + existingTurn.value.assistantMessageId === event.payload.messageId + ) { + const replacement = + yield* projectionThreadMessageRepository.getLatestAssistantMessageIdForTurn({ + threadId: event.payload.threadId, + turnId: event.payload.turnId, + excludeMessageId: event.payload.messageId, + }); + yield* projectionTurnRepository.upsertByTurnId({ + ...existingTurn.value, + assistantMessageId: Option.getOrNull(replacement), + }); + } + return; + } + if (event.payload.role !== "assistant") { return; } // A completed assistant message only settles the turn once the diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 4c6e3703ba82..6e976d8a1d0d 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1439,6 +1439,188 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + it("keeps a completed OpenCode turn with reasoning but no answer visible", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-empty-answer"); + const now = "2026-01-01T00:00:00.000Z"; + await harness.emitAndDrain([ + { + type: "content.delta", + eventId: asEventId("evt-empty-reasoning"), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId, + turnId, + itemId: asItemId("reasoning-part"), + payload: { streamKind: "reasoning_summary_text", delta: "Checked the branch." }, + }, + { + type: "turn.completed", + eventId: asEventId("evt-empty-turn-complete"), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId, + turnId, + status: "completed", + }, + ]); + + const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId)!; + expect( + thread.messages.find((message: ProviderRuntimeTestMessage) => message.role === "reasoning"), + ).toMatchObject({ text: "Checked the branch.", streaming: false }); + expect( + thread.messages.some((message: ProviderRuntimeTestMessage) => message.role === "assistant"), + ).toBe(false); + }); + + it("reclassifies completed OpenCode progress text without changing the final answer", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-progress"); + const now = "2026-01-01T00:00:00.000Z"; + const emitText = (item: string, text: string, event: string) => + harness.emitAndDrain([ + { + type: "content.delta", + eventId: asEventId(event), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId, + turnId, + itemId: asItemId(item), + payload: { streamKind: "assistant_text", delta: text }, + }, + ]); + const complete = (item: string, event: string, presentation?: "progress", detail?: string) => + harness.emitAndDrain([ + { + type: "item.completed", + eventId: asEventId(event), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId, + turnId, + itemId: asItemId(item), + payload: { + itemType: "assistant_message", + status: "completed", + ...(presentation ? { presentation } : {}), + ...(detail ? { detail } : {}), + }, + }, + ]); + + await emitText("preface-part", "Starting the review.", "evt-preface-text"); + await emitText("progress-part", "Checking the reviews.", "evt-progress-text"); + await complete("progress-part", "evt-progress-complete", undefined, "Checking the reviews."); + await emitText("answer-part", "The review ", "evt-answer-text-first"); + await complete("progress-part", "evt-progress-classified", "progress"); + await emitText("answer-part", "is complete.", "evt-answer-text-second"); + await complete("answer-part", "evt-answer-complete"); + + const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId)!; + expect( + thread.messages + .filter((message: ProviderRuntimeTestMessage) => + ["assistant:preface-part", "assistant:progress-part", "assistant:answer-part"].includes( + message.id, + ), + ) + .map((message: ProviderRuntimeTestMessage) => [message.id, message.role, message.text]) + .sort(), + ).toEqual([ + ["assistant:answer-part", "assistant", "The review is complete."], + ["assistant:preface-part", "assistant", "Starting the review."], + ["assistant:progress-part", "reasoning", "Checking the reviews."], + ]); + }); + + it("keeps a classified OpenCode progress part out of the following answer segment", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-progress-stream"); + const now = "2026-01-01T00:00:00.000Z"; + const emitText = ( + item: string, + text: string, + event: string, + streamKind: "assistant_text" | "assistant_progress_text", + ) => + harness.emitAndDrain([ + { + type: "content.delta", + eventId: asEventId(event), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId, + turnId, + itemId: asItemId(item), + payload: { streamKind, delta: text }, + }, + ]); + const complete = (item: string, event: string, presentation?: "progress") => + harness.emitAndDrain([ + { + type: "item.completed", + eventId: asEventId(event), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId, + turnId, + itemId: asItemId(item), + payload: { + itemType: "assistant_message", + status: "completed", + ...(presentation ? { presentation } : {}), + }, + }, + ]); + + await emitText("progress-part", "Checking ", "evt-progress-first", "assistant_text"); + await complete("progress-part", "evt-progress-classified", "progress"); + await emitText("progress-part", "the ", "evt-progress-second", "assistant_progress_text"); + await emitText("progress-part", "reviews.", "evt-progress-third", "assistant_progress_text"); + await emitText("answer-part", "Review complete.", "evt-answer", "assistant_text"); + await complete("answer-part", "evt-answer-complete"); + await harness.emitAndDrain([ + { + type: "turn.completed", + eventId: asEventId("evt-turn-complete"), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId, + turnId, + status: "completed", + }, + ]); + + const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId)!; + expect( + thread.messages + .map((message: ProviderRuntimeTestMessage) => [message.id, message.role, message.text]) + .sort(), + ).toEqual([ + ["assistant:answer-part", "assistant", "Review complete."], + ["assistant:progress-part", "reasoning", "Checking the reviews."], + ]); + + await emitText("progress-part", " Later note.", "evt-progress-late", "assistant_progress_text"); + const withLateProgress = (await harness.readModel()).threads.find( + (entry) => entry.id === threadId, + )!; + expect( + withLateProgress.messages.find( + (message: ProviderRuntimeTestMessage) => message.id === "assistant:progress-part", + ), + ).toMatchObject({ + role: "reasoning", + text: "Checking the reviews. Later note.", + streaming: false, + }); + }); + it("streams reasoning deltas into a finalized reasoning message", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 2584a7578200..901209ed957e 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -311,6 +311,8 @@ function assistantSegmentBaseKeyFromEvent(event: ProviderRuntimeEvent): string { * machinery; only the message id namespace differs. The prefix is what tells a * buffered segment apart when it is flushed or finalized long after the delta * that opened it, so the role never has to be threaded through those paths. + * OpenCode progress keeps its assistant id because it can be classified after + * assistant text has already reached the client. */ type MessageStreamRole = "assistant" | "reasoning"; @@ -1075,6 +1077,12 @@ const make = Effect.gen(function* () { timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL, lookup: () => Effect.succeed(0), }); + // Late progress deltas still use the message id that was first streamed as assistant text. + const progressMessageIds = yield* Cache.make({ + capacity: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY, + timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL, + lookup: () => Effect.succeed(false), + }); // When a thinking block opened, so "Thought for ..." measures the model's // time and not the moment buffered text happened to be flushed. @@ -1267,22 +1275,38 @@ const make = Effect.gen(function* () { turnId?: TurnId; }) => Effect.gen(function* () { + const baseKey = assistantSegmentBaseKeyFromEvent(input.event); if (!input.turnId) { - return assistantSegmentMessageId(assistantSegmentBaseKeyFromEvent(input.event), 0); + return assistantSegmentMessageId(baseKey, 0); } - const activeMessageId = yield* getActiveAssistantMessageIdForTurn( - input.threadId, - input.turnId, - ); - if (Option.isSome(activeMessageId)) { - return activeMessageId.value; + const state = yield* getAssistantSegmentStateForTurn(input.threadId, input.turnId); + const activeMessageId = Option.getOrUndefined(state)?.activeMessageId; + if (activeMessageId) { + // OpenCode can classify an already streamed text part as tool-call + // progress when its parent message finishes. Keep each part separate + // so that reclassification cannot move text from an earlier answer. + if ( + input.event.provider !== "opencode" || + Option.getOrUndefined(state)?.baseKey === baseKey + ) { + return activeMessageId; + } + yield* finalizeActiveSegmentForTurn({ + event: input.event, + threadId: input.threadId, + turnId: input.turnId, + createdAt: input.event.createdAt, + commandTag: "assistant-complete-on-new-part", + finalDeltaCommandTag: "assistant-delta-finalize-on-new-part", + hasProjectedMessage: false, + }); } return yield* startAssistantSegmentForTurn({ threadId: input.threadId, turnId: input.turnId, - baseKey: assistantSegmentBaseKeyFromEvent(input.event), + baseKey, }); }); @@ -1423,6 +1447,7 @@ const make = Effect.gen(function* () { const clearAssistantMessageState = (messageId: MessageId) => clearBufferedAssistantText(messageId).pipe( + Effect.andThen(Cache.invalidate(progressMessageIds, messageId)), Effect.andThen(Cache.invalidate(reasoningPartIndexByMessageId, messageId)), Effect.andThen(Cache.invalidate(reasoningStartedAtByMessageId, messageId)), ); @@ -1446,7 +1471,9 @@ const make = Effect.gen(function* () { return false; } - const isReasoning = messageStreamRoleOf(input.messageId) === "reasoning"; + const isReasoning = + messageStreamRoleOf(input.messageId) === "reasoning" || + Option.isSome(yield* Cache.getOption(progressMessageIds, input.messageId)); yield* orchestrationEngine.dispatch({ type: isReasoning ? "thread.message.reasoning.delta" : "thread.message.assistant.delta", commandId: yield* providerCommandId(input.event, input.commandTag), @@ -1504,6 +1531,7 @@ const make = Effect.gen(function* () { finalDeltaCommandTag: string; fallbackText?: string; hasProjectedMessage?: boolean; + presentation?: "progress"; }) => Effect.gen(function* () { const bufferedText = yield* takeBufferedAssistantText(input.messageId); @@ -1515,7 +1543,10 @@ const make = Effect.gen(function* () { : ""; const hasRenderableText = hasRenderableAssistantText(text); - const isReasoning = messageStreamRoleOf(input.messageId) === "reasoning"; + const isReasoning = + input.presentation === "progress" || + messageStreamRoleOf(input.messageId) === "reasoning" || + Option.isSome(yield* Cache.getOption(progressMessageIds, input.messageId)); if (hasRenderableText) { yield* orchestrationEngine.dispatch({ @@ -1784,6 +1815,7 @@ const make = Effect.gen(function* () { if ( event.type === "content.delta" && event.payload.streamKind !== "assistant_text" && + event.payload.streamKind !== "assistant_progress_text" && event.payload.streamKind !== "reasoning_text" && event.payload.streamKind !== "reasoning_summary_text" ) { @@ -1962,6 +1994,10 @@ const make = Effect.gen(function* () { event.type === "content.delta" && event.payload.streamKind === "assistant_text" ? event.payload.delta : undefined; + const assistantProgressDelta = + event.type === "content.delta" && event.payload.streamKind === "assistant_progress_text" + ? event.payload.delta + : undefined; const reasoningDelta = event.type === "content.delta" && (event.payload.streamKind === "reasoning_text" || @@ -2097,6 +2133,61 @@ const make = Effect.gen(function* () { } } + if (assistantProgressDelta && assistantProgressDelta.length > 0) { + const turnId = toTurnId(event.turnId); + const messageId = MessageId.make( + `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, + ); + if (turnId) { + yield* rememberAssistantMessageId(thread.id, turnId, messageId); + } + yield* Cache.set(progressMessageIds, messageId, true); + const streamingMode = yield* resolveResponseStreamingMode(thread.projectId); + const progressMode = streamingMode === "token" ? "paragraph" : streamingMode; + const spillChunk = yield* appendBufferedAssistantText( + messageId, + assistantProgressDelta, + progressMode, + yield* Clock.currentTimeMillis, + ); + if (spillChunk.length > 0) { + yield* orchestrationEngine.dispatch({ + type: "thread.message.reasoning.delta", + commandId: yield* providerCommandId(event, "assistant-progress-delta-buffer-spill"), + threadId: thread.id, + messageId, + delta: spillChunk, + ...(turnId ? { turnId } : {}), + createdAt: now, + }); + } + // OpenCode can append text after classifying a completed part as + // progress. Finish a live part at item or turn completion instead of + // persisting a completion for every token. Late deltas from a settled + // turn still need immediate completion. + const progressTurn = + turnId && (activeTurnId === null || !sameId(activeTurnId, turnId)) + ? yield* projectionTurnRepository.getByTurnId({ threadId: thread.id, turnId }) + : Option.none(); + const isSettledProgressTurn = + Option.isSome(progressTurn) && + progressTurn.value.state !== "pending" && + progressTurn.value.state !== "running"; + if (!turnId || isSettledProgressTurn) { + yield* finalizeAssistantMessage({ + event, + threadId: thread.id, + messageId, + ...(turnId ? { turnId } : {}), + createdAt: now, + commandTag: "assistant-progress-complete", + finalDeltaCommandTag: "assistant-progress-delta-finalize", + hasProjectedMessage: (yield* getThreadMessageById(thread.id, messageId)) !== undefined, + presentation: "progress", + }); + } + } + const pauseForUserTurnId = event.type === "request.opened" || (event.type === "user-input.requested" && event.payload.responseMode !== "message") @@ -2256,6 +2347,7 @@ const make = Effect.gen(function* () { `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, ), fallbackText: event.payload.detail, + presentation: event.payload.presentation, } : undefined; const proposedPlanCompletion = @@ -2284,10 +2376,15 @@ const make = Effect.gen(function* () { const activeAssistantMessageId = turnId ? yield* getActiveAssistantMessageIdForTurn(thread.id, turnId) : Option.none(); - const assistantMessageId = Option.getOrElse( - activeAssistantMessageId, - () => assistantCompletion.messageId, - ); + const activeMessageMatchesItem = + Option.isSome(activeAssistantMessageId) && + (activeAssistantMessageId.value === assistantCompletion.messageId || + activeAssistantMessageId.value.startsWith(`${assistantCompletion.messageId}:segment:`)); + const assistantMessageId = + assistantCompletion.presentation === "progress" || + (event.provider === "opencode" && !activeMessageMatchesItem) + ? assistantCompletion.messageId + : Option.getOrElse(activeAssistantMessageId, () => assistantCompletion.messageId); const [existingAssistantMessage, hasAssistantMessagesForTurn] = yield* Effect.all([ getThreadMessageById(thread.id, assistantMessageId), turnId === undefined @@ -2302,6 +2399,7 @@ const make = Effect.gen(function* () { !existingAssistantMessage || existingAssistantMessage.text.length === 0; const shouldSkipRedundantCompletion = + assistantCompletion.presentation !== "progress" && Option.isNone(activeAssistantMessageId) && turnId !== undefined && hasAssistantMessagesForTurn && @@ -2317,10 +2415,16 @@ const make = Effect.gen(function* () { threadId: thread.id, messageId: assistantMessageId, ...(turnId ? { turnId } : {}), - createdAt: now, + createdAt: + assistantCompletion.presentation === "progress" + ? (existingAssistantMessage?.createdAt ?? now) + : now, commandTag: "assistant-complete", finalDeltaCommandTag: "assistant-delta-finalize", hasProjectedMessage: existingAssistantMessage !== undefined, + ...(assistantCompletion.presentation !== undefined + ? { presentation: assistantCompletion.presentation } + : {}), ...(assistantCompletion.fallbackText !== undefined && shouldApplyFallbackCompletionText ? { fallbackText: assistantCompletion.fallbackText } : {}), @@ -2331,7 +2435,12 @@ const make = Effect.gen(function* () { } } - if (turnId) { + if ( + turnId && + (assistantCompletion.presentation !== "progress" || + (Option.isSome(activeAssistantMessageId) && + activeAssistantMessageId.value === assistantMessageId)) + ) { yield* clearAssistantSegmentStateForTurn(thread.id, turnId); } } @@ -2406,6 +2515,9 @@ const make = Effect.gen(function* () { commandTag: "assistant-complete-finalize", finalDeltaCommandTag: "assistant-delta-finalize-fallback", hasProjectedMessage: existingMessage !== undefined, + ...(existingMessage?.role === "reasoning" + ? { presentation: "progress" as const } + : {}), }), ), ), diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 516b85fbdae8..a3a7966f2f13 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -1,9 +1,11 @@ import { CommandId, EventId, + MessageId, ProjectId, ProviderDriverKind, ThreadId, + TurnId, type OrchestrationEvent, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -672,9 +674,49 @@ describe("orchestration projector", () => { ), ); + const withAssistantAnchor = { + ...afterDelta, + threads: afterDelta.threads.map((thread) => ({ + ...thread, + latestTurn: { + turnId: TurnId.make("turn-1"), + state: "running" as const, + requestedAt: deltaAt, + startedAt: deltaAt, + completedAt: null, + assistantMessageId: MessageId.make("assistant:msg-1"), + }, + })), + }; + const afterTurnlessReclassification = await Effect.runPromise( + projectEvent( + withAssistantAnchor, + makeEvent({ + sequence: 3, + type: "thread.message-sent", + aggregateKind: "thread", + aggregateId: "thread-1", + occurredAt: completeAt, + commandId: "cmd-turnless", + payload: { + threadId: "thread-1", + messageId: "assistant:msg-1", + role: "reasoning", + text: "", + turnId: null, + streaming: false, + createdAt: completeAt, + updatedAt: completeAt, + }, + }), + ), + ); + expect(afterTurnlessReclassification.threads[0]?.latestTurn?.assistantMessageId).toBe( + "assistant:msg-1", + ); const afterComplete = await Effect.runPromise( projectEvent( - afterDelta, + withAssistantAnchor, makeEvent({ sequence: 3, type: "thread.message-sent", @@ -685,7 +727,7 @@ describe("orchestration projector", () => { payload: { threadId: "thread-1", messageId: "assistant:msg-1", - role: "assistant", + role: "reasoning", text: "", turnId: "turn-1", streaming: false, @@ -698,9 +740,11 @@ describe("orchestration projector", () => { const message = afterComplete.threads[0]?.messages[0]; expect(message?.id).toBe("assistant:msg-1"); + expect(message?.role).toBe("reasoning"); expect(message?.text).toBe("hello"); expect(message?.streaming).toBe(false); expect(message?.updatedAt).toBe(completeAt); + expect(afterComplete.threads[0]?.latestTurn?.assistantMessageId).toBeNull(); }); it("prunes reverted turn messages from in-memory thread snapshot", async () => { diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index dd418b95b9bc..74aa8f7df1c7 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -9,6 +9,7 @@ import type { } from "@t3tools/contracts"; import { isImportedAgentSessionMessageId, + MAX_THREAD_MESSAGES, OrchestrationCheckpointSummary, OrchestrationMessage, OrchestrationSession, @@ -57,7 +58,6 @@ import { } from "./Schemas.ts"; type ThreadPatch = Partial>; -const MAX_THREAD_MESSAGES = 2_000; const MAX_THREAD_CHECKPOINTS = 500; // Async questions can stay open while the agent produces more activity. @@ -810,6 +810,7 @@ export function projectEvent( entry.id === message.id ? { ...entry, + role: message.role, text: message.streaming ? `${entry.text}${message.text}` : message.text.length > 0 @@ -827,11 +828,33 @@ export function projectEvent( ) : [...thread.messages, message]; const cappedMessages = messages.slice(-MAX_THREAD_MESSAGES); + const reclassifiedAssistant = + existingMessage?.role === "assistant" && + message.role === "reasoning" && + message.turnId !== null; + const replacementAssistantMessageId = reclassifiedAssistant + ? (cappedMessages.findLast( + (entry) => entry.role === "assistant" && entry.turnId === message.turnId, + )?.id ?? null) + : null; + const latestTurn = + reclassifiedAssistant && thread.latestTurn?.assistantMessageId === message.id + ? { ...thread.latestTurn, assistantMessageId: replacementAssistantMessageId } + : thread.latestTurn; + const checkpoints = reclassifiedAssistant + ? thread.checkpoints.map((entry) => + entry.assistantMessageId === message.id + ? { ...entry, assistantMessageId: replacementAssistantMessageId } + : entry, + ) + : thread.checkpoints; return { ...nextBase, threads: patchThreadAt(nextBase.threads, threadIndex, { messages: cappedMessages, + latestTurn, + checkpoints, updatedAt: event.occurredAt, }), }; diff --git a/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts b/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts index 87a15b95e413..a597bc41fc58 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts @@ -2,6 +2,7 @@ import { MessageId, ThreadId, TurnId } from "@t3tools/contracts"; import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import { ProjectionThreadMessageRepository } from "../Services/ProjectionThreadMessages.ts"; import { ProjectionThreadMessageRepositoryLive } from "./ProjectionThreadMessages.ts"; @@ -12,6 +13,46 @@ const layer = it.layer( ); layer("ProjectionThreadMessageRepository", (it) => { + it.effect("orders assistant replacements by time then message id within one turn", () => + Effect.gen(function* () { + const repository = yield* ProjectionThreadMessageRepository; + const threadId = ThreadId.make("thread-reclassified-message"); + const turnId = TurnId.make("turn-reclassified-message"); + for (const [index, role, messageTurnId] of [ + [0, "assistant", turnId], + [1, "assistant", TurnId.make("other-turn")], + [2, "reasoning", turnId], + [3, "assistant", turnId], + [4, "assistant", turnId], + [5, "assistant", turnId], + ] as const) { + yield* repository.upsert({ + messageId: MessageId.make(`reclassified-message-${index}`), + threadId, + turnId: messageTurnId, + role, + text: "Message body", + isStreaming: false, + createdAt: `2026-02-28T19:05:0${Math.min(index, 3)}.000Z`, + updatedAt: `2026-02-28T19:05:0${Math.min(index, 3)}.000Z`, + }); + } + + const replacement = yield* repository.getLatestAssistantMessageIdForTurn({ + threadId, + turnId, + excludeMessageId: MessageId.make("reclassified-message-3"), + }); + assert.strictEqual(Option.getOrNull(replacement), MessageId.make("reclassified-message-5")); + const none = yield* repository.getLatestAssistantMessageIdForTurn({ + threadId, + turnId: TurnId.make("other-turn"), + excludeMessageId: MessageId.make("reclassified-message-1"), + }); + assert.isTrue(Option.isNone(none)); + }), + ); + it.effect("finds the latest live user-message time within one thread", () => Effect.gen(function* () { const repository = yield* ProjectionThreadMessageRepository; diff --git a/apps/server/src/persistence/Layers/ProjectionThreadMessages.ts b/apps/server/src/persistence/Layers/ProjectionThreadMessages.ts index 28aeb6d794e9..044b5ba37f31 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadMessages.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadMessages.ts @@ -11,6 +11,7 @@ import { toPersistenceSqlError } from "../Errors.ts"; import { AppendStreamingProjectionThreadMessage, GetProjectionThreadMessageInput, + GetLatestAssistantMessageIdForTurnInput, HasProjectionThreadAssistantMessageInput, ProjectionThreadMessageRepository, type ProjectionThreadMessageRepositoryShape, @@ -204,6 +205,21 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () { `, }); + const getLatestAssistantMessageIdForTurnRow = SqlSchema.findOneOption({ + Request: GetLatestAssistantMessageIdForTurnInput, + Result: Schema.Struct({ messageId: ProjectionThreadMessage.fields.messageId }), + execute: ({ threadId, turnId, excludeMessageId }) => sql` + SELECT message_id AS "messageId" + FROM projection_thread_messages + WHERE thread_id = ${threadId} + AND turn_id = ${turnId} + AND role = 'assistant' + AND message_id != ${excludeMessageId} + ORDER BY created_at DESC, message_id DESC + LIMIT 1 + `, + }); + const listProjectionThreadMessageRows = SqlSchema.findAll({ Request: ListProjectionThreadMessagesInput, Result: ProjectionThreadMessageDbRowSchema, @@ -279,6 +295,17 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () { Effect.map((row) => row.exists === 1), ); + const getLatestAssistantMessageIdForTurn: ProjectionThreadMessageRepositoryShape["getLatestAssistantMessageIdForTurn"] = + (input) => + getLatestAssistantMessageIdForTurnRow(input).pipe( + Effect.mapError( + toPersistenceSqlError( + "ProjectionThreadMessageRepository.getLatestAssistantMessageIdForTurn:query", + ), + ), + Effect.map(Option.map((row) => row.messageId)), + ); + const listByThreadId: ProjectionThreadMessageRepositoryShape["listByThreadId"] = (input) => listProjectionThreadMessageRows(input).pipe( Effect.mapError( @@ -309,6 +336,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () { appendStreaming, getByMessageId, hasAssistantMessageForTurn, + getLatestAssistantMessageIdForTurn, listByThreadId, getLatestUserMessageAt, deleteByThreadId, diff --git a/apps/server/src/persistence/Services/ProjectionThreadMessages.ts b/apps/server/src/persistence/Services/ProjectionThreadMessages.ts index e3e5b6e5151d..f792bf9a0aa7 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadMessages.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadMessages.ts @@ -61,6 +61,14 @@ export const HasProjectionThreadAssistantMessageInput = Schema.Struct({ export type HasProjectionThreadAssistantMessageInput = typeof HasProjectionThreadAssistantMessageInput.Type; +export const GetLatestAssistantMessageIdForTurnInput = Schema.Struct({ + threadId: ThreadId, + turnId: TurnId, + excludeMessageId: MessageId, +}); +export type GetLatestAssistantMessageIdForTurnInput = + typeof GetLatestAssistantMessageIdForTurnInput.Type; + export const DeleteProjectionThreadMessagesInput = Schema.Struct({ threadId: ThreadId, }); @@ -98,6 +106,11 @@ export interface ProjectionThreadMessageRepositoryShape { input: HasProjectionThreadAssistantMessageInput, ) => Effect.Effect; + /** Find the last assistant message in a turn without loading message bodies. */ + readonly getLatestAssistantMessageIdForTurn: ( + input: GetLatestAssistantMessageIdForTurnInput, + ) => Effect.Effect, ProjectionRepositoryError>; + /** * List projected thread messages for a thread. * diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 40d034089ae9..47ab3519fb9b 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -62,6 +62,8 @@ type MessageEntry = { info: { id: string; role: "user" | "assistant"; + parentID?: string; + finish?: string; }; parts: Array; }; @@ -102,6 +104,7 @@ const runtimeMock = { promptEchoEvents: [] as Array, closeError: null as Error | null, messages: [] as MessageEntry[], + sessionMessagesCalls: 0, forkMessagesBySession: new Map(), forkPreservesBoundary: true, subscribedEvents: [] as Array>, @@ -166,6 +169,7 @@ const runtimeMock = { this.state.promptEchoEvents.length = 0; this.state.closeError = null; this.state.messages = []; + this.state.sessionMessagesCalls = 0; this.state.forkMessagesBySession.clear(); this.state.forkPreservesBoundary = true; this.state.subscribedEvents = []; @@ -406,10 +410,13 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { runtimeMock.state.summarizeCalls.push(input); return { data: true }; }, - messages: async ({ sessionID }: { sessionID: string }) => ({ - data: - runtimeMock.state.forkMessagesBySession.get(sessionID) ?? runtimeMock.state.messages, - }), + messages: async ({ sessionID }: { sessionID: string }) => { + runtimeMock.state.sessionMessagesCalls += 1; + return { + data: + runtimeMock.state.forkMessagesBySession.get(sessionID) ?? runtimeMock.state.messages, + }; + }, message: async ({ sessionID, messageID }: { sessionID: string; messageID: string }) => { runtimeMock.state.messageCalls.push({ sessionID, messageID }); if (runtimeMock.state.messageFailures > 0) { @@ -2194,10 +2201,15 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const threadId = asThreadId("thread-steer-reconnect-before-acceptance"); const firstUserMessageEvent = promiseWithResolvers(); const reconnectEvent = promiseWithResolvers(); + const assistantEvent = promiseWithResolvers(); const steerStarted = promiseWithResolvers(); const steerRelease = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [firstUserMessageEvent.promise, reconnectEvent.promise]; + runtimeMock.state.subscribedEvents = [ + firstUserMessageEvent.promise, + reconnectEvent.promise, + assistantEvent.promise, + ]; runtimeMock.state.promptAsyncImplementation = async () => { if (runtimeMock.state.promptCalls.length === 2) { steerStarted.resolve(undefined); @@ -2262,6 +2274,19 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { steerRelease.resolve(undefined); yield* Fiber.join(steerFiber); yield* advanceTestClock(250); + assistantEvent.resolve({ + id: "evt-assistant-after-reconnect-steer", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-assistant-after-reconnect-steer", + role: "assistant", + parentID: steerMessageId, + finish: "stop", + }, + }, + }); const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), @@ -3027,13 +3052,18 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const threadId = asThreadId("thread-stale-admission-status-after-stop"); const idleEvent = promiseWithResolvers(); const userMessageEvent = promiseWithResolvers(); + const assistantEvent = promiseWithResolvers(); const staleStatusStarted = promiseWithResolvers(); const staleStatusRelease = promiseWithResolvers(); const staleStatusReturned = promiseWithResolvers(); const activePromptStarted = promiseWithResolvers(); const activePromptRelease = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; + runtimeMock.state.subscribedEvents = [ + idleEvent.promise, + userMessageEvent.promise, + assistantEvent.promise, + ]; runtimeMock.state.sessionStatusImplementation = async () => { if (runtimeMock.state.sessionStatusCalls === 1) { staleStatusStarted.resolve(undefined); @@ -3124,6 +3154,19 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { activePromptRelease.resolve(undefined); const activeTurn = yield* Fiber.join(activeTurnFiber); yield* advanceTestClock(250); + assistantEvent.resolve({ + id: "evt-assistant-after-stale-status", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-assistant-after-stale-status", + role: "assistant", + parentID: activeMessageId, + finish: "stop", + }, + }, + }); const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), @@ -3320,14 +3363,19 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("reconciles a sole idle when the matching prompt echo arrives later", () => + it.effect("reconciles a length-limited reply when the matching prompt echo arrives later", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-idle-before-delayed-echo"); const idleEvent = promiseWithResolvers(); const userMessageEvent = promiseWithResolvers(); + const assistantEvent = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; + runtimeMock.state.subscribedEvents = [ + idleEvent.promise, + userMessageEvent.promise, + assistantEvent.promise, + ]; runtimeMock.state.sessionStatusImplementation = async () => ({ data: {} }); const completedFiber = yield* adapter.streamEvents.pipe( @@ -3367,6 +3415,25 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }); yield* advanceTestClock(250); + const waiting = (yield* adapter.listSessions()).find( + (candidate) => candidate.threadId === threadId, + ); + NodeAssert.equal(waiting?.status, "running"); + assistantEvent.resolve({ + id: "evt-delayed-assistant-response", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-delayed-assistant", + role: "assistant", + parentID: messageId, + finish: "length", + }, + }, + }); + yield* advanceTestClock(2_000); + const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), ); @@ -3440,15 +3507,15 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { NodeAssert.equal(runtimeMock.state.sessionStatusCalls > 0, true); const sessions = yield* adapter.listSessions(); const session = sessions.find((candidate) => candidate.threadId === threadId); - NodeAssert.equal(session?.status, "ready"); - NodeAssert.equal(session?.activeTurnId, undefined); + NodeAssert.equal(session?.status, "running"); + NodeAssert.equal(session?.activeTurnId, activeTurn.turnId); NodeAssert.notEqual(activeTurn.turnId, stoppedTurn.turnId); yield* adapter.stopSession(threadId); }), ); - it.effect("reconciles a sole idle after a stop when the exact prompt echo arrives", () => + it.effect("keeps the turn running when an idle precedes the exact prompt echo", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-idle-before-exact-echo-after-stop"); @@ -3457,12 +3524,6 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { runtimeMock.state.autoPromptEcho = false; runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; runtimeMock.state.sessionStatusImplementation = async () => ({ data: {} }); - - const completedFiber = yield* adapter.streamEvents.pipe( - Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), - Stream.runHead, - Effect.forkChild, - ); yield* adapter.startSession({ provider: ProviderDriverKind.make("opencode"), threadId, @@ -3514,14 +3575,11 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }); yield* advanceTestClock(250); - const completed = Option.getOrUndefined( - yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), + const waiting = (yield* adapter.listSessions()).find( + (candidate) => candidate.threadId === threadId, ); - NodeAssert.equal(completed?.turnId, activeTurn.turnId); - const sessions = yield* adapter.listSessions(); - const session = sessions.find((candidate) => candidate.threadId === threadId); - NodeAssert.equal(session?.status, "ready"); - NodeAssert.equal(session?.activeTurnId, undefined); + NodeAssert.equal(waiting?.status, "running"); + NodeAssert.equal(waiting?.activeTurnId, activeTurn.turnId); yield* adapter.stopSession(threadId); }), @@ -3533,10 +3591,15 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const threadId = asThreadId("thread-idle-and-echo-before-acceptance-after-stop"); const idleEvent = promiseWithResolvers(); const userMessageEvent = promiseWithResolvers(); + const assistantEvent = promiseWithResolvers(); const activePromptStarted = promiseWithResolvers(); const activePromptRelease = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; + runtimeMock.state.subscribedEvents = [ + idleEvent.promise, + userMessageEvent.promise, + assistantEvent.promise, + ]; runtimeMock.state.sessionStatusImplementation = async () => ({ data: {} }); runtimeMock.state.promptAsyncImplementation = async () => { if (runtimeMock.state.promptCalls.length === 2) { @@ -3602,6 +3665,25 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const activeTurn = yield* Fiber.join(activeTurnFiber); yield* advanceTestClock(250); + const waiting = (yield* adapter.listSessions()).find( + (candidate) => candidate.threadId === threadId, + ); + NodeAssert.equal(waiting?.activeTurnId, activeTurn.turnId); + assistantEvent.resolve({ + id: "evt-assistant-after-held-acceptance", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-after-held-acceptance", + role: "assistant", + parentID: activeMessageId, + finish: "stop", + }, + }, + }); + yield* advanceTestClock(2_000); + const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), ); @@ -7024,6 +7106,108 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect( + "marks tool-call continuation text as progress while keeping length and stop output visible", + () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-progress-text"); + const sessionID = "http://127.0.0.1:9999/session"; + runtimeMock.state.subscribedEvents = (["tool-calls", "length", "stop"] as const).flatMap( + (finish, index) => { + const messageID = `msg-progress-${index}`; + return [ + { + type: "message.updated", + properties: { sessionID, info: { id: messageID, role: "assistant" } }, + }, + { + type: "message.part.updated", + properties: { + sessionID, + part: { + id: `part-progress-${index}`, + messageID, + sessionID, + type: "text", + text: `Text ${index}`, + time: { start: index + 1, end: index + 2 }, + }, + }, + }, + { + type: "message.updated", + properties: { sessionID, info: { id: messageID, role: "assistant", finish } }, + }, + ...(index === 0 + ? [ + { + type: "message.part.updated", + properties: { + sessionID, + part: { + id: `part-progress-${index}`, + messageID, + sessionID, + type: "text", + text: "Text 0 updated", + time: { start: 1, end: 2 }, + }, + }, + }, + ] + : []), + ]; + }, + ); + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "content.delta" || event.type === "item.completed"), + ), + Stream.take(8), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const completions = events.filter((event) => event.type === "item.completed"); + NodeAssert.deepEqual( + completions.map((event) => + event.type === "item.completed" ? [event.itemId, event.payload.presentation] : [], + ), + [ + ["part-progress-0", undefined], + ["part-progress-0", "progress"], + ["part-progress-1", undefined], + ["part-progress-2", undefined], + ], + ); + NodeAssert.deepEqual( + events + .filter((event) => event.type === "content.delta") + .map((event) => + event.type === "content.delta" + ? [event.itemId, event.payload.streamKind, event.payload.delta] + : [], + ), + [ + ["part-progress-0", "assistant_text", "Text 0"], + ["part-progress-0", "assistant_progress_text", " updated"], + ["part-progress-1", "assistant_text", "Text 1"], + ["part-progress-2", "assistant_text", "Text 2"], + ], + ); + }), + ); + it.effect("emits tool lifecycle events before late assistant metadata", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; @@ -7151,6 +7335,218 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("flushes retained assistant text before completing its turn", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-terminal-retained-text"); + const sessionID = "http://127.0.0.1:9999/session"; + const enqueue = makeOpenCodeEventQueue(); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "content.delta" || + event.type === "item.completed" || + event.type === "turn.completed"), + ), + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.sendTurn({ + threadId, + input: "Finish the report", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const promptMessageId = (runtimeMock.state.promptCalls.at(-1) as { messageID: string }) + .messageID; + enqueue({ type: "session.status", properties: { sessionID, status: { type: "busy" } } }); + enqueue({ + type: "message.part.updated", + properties: { + sessionID, + part: { + id: "part-retained-report", + messageID: "msg-retained-report", + sessionID, + type: "text", + text: "Final report", + time: { start: 1, end: 2 }, + }, + }, + }); + enqueue({ + type: "message.updated", + properties: { + sessionID, + info: { + id: "msg-retained-report", + role: "assistant", + parentID: promptMessageId, + finish: "stop", + }, + }, + }); + enqueue({ type: "session.status", properties: { sessionID, status: { type: "idle" } } }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + NodeAssert.deepEqual( + events.map((event) => event.type), + ["content.delta", "item.completed", "turn.completed"], + ); + if (events[0]?.type === "content.delta") { + NodeAssert.equal(events[0].payload.delta, "Final report"); + } + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("keeps late final reports with the prompt's completed turn", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-late-final-report"); + const sessionID = "http://127.0.0.1:9999/session"; + const enqueue = makeOpenCodeEventQueue(); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const completedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.runHead, + Effect.forkChild, + ); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Audit this branch", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const promptMessageId = (runtimeMock.state.promptCalls.at(-1) as { messageID: string }) + .messageID; + enqueue({ type: "session.status", properties: { sessionID, status: { type: "busy" } } }); + enqueue({ + type: "message.updated", + properties: { + sessionID, + info: { + id: "msg-first-report", + role: "assistant", + parentID: promptMessageId, + finish: "stop", + }, + }, + }); + enqueue({ type: "session.status", properties: { sessionID, status: { type: "idle" } } }); + const completed = Option.getOrUndefined( + yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(completed?.turnId, turn.turnId); + + const nextTurn = yield* adapter.sendTurn({ + threadId, + input: "Continue the audit", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const lateDeltaFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + event.type === "content.delta" && + event.itemId === "part-late-report", + ), + Stream.runHead, + Effect.forkChild, + ); + enqueue({ + type: "message.updated", + properties: { + sessionID, + info: { + id: "msg-late-report", + role: "assistant", + parentID: promptMessageId, + finish: "stop", + }, + }, + }); + enqueue({ + type: "message.part.updated", + properties: { + sessionID, + part: { + id: "part-late-report", + messageID: "msg-late-report", + sessionID, + type: "text", + text: "Revised audit", + time: { start: 1, end: 2 }, + }, + }, + }); + const lateDelta = Option.getOrUndefined( + yield* Fiber.join(lateDeltaFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(lateDelta?.turnId, turn.turnId); + NodeAssert.notEqual(lateDelta?.turnId, nextTurn.turnId); + const unassociatedDeltaFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + event.type === "content.delta" && + event.itemId === "part-unassociated-report", + ), + Stream.runHead, + Effect.forkChild, + ); + enqueue({ + type: "message.updated", + properties: { + sessionID, + info: { + id: "msg-unassociated-report", + role: "assistant", + parentID: "evicted-prompt-id", + finish: "stop", + }, + }, + }); + enqueue({ + type: "message.part.updated", + properties: { + sessionID, + part: { + id: "part-unassociated-report", + messageID: "msg-unassociated-report", + sessionID, + type: "text", + text: "Earlier report", + time: { start: 1, end: 2 }, + }, + }, + }); + const unassociatedDelta = Option.getOrUndefined( + yield* Fiber.join(unassociatedDeltaFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(unassociatedDelta?.turnId, undefined); + yield* adapter.stopSession(threadId); + }), + ); + it.effect("processes late assistant metadata without visiting completed turns", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; @@ -7177,9 +7573,23 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { "opencode/kimi-k3", ), }); + const promptMessageId = (runtimeMock.state.promptCalls.at(-1) as { messageID: string }) + .messageID; + enqueue({ + type: "session.status", + properties: { sessionID, status: { type: "busy" } }, + }); enqueue({ type: "message.updated", - properties: { sessionID, info: { id: `history-message-${index}`, role: "assistant" } }, + properties: { + sessionID, + info: { + id: `history-message-${index}`, + role: "assistant", + parentID: promptMessageId, + finish: "stop", + }, + }, }); enqueue({ type: "message.part.updated", @@ -7282,6 +7692,7 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { "opencode/kimi-k3", ), }); + enqueue({ type: "session.status", properties: { sessionID, status: { type: "busy" } } }); for (const parentID of ["", promptMessageId]) { enqueue({ type: "message.updated", @@ -7612,6 +8023,78 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("recovers a finished assistant message whose terminal event was lost", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-reconnect-lost-terminal-message"); + const reconnect = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [reconnect.promise]; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Finish during reconnect", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const promptMessageId = (runtimeMock.state.promptCalls[0] as { messageID: string }).messageID; + runtimeMock.state.messages.push( + { + info: { + id: "assistant-from-previous-turn", + role: "assistant", + parentID: "other-prompt", + finish: "stop", + }, + parts: [], + }, + { + info: { + id: "assistant-finished-during-reconnect", + role: "assistant", + parentID: promptMessageId, + finish: "length", + }, + parts: [], + }, + ); + const warningFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "runtime.warning"), + Stream.runHead, + Effect.forkChild, + ); + const completedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.runHead, + Effect.forkChild, + ); + runtimeMock.state.eventStreamError?.(new Error("socket closed")); + yield* Fiber.join(warningFiber); + reconnect.resolve({ + id: "evt-reconnected-after-finished-message", + type: "server.connected", + properties: {}, + } satisfies OpenCodeEvent); + yield* advanceTestClock(6_000); + const completed = Option.getOrUndefined( + yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(completed?.turnId, turn.turnId); + NodeAssert.equal(runtimeMock.state.sessionMessagesCalls, 1); + NodeAssert.equal( + (yield* adapter.listSessions()).find((candidate) => candidate.threadId === threadId) + ?.status, + "ready", + ); + yield* adapter.stopSession(threadId); + }), + ); + it.effect( "ends a running session on clean stream closure without discarding unresolved permissions", () => diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 34ebd1b544fb..456b11071305 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -213,6 +213,7 @@ interface OpenCodeIdleReconciliation { } interface OpenCodePromptAdmission { + steering: boolean; readonly generation: number; readonly turnId: TurnId; readonly messageId: string; @@ -222,6 +223,8 @@ interface OpenCodePromptAdmission { idleDuringAdmission: { readonly turnId: TurnId; readonly raw: unknown } | undefined; idleObservedAfterMessage: boolean; messageObserved: boolean; + assistantResponseFinished: boolean; + responseReceipt: Deferred.Deferred; busyObserved: boolean; idleStatusConfirmations: number; accepted: boolean; @@ -332,6 +335,7 @@ type OpenCodeTextPartState = Pick, "id" | "tokens">; @@ -350,6 +354,7 @@ interface OpenCodeSessionContext { readonly pendingPermissions: Map; readonly pendingQuestions: Map; readonly messageRoleById: Map; + readonly messageTurnIdById: Map; // OpenCode permits edits to completed parts. Keep text for snapshot comparison // until native removal or session teardown, but do not retain other part payloads. readonly textPartsByMessageId: Map>; @@ -597,8 +602,14 @@ function normalizeQuestionRequest(request: QuestionRequest): ReadonlyArray): "assistant_text" | "reasoning_text" { - return part.type === "reasoning" ? "reasoning_text" : "assistant_text"; +function resolveTextStreamKind( + part: Pick, +): "assistant_text" | "assistant_progress_text" | "reasoning_text" { + return part.type === "reasoning" + ? "reasoning_text" + : part.progressClassified + ? "assistant_progress_text" + : "assistant_text"; } function retainOpenCodeTextPart( @@ -616,6 +627,7 @@ function retainOpenCodeTextPart( ...(part.time !== undefined ? { time: part.time } : {}), emittedText: previous?.emittedText, completed: previous?.completed ?? false, + progressClassified: previous?.progressClassified ?? false, }; parts.set(part.id, state); context.textPartsByMessageId.set(part.messageID, parts); @@ -687,6 +699,21 @@ function messageRoleForPart( return part.type === "tool" ? "assistant" : undefined; } +const MAX_RECENT_MESSAGE_TURN_IDS = 4_096; + +function rememberMessageTurnId( + context: OpenCodeSessionContext, + messageId: string, + turnId: TurnId | null, +) { + const associations = context.messageTurnIdById; + associations.delete(messageId); + associations.set(messageId, turnId); + if (associations.size > MAX_RECENT_MESSAGE_TURN_IDS) { + associations.delete(associations.keys().next().value!); + } +} + function detailFromToolPart(part: Extract): string | undefined { switch (part.state.status) { case "completed": @@ -873,12 +900,26 @@ const cancelPendingOpenCodePrompt = Effect.fn("cancelPendingOpenCodePrompt")(fun return; } admission.cancelled = true; + if (admission.recoveryFiber) { + yield* Fiber.interrupt(admission.recoveryFiber); + } if (admission.promptFiber) { yield* Fiber.interrupt(admission.promptFiber); } yield* Deferred.await(admission.submissionSettled); }); +const retirePromptAdmission = Effect.fnUntraced(function* ( + context: OpenCodeSessionContext, + admission: OpenCodePromptAdmission, +) { + admission.cancelled = true; + context.promptAdmission = undefined; + if (admission.recoveryFiber) { + yield* Fiber.interrupt(admission.recoveryFiber); + } +}); + const closeStartingOpenCodeContext = Effect.fn("closeStartingOpenCodeContext")(function* ( context: OpenCodeSessionContext, abortRemote: boolean, @@ -1357,6 +1398,7 @@ export function makeOpenCodeAdapter( if (!promptAdmission.requiresMessageReceipt) { yield* Deferred.await(promptAdmission.acceptance); } + let lastStatusIsIdle = false; for ( let retryCount = 0; retryCount < 5 || (promptAdmission.requiresMessageReceipt && !promptAdmission.accepted); @@ -1433,6 +1475,7 @@ export function makeOpenCodeAdapter( const status = statusData?.[context.openCodeSessionId]; const isIdle = statusData !== undefined && (status === undefined || status.type === "idle"); + lastStatusIsIdle = isIdle; const isBusy = status?.type === "busy" || status?.type === "retry"; if (isBusy) { promptAdmission.busyObserved = true; @@ -1446,14 +1489,14 @@ export function makeOpenCodeAdapter( if ( isIdle && idle !== undefined && - (promptAdmission.messageObserved || promptAdmission.busyObserved) + (promptAdmission.assistantResponseFinished || promptAdmission.busyObserved) ) { context.promptAdmission = undefined; context.awaitingBusyAfterInterruption = false; yield* scheduleIdleReconciliation(context, promptAdmission.turnId, idle.raw); return; } - if (isIdle && promptAdmission.messageObserved) { + if (isIdle && promptAdmission.assistantResponseFinished) { promptAdmission.idleStatusConfirmations += 1; if (promptAdmission.idleStatusConfirmations >= 2) { context.promptAdmission = undefined; @@ -1474,7 +1517,7 @@ export function makeOpenCodeAdapter( } if ( isIdle && - promptAdmission.messageObserved && + promptAdmission.assistantResponseFinished && promptAdmission.recoveryRaw !== undefined ) { context.promptAdmission = undefined; @@ -1490,6 +1533,61 @@ export function makeOpenCodeAdapter( const delayMs = Math.min(250 * 2 ** retryCount, 2_000); yield* Effect.sleep(`${delayMs} millis`); } + // An idle status can precede OpenCode's first assistant response. Keep + // the turn owned until that response arrives; its terminal update + // restarts reconciliation without polling an idle session indefinitely. + if (promptAdmission.messageObserved && !promptAdmission.assistantResponseFinished) { + // The terminal assistant event may have been lost during a reconnect. + // Confirm it from OpenCode's durable messages before waiting for the stream. + if (lastStatusIsIdle) { + const messages = yield* runOpenCodeSdk("session.messages", (signal) => + context.client.session.messages({ sessionID: context.openCodeSessionId }, { signal }), + ).pipe(Effect.timeout("1 second"), Effect.option); + if ( + (yield* Ref.get(context.stopped)) || + sessions.get(context.session.threadId) !== context || + context.promptAdmission !== promptAdmission || + context.activeTurnId !== promptAdmission.turnId || + context.promptGeneration !== promptAdmission.generation || + promptAdmission.cancelled + ) { + return; + } + const finished = + Option.isSome(messages) && + messages.value.data?.some( + (message) => + message.info.role === "assistant" && + message.info.parentID === promptAdmission.messageId && + (message.info.finish === "stop" || message.info.finish === "length"), + ); + if (finished) { + promptAdmission.assistantResponseFinished = true; + yield* Deferred.succeed(promptAdmission.responseReceipt, undefined); + } + } + const response = yield* Deferred.await(promptAdmission.responseReceipt).pipe( + Effect.timeoutOption("2 minutes"), + ); + if (Option.isNone(response)) { + yield* failPromptAdmissionRecovery(context, promptAdmission); + return; + } + if ( + context.promptAdmission === promptAdmission && + context.activeTurnId === promptAdmission.turnId && + context.promptGeneration === promptAdmission.generation && + !promptAdmission.cancelled + ) { + context.promptAdmission = undefined; + yield* scheduleIdleReconciliation( + context, + promptAdmission.turnId, + promptAdmission.recoveryRaw, + ); + } + return; + } yield* failPromptAdmissionRecovery(context, promptAdmission); }).pipe( Effect.ignoreCause, @@ -1613,6 +1711,34 @@ export function makeOpenCodeAdapter( yield* Scope.close(context.sessionScope, Exit.void); }); + const emitAssistantProgressCompletion = Effect.fn("emitAssistantProgressCompletion")(function* ( + context: OpenCodeSessionContext, + part: OpenCodeTextPartState, + turnId: TurnId | undefined, + raw: unknown, + ) { + if (part.text === undefined || part.text.trim().length === 0) return; + part.progressClassified = true; + part.completed = true; + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + itemId: part.id, + createdAt: part.time?.end !== undefined ? isoFromEpochMs(part.time.end) : undefined, + raw, + })), + type: "item.completed", + payload: { + itemType: "assistant_message", + status: "completed", + presentation: "progress", + title: "Assistant message", + detail: part.text, + }, + }); + }); + /** Emit content.delta and item.completed events for an assistant text part. */ const emitAssistantTextDelta = Effect.fn("emitAssistantTextDelta")(function* ( context: OpenCodeSessionContext, @@ -2264,7 +2390,34 @@ export function makeOpenCodeAdapter( return; } - const turnId = context.activeTurnId; + if (event.type === "message.updated" && event.properties.info.role === "assistant") { + const parentId = event.properties.info.parentID; + const parentTurnId = parentId ? context.messageTurnIdById.get(parentId) : undefined; + if (parentTurnId) { + rememberMessageTurnId(context, event.properties.info.id, parentTurnId); + } else if (parentId && context.turnTokenUsage?.promptMessageIds.has(parentId)) { + if (context.activeTurnId) { + rememberMessageTurnId(context, event.properties.info.id, context.activeTurnId); + } + } else if (parentId && context.turnTokenUsage !== undefined) { + // An unknown parent may be an old prompt whose mapping was evicted. + // Do not attach its late response to the current turn. + rememberMessageTurnId(context, event.properties.info.id, null); + } + } + const assistantMessageId = + event.type === "message.updated" && event.properties.info.role === "assistant" + ? event.properties.info.id + : event.type === "message.part.updated" + ? event.properties.part.messageID + : event.type === "message.part.delta" + ? event.properties.messageID + : undefined; + const associatedTurnId = assistantMessageId + ? context.messageTurnIdById.get(assistantMessageId) + : undefined; + const turnId = + associatedTurnId === null ? undefined : (associatedTurnId ?? context.activeTurnId); yield* writeNativeEventBestEffort(context.session.threadId, { observedAt: yield* nowIso, event: { @@ -2336,15 +2489,7 @@ export function makeOpenCodeAdapter( promptAdmission.messageObserved = true; yield* Deferred.succeed(promptAdmission.messageReceipt, undefined); if (promptAdmission.accepted) { - const idle = promptAdmission.idleDuringAdmission; - context.awaitingBusyAfterInterruption = false; - context.promptAdmission = undefined; - if (promptAdmission.recoveryFiber) { - yield* Fiber.interrupt(promptAdmission.recoveryFiber); - } - if (idle) { - yield* scheduleIdleReconciliation(context, idle.turnId, idle.raw); - } + yield* schedulePromptAdmissionRecovery(context, event); } } context.messageRoleById.set(event.properties.info.id, event.properties.info.role); @@ -2352,6 +2497,11 @@ export function makeOpenCodeAdapter( context.textPartsByMessageId.delete(event.properties.info.id); } if (event.properties.info.role === "assistant") { + const finishesPromptResponse = + promptAdmission !== undefined && + promptAdmission.messageId === event.properties.info.parentID && + (event.properties.info.finish === "stop" || + event.properties.info.finish === "length"); const usage = context.turnTokenUsage; const parentMessageId = typeof event.properties.info.parentID === "string" && @@ -2387,6 +2537,25 @@ export function makeOpenCodeAdapter( .get(event.properties.info.id) ?.values() ?? []) { yield* emitAssistantTextDelta(context, part, turnId, event); + if ( + part.type === "text" && + !part.progressClassified && + event.properties.info.finish === "tool-calls" + ) { + yield* emitAssistantProgressCompletion(context, part, turnId, event); + } + } + if (finishesPromptResponse) { + promptAdmission.assistantResponseFinished = true; + yield* Deferred.succeed(promptAdmission.responseReceipt, undefined); + if (promptAdmission.accepted) { + if (promptAdmission.recoveryFiber) { + yield* Fiber.interrupt(promptAdmission.recoveryFiber); + } + context.promptAdmission = undefined; + context.awaitingBusyAfterInterruption = false; + yield* scheduleIdleReconciliation(context, promptAdmission.turnId, event); + } } } break; @@ -2594,8 +2763,13 @@ export function makeOpenCodeAdapter( yield* cancelIdleReconciliation(context); context.awaitingBusyAfterInterruption = false; if (context.promptAdmission?.turnId === turnId) { - context.promptAdmission.busyObserved = true; - yield* schedulePromptAdmissionRecovery(context, event); + const admission = context.promptAdmission; + admission.busyObserved = true; + if (admission.accepted && admission.messageObserved) { + yield* retirePromptAdmission(context, admission); + } else { + yield* schedulePromptAdmissionRecovery(context, event); + } } yield* updateProviderSession(context, { status: "running", @@ -2625,9 +2799,19 @@ export function makeOpenCodeAdapter( break; } if (context.promptAdmission?.turnId === turnId) { - context.promptAdmission.idleDuringAdmission = { turnId, raw: event }; - context.promptAdmission.idleObservedAfterMessage = - context.promptAdmission.messageObserved; + const admission = context.promptAdmission; + const priorIdle = admission.idleDuringAdmission ?? admission.priorIdle; + admission.idleDuringAdmission = { turnId, raw: event }; + admission.idleObservedAfterMessage = admission.messageObserved; + if ( + admission.accepted && + admission.messageObserved && + (priorIdle || !context.awaitingBusyAfterInterruption) + ) { + yield* retirePromptAdmission(context, admission); + yield* scheduleIdleReconciliation(context, turnId, event); + break; + } yield* schedulePromptAdmissionRecovery(context, event); break; } @@ -2662,6 +2846,10 @@ export function makeOpenCodeAdapter( break; } } + const promptAdmission = context.promptAdmission; + if (promptAdmission) { + yield* retirePromptAdmission(context, promptAdmission); + } yield* cancelIdleReconciliation(context); const terminalCancellation = activeTurnId !== undefined && cancellation?.turnId === activeTurnId @@ -3018,6 +3206,7 @@ export function makeOpenCodeAdapter( pendingQuestions: new Map(), textPartsByMessageId: new Map(), messageRoleById: new Map(), + messageTurnIdById: new Map(), turnTokenUsage: undefined, activeTurnId: undefined, activeAgent: undefined, @@ -3178,7 +3367,12 @@ export function makeOpenCodeAdapter( : undefined; context.pendingIdleReconciliation = undefined; const promptGeneration = context.promptGeneration + 1; + const previousAdmission = context.promptAdmission; + if (previousAdmission?.recoveryFiber) { + yield* Fiber.interrupt(previousAdmission.recoveryFiber); + } const promptAdmission: OpenCodePromptAdmission = { + steering: steeringTurnId !== undefined, generation: promptGeneration, turnId, messageId, @@ -3188,6 +3382,8 @@ export function makeOpenCodeAdapter( idleDuringAdmission: undefined, idleObservedAfterMessage: false, messageObserved: false, + assistantResponseFinished: false, + responseReceipt: Deferred.makeUnsafe(), busyObserved: false, idleStatusConfirmations: 0, accepted: false, @@ -3201,6 +3397,7 @@ export function makeOpenCodeAdapter( context.promptAdmission = promptAdmission; context.activeTurnId = turnId; + rememberMessageTurnId(context, messageId, turnId); if (steeringTurnId === undefined) { context.turnTokenUsage = makeOpenCodeTurnTokenUsageAccumulator(); } @@ -3491,13 +3688,29 @@ export function makeOpenCodeAdapter( ) { context.awaitingBusyAfterInterruption = false; const idle = promptAdmission.idleDuringAdmission; - if (idle && !promptAdmission.idleObservedAfterMessage) { - yield* schedulePromptAdmissionRecovery(context, idle.raw); - } else { + if (promptAdmission.busyObserved && promptAdmission.messageObserved) { + if (promptAdmission.recoveryFiber) { + yield* Fiber.interrupt(promptAdmission.recoveryFiber); + } + context.promptAdmission = undefined; + if (idle) { + yield* scheduleIdleReconciliation(context, turnId, idle.raw); + } + } else if (promptAdmission.assistantResponseFinished) { + if (promptAdmission.recoveryFiber) { + yield* Fiber.interrupt(promptAdmission.recoveryFiber); + } + context.promptAdmission = undefined; + yield* scheduleIdleReconciliation(context, turnId, promptAdmission.recoveryRaw); + } else if ( + idle && + promptAdmission.idleObservedAfterMessage && + promptAdmission.steering + ) { context.promptAdmission = undefined; - } - if (idle && promptAdmission.idleObservedAfterMessage) { yield* scheduleIdleReconciliation(context, turnId, idle.raw); + } else { + yield* schedulePromptAdmissionRecovery(context, idle?.raw); } } else { yield* schedulePromptAdmissionRecovery(context, promptAdmission.recoveryRaw); @@ -3636,11 +3849,7 @@ export function makeOpenCodeAdapter( context.cancellation = cancellation; const promptAdmission = context.promptAdmission; if (promptAdmission !== undefined && promptAdmission.turnId === interruptedTurnId) { - promptAdmission.cancelled = true; - if (promptAdmission.promptFiber) { - yield* Fiber.interrupt(promptAdmission.promptFiber); - } - yield* Deferred.await(promptAdmission.submissionSettled); + yield* cancelPendingOpenCodePrompt(context); } yield* Effect.forEach([...context.commandFibers], Fiber.interrupt, { discard: true }); @@ -3987,6 +4196,7 @@ export function makeOpenCodeAdapter( context.relatedSessionIds.clear(); context.relatedSessionIds.add(forkedSessionId); context.messageRoleById.clear(); + context.messageTurnIdById.clear(); context.textPartsByMessageId.clear(); context.turnTokenUsage = undefined; context.activeTurnId = undefined; diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 0330d3dd8bd9..2efc9fd6966d 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -615,6 +615,136 @@ describe("applyThreadDetailEvent", () => { } }); + it("updates a streamed assistant message to a progress summary", () => { + const messageId = MessageId.make("assistant:progress-part"); + const thread = { + ...baseThread, + latestTurn: { + turnId: TurnId.make("turn-progress"), + state: "running" as const, + requestedAt: baseThread.createdAt, + startedAt: baseThread.createdAt, + completedAt: null, + assistantMessageId: messageId, + }, + checkpoints: [ + { + turnId: TurnId.make("turn-progress"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("ref-progress"), + status: "ready" as const, + files: [], + assistantMessageId: messageId, + completedAt: baseThread.updatedAt, + }, + ], + messages: [ + { + id: messageId, + role: "assistant" as const, + text: "Checking the reviews.", + turnId: TurnId.make("turn-progress"), + streaming: false, + createdAt: baseThread.createdAt, + updatedAt: baseThread.updatedAt, + }, + ], + }; + const reclassificationEvent = { + ...baseEventFields, + sequence: 6, + occurredAt: baseThread.updatedAt, + aggregateKind: "thread", + aggregateId: baseThread.id, + type: "thread.message-sent", + payload: { + threadId: baseThread.id, + messageId, + role: "reasoning", + text: "", + turnId: TurnId.make("turn-progress"), + streaming: false, + createdAt: baseThread.createdAt, + updatedAt: baseThread.updatedAt, + }, + } as const; + const turnlessResult = applyThreadDetailEvent(thread, { + ...reclassificationEvent, + payload: { ...reclassificationEvent.payload, turnId: null }, + }); + expect(turnlessResult.kind).toBe("updated"); + if (turnlessResult.kind === "updated") { + expect(turnlessResult.thread.latestTurn?.assistantMessageId).toBe(messageId); + expect(turnlessResult.thread.checkpoints[0]?.assistantMessageId).toBe(messageId); + } + const result = applyThreadDetailEvent(thread, reclassificationEvent); + + expect(result.kind).toBe("updated"); + if (result.kind !== "updated") return; + expect(result.thread.messages[0]).toMatchObject({ + id: messageId, + role: "reasoning", + text: "Checking the reviews.", + }); + expect(result.thread.latestTurn?.assistantMessageId).toBeNull(); + expect(result.thread.checkpoints[0]?.assistantMessageId).toBeNull(); + }); + + it("does not rebind a reclassified answer to a message beyond the server cap", () => { + const turnId = TurnId.make("turn-capped"); + const oldAnswerId = MessageId.make("old-answer"); + const currentAnswerId = MessageId.make("current-answer"); + const message = (id: MessageId, role: "user" | "assistant") => ({ + id, + role, + text: "text", + turnId: role === "assistant" ? turnId : null, + streaming: false, + createdAt: baseThread.createdAt, + updatedAt: baseThread.updatedAt, + }); + const thread: OrchestrationThread = { + ...baseThread, + latestTurn: { + turnId, + state: "completed", + requestedAt: baseThread.createdAt, + startedAt: baseThread.createdAt, + completedAt: baseThread.updatedAt, + assistantMessageId: currentAnswerId, + }, + messages: [ + message(oldAnswerId, "assistant"), + ...Array.from({ length: 1_999 }, (_, index) => + message(MessageId.make(`filler-${index}`), "user"), + ), + message(currentAnswerId, "assistant"), + ], + }; + const result = applyThreadDetailEvent(thread, { + ...baseEventFields, + sequence: 6, + occurredAt: baseThread.updatedAt, + aggregateKind: "thread", + aggregateId: baseThread.id, + type: "thread.message-sent", + payload: { + threadId: baseThread.id, + messageId: currentAnswerId, + role: "reasoning", + text: "", + turnId, + streaming: false, + createdAt: baseThread.createdAt, + updatedAt: baseThread.updatedAt, + }, + }); + expect(result.kind).toBe("updated"); + if (result.kind !== "updated") return; + expect(result.thread.messages).toHaveLength(2_000); + expect(result.thread.latestTurn?.assistantMessageId).toBeNull(); + }); + it("appends a new message", () => { const result = applyThreadDetailEvent(baseThread, { ...baseEventFields, diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index 66f78a0464ec..3df1a0c57a8a 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -14,7 +14,7 @@ import type { TurnId, } from "@t3tools/contracts"; import { threadPullRequestKeysEqual } from "@t3tools/shared/threadPullRequests"; -import { isImportedAgentSessionMessageId } from "@t3tools/contracts"; +import { isImportedAgentSessionMessageId, MAX_THREAD_MESSAGES } from "@t3tools/contracts"; import { compareDateTimeStrings } from "@t3tools/shared/dateTime"; export type ThreadDetailReducerResult = @@ -407,6 +407,7 @@ export function applyThreadDetailEvent( found = true; return { ...entry, + role: message.role, text: message.streaming ? `${entry.text}${message.text}` : message.text.length > 0 @@ -420,6 +421,16 @@ export function applyThreadDetailEvent( }; }); if (!found) messages.push(message); + const cappedMessages = messages.slice(-MAX_THREAD_MESSAGES); + const reclassifiedAssistant = + message.role === "reasoning" && + message.turnId !== null && + thread.messages.some((entry) => entry.id === message.id && entry.role === "assistant"); + const replacementAssistantMessageId = reclassifiedAssistant + ? (cappedMessages.findLast( + (entry) => entry.role === "assistant" && entry.turnId === message.turnId, + )?.id ?? null) + : null; // 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 @@ -431,7 +442,7 @@ export function applyThreadDetailEvent( thread.session?.status === "running" && thread.session.activeTurnId === event.payload.turnId; const settlesTurn = !event.payload.streaming && !turnStillRunning; - const latestTurn = reuseLatestTurn( + const projectedLatestTurn = reuseLatestTurn( thread.latestTurn, event.payload.role === "assistant" && event.payload.turnId !== null && @@ -465,7 +476,7 @@ export function applyThreadDetailEvent( // Rebind checkpoint assistant message IDs for assistant messages. The // helper hands back the same array when the entry is already bound. - const checkpoints = + const projectedCheckpoints = event.payload.role === "assistant" && event.payload.turnId !== null ? rebindCheckpointAssistantMessage( thread.checkpoints, @@ -473,12 +484,23 @@ export function applyThreadDetailEvent( event.payload.messageId, ) : thread.checkpoints; + const latestTurn = + reclassifiedAssistant && projectedLatestTurn?.assistantMessageId === message.id + ? { ...projectedLatestTurn, assistantMessageId: replacementAssistantMessageId } + : projectedLatestTurn; + const checkpoints = reclassifiedAssistant + ? Arr.map(projectedCheckpoints, (entry) => + entry.assistantMessageId === message.id + ? { ...entry, assistantMessageId: replacementAssistantMessageId } + : entry, + ) + : projectedCheckpoints; return { kind: "updated", thread: { ...thread, - messages, + messages: cappedMessages, checkpoints, latestTurn, updatedAt: event.occurredAt, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 1ae704d3f666..764a2c83193f 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -560,9 +560,10 @@ export const OrchestrationProject = Schema.Struct({ }); export type OrchestrationProject = typeof OrchestrationProject.Type; -/** `reasoning` carries a provider's thinking trace: a reasoning summary, or - * the raw chain of thought when the model exposes one. It is a sibling of the - * assistant text it precedes, not a replacement for it. */ +export const MAX_THREAD_MESSAGES = 2_000; + +/** `reasoning` carries a provider's thinking summary, raw trace, or OpenCode + * continuation text. It sits alongside the assistant answer. */ export const OrchestrationMessageRole = Schema.Literals([ "user", "assistant", diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index aa2b3b7e1580..886617a9c3dc 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -80,8 +80,11 @@ export type RuntimePlanStepStatus = typeof RuntimePlanStepStatus.Type; const RuntimeItemStatus = Schema.Literals(["inProgress", "completed", "failed", "declined"]); export type RuntimeItemStatus = typeof RuntimeItemStatus.Type; +// Providers emit only the kinds they support. OpenCode can later mark streamed +// assistant text as progress; other adapters need not emit that kind. const RuntimeContentStreamKind = Schema.Literals([ "assistant_text", + "assistant_progress_text", "reasoning_text", "reasoning_summary_text", "plan_text", @@ -434,6 +437,7 @@ export const ItemLifecyclePayload = Schema.Struct({ status: Schema.optional(RuntimeItemStatus), title: Schema.optional(TrimmedNonEmptyStringSchema), detail: Schema.optional(TrimmedNonEmptyStringSchema), + presentation: Schema.optional(Schema.Literal("progress")), toolSurface: Schema.optional(ToolActivitySurface), toolIcon: Schema.optional(ToolActivityIcon), toolSource: Schema.optional(ToolActivitySource), From d3dd8816889e646276eb53f77fe497ecd6ffde72 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Fri, 2 Oct 2026 13:00:57 +0200 Subject: [PATCH 2/4] refactor(opencode): defer independent prompt recovery changes --- .../provider/Layers/OpenCodeAdapter.test.ts | 221 +++--------------- .../src/provider/Layers/OpenCodeAdapter.ts | 175 +++----------- 2 files changed, 53 insertions(+), 343 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 47ab3519fb9b..28281fab00d8 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -62,8 +62,6 @@ type MessageEntry = { info: { id: string; role: "user" | "assistant"; - parentID?: string; - finish?: string; }; parts: Array; }; @@ -104,7 +102,6 @@ const runtimeMock = { promptEchoEvents: [] as Array, closeError: null as Error | null, messages: [] as MessageEntry[], - sessionMessagesCalls: 0, forkMessagesBySession: new Map(), forkPreservesBoundary: true, subscribedEvents: [] as Array>, @@ -169,7 +166,6 @@ const runtimeMock = { this.state.promptEchoEvents.length = 0; this.state.closeError = null; this.state.messages = []; - this.state.sessionMessagesCalls = 0; this.state.forkMessagesBySession.clear(); this.state.forkPreservesBoundary = true; this.state.subscribedEvents = []; @@ -410,13 +406,10 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { runtimeMock.state.summarizeCalls.push(input); return { data: true }; }, - messages: async ({ sessionID }: { sessionID: string }) => { - runtimeMock.state.sessionMessagesCalls += 1; - return { - data: - runtimeMock.state.forkMessagesBySession.get(sessionID) ?? runtimeMock.state.messages, - }; - }, + messages: async ({ sessionID }: { sessionID: string }) => ({ + data: + runtimeMock.state.forkMessagesBySession.get(sessionID) ?? runtimeMock.state.messages, + }), message: async ({ sessionID, messageID }: { sessionID: string; messageID: string }) => { runtimeMock.state.messageCalls.push({ sessionID, messageID }); if (runtimeMock.state.messageFailures > 0) { @@ -2201,15 +2194,10 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const threadId = asThreadId("thread-steer-reconnect-before-acceptance"); const firstUserMessageEvent = promiseWithResolvers(); const reconnectEvent = promiseWithResolvers(); - const assistantEvent = promiseWithResolvers(); const steerStarted = promiseWithResolvers(); const steerRelease = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [ - firstUserMessageEvent.promise, - reconnectEvent.promise, - assistantEvent.promise, - ]; + runtimeMock.state.subscribedEvents = [firstUserMessageEvent.promise, reconnectEvent.promise]; runtimeMock.state.promptAsyncImplementation = async () => { if (runtimeMock.state.promptCalls.length === 2) { steerStarted.resolve(undefined); @@ -2274,19 +2262,6 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { steerRelease.resolve(undefined); yield* Fiber.join(steerFiber); yield* advanceTestClock(250); - assistantEvent.resolve({ - id: "evt-assistant-after-reconnect-steer", - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-assistant-after-reconnect-steer", - role: "assistant", - parentID: steerMessageId, - finish: "stop", - }, - }, - }); const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), @@ -3052,18 +3027,13 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const threadId = asThreadId("thread-stale-admission-status-after-stop"); const idleEvent = promiseWithResolvers(); const userMessageEvent = promiseWithResolvers(); - const assistantEvent = promiseWithResolvers(); const staleStatusStarted = promiseWithResolvers(); const staleStatusRelease = promiseWithResolvers(); const staleStatusReturned = promiseWithResolvers(); const activePromptStarted = promiseWithResolvers(); const activePromptRelease = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [ - idleEvent.promise, - userMessageEvent.promise, - assistantEvent.promise, - ]; + runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; runtimeMock.state.sessionStatusImplementation = async () => { if (runtimeMock.state.sessionStatusCalls === 1) { staleStatusStarted.resolve(undefined); @@ -3154,19 +3124,6 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { activePromptRelease.resolve(undefined); const activeTurn = yield* Fiber.join(activeTurnFiber); yield* advanceTestClock(250); - assistantEvent.resolve({ - id: "evt-assistant-after-stale-status", - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-assistant-after-stale-status", - role: "assistant", - parentID: activeMessageId, - finish: "stop", - }, - }, - }); const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), @@ -3363,19 +3320,14 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("reconciles a length-limited reply when the matching prompt echo arrives later", () => + it.effect("reconciles a sole idle when the matching prompt echo arrives later", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-idle-before-delayed-echo"); const idleEvent = promiseWithResolvers(); const userMessageEvent = promiseWithResolvers(); - const assistantEvent = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [ - idleEvent.promise, - userMessageEvent.promise, - assistantEvent.promise, - ]; + runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; runtimeMock.state.sessionStatusImplementation = async () => ({ data: {} }); const completedFiber = yield* adapter.streamEvents.pipe( @@ -3415,25 +3367,6 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }); yield* advanceTestClock(250); - const waiting = (yield* adapter.listSessions()).find( - (candidate) => candidate.threadId === threadId, - ); - NodeAssert.equal(waiting?.status, "running"); - assistantEvent.resolve({ - id: "evt-delayed-assistant-response", - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-delayed-assistant", - role: "assistant", - parentID: messageId, - finish: "length", - }, - }, - }); - yield* advanceTestClock(2_000); - const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), ); @@ -3507,15 +3440,15 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { NodeAssert.equal(runtimeMock.state.sessionStatusCalls > 0, true); const sessions = yield* adapter.listSessions(); const session = sessions.find((candidate) => candidate.threadId === threadId); - NodeAssert.equal(session?.status, "running"); - NodeAssert.equal(session?.activeTurnId, activeTurn.turnId); + NodeAssert.equal(session?.status, "ready"); + NodeAssert.equal(session?.activeTurnId, undefined); NodeAssert.notEqual(activeTurn.turnId, stoppedTurn.turnId); yield* adapter.stopSession(threadId); }), ); - it.effect("keeps the turn running when an idle precedes the exact prompt echo", () => + it.effect("reconciles a sole idle after a stop when the exact prompt echo arrives", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-idle-before-exact-echo-after-stop"); @@ -3524,6 +3457,12 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { runtimeMock.state.autoPromptEcho = false; runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; runtimeMock.state.sessionStatusImplementation = async () => ({ data: {} }); + + const completedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.runHead, + Effect.forkChild, + ); yield* adapter.startSession({ provider: ProviderDriverKind.make("opencode"), threadId, @@ -3575,11 +3514,14 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }); yield* advanceTestClock(250); - const waiting = (yield* adapter.listSessions()).find( - (candidate) => candidate.threadId === threadId, + const completed = Option.getOrUndefined( + yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), ); - NodeAssert.equal(waiting?.status, "running"); - NodeAssert.equal(waiting?.activeTurnId, activeTurn.turnId); + NodeAssert.equal(completed?.turnId, activeTurn.turnId); + const sessions = yield* adapter.listSessions(); + const session = sessions.find((candidate) => candidate.threadId === threadId); + NodeAssert.equal(session?.status, "ready"); + NodeAssert.equal(session?.activeTurnId, undefined); yield* adapter.stopSession(threadId); }), @@ -3591,15 +3533,10 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const threadId = asThreadId("thread-idle-and-echo-before-acceptance-after-stop"); const idleEvent = promiseWithResolvers(); const userMessageEvent = promiseWithResolvers(); - const assistantEvent = promiseWithResolvers(); const activePromptStarted = promiseWithResolvers(); const activePromptRelease = promiseWithResolvers(); runtimeMock.state.autoPromptEcho = false; - runtimeMock.state.subscribedEvents = [ - idleEvent.promise, - userMessageEvent.promise, - assistantEvent.promise, - ]; + runtimeMock.state.subscribedEvents = [idleEvent.promise, userMessageEvent.promise]; runtimeMock.state.sessionStatusImplementation = async () => ({ data: {} }); runtimeMock.state.promptAsyncImplementation = async () => { if (runtimeMock.state.promptCalls.length === 2) { @@ -3665,25 +3602,6 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const activeTurn = yield* Fiber.join(activeTurnFiber); yield* advanceTestClock(250); - const waiting = (yield* adapter.listSessions()).find( - (candidate) => candidate.threadId === threadId, - ); - NodeAssert.equal(waiting?.activeTurnId, activeTurn.turnId); - assistantEvent.resolve({ - id: "evt-assistant-after-held-acceptance", - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-after-held-acceptance", - role: "assistant", - parentID: activeMessageId, - finish: "stop", - }, - }, - }); - yield* advanceTestClock(2_000); - const completed = Option.getOrUndefined( yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), ); @@ -7573,23 +7491,9 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { "opencode/kimi-k3", ), }); - const promptMessageId = (runtimeMock.state.promptCalls.at(-1) as { messageID: string }) - .messageID; - enqueue({ - type: "session.status", - properties: { sessionID, status: { type: "busy" } }, - }); enqueue({ type: "message.updated", - properties: { - sessionID, - info: { - id: `history-message-${index}`, - role: "assistant", - parentID: promptMessageId, - finish: "stop", - }, - }, + properties: { sessionID, info: { id: `history-message-${index}`, role: "assistant" } }, }); enqueue({ type: "message.part.updated", @@ -7692,7 +7596,6 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { "opencode/kimi-k3", ), }); - enqueue({ type: "session.status", properties: { sessionID, status: { type: "busy" } } }); for (const parentID of ["", promptMessageId]) { enqueue({ type: "message.updated", @@ -8023,78 +7926,6 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("recovers a finished assistant message whose terminal event was lost", () => - Effect.gen(function* () { - const adapter = yield* OpenCodeAdapter; - const threadId = asThreadId("thread-reconnect-lost-terminal-message"); - const reconnect = promiseWithResolvers(); - runtimeMock.state.subscribedEvents = [reconnect.promise]; - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); - const turn = yield* adapter.sendTurn({ - threadId, - input: "Finish during reconnect", - modelSelection: createModelSelection( - ProviderInstanceId.make("opencode"), - "opencode/kimi-k3", - ), - }); - const promptMessageId = (runtimeMock.state.promptCalls[0] as { messageID: string }).messageID; - runtimeMock.state.messages.push( - { - info: { - id: "assistant-from-previous-turn", - role: "assistant", - parentID: "other-prompt", - finish: "stop", - }, - parts: [], - }, - { - info: { - id: "assistant-finished-during-reconnect", - role: "assistant", - parentID: promptMessageId, - finish: "length", - }, - parts: [], - }, - ); - const warningFiber = yield* adapter.streamEvents.pipe( - Stream.filter((event) => event.threadId === threadId && event.type === "runtime.warning"), - Stream.runHead, - Effect.forkChild, - ); - const completedFiber = yield* adapter.streamEvents.pipe( - Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), - Stream.runHead, - Effect.forkChild, - ); - runtimeMock.state.eventStreamError?.(new Error("socket closed")); - yield* Fiber.join(warningFiber); - reconnect.resolve({ - id: "evt-reconnected-after-finished-message", - type: "server.connected", - properties: {}, - } satisfies OpenCodeEvent); - yield* advanceTestClock(6_000); - const completed = Option.getOrUndefined( - yield* Fiber.join(completedFiber).pipe(Effect.timeout("1 second")), - ); - NodeAssert.equal(completed?.turnId, turn.turnId); - NodeAssert.equal(runtimeMock.state.sessionMessagesCalls, 1); - NodeAssert.equal( - (yield* adapter.listSessions()).find((candidate) => candidate.threadId === threadId) - ?.status, - "ready", - ); - yield* adapter.stopSession(threadId); - }), - ); - it.effect( "ends a running session on clean stream closure without discarding unresolved permissions", () => diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 456b11071305..9914c29e293c 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -213,7 +213,6 @@ interface OpenCodeIdleReconciliation { } interface OpenCodePromptAdmission { - steering: boolean; readonly generation: number; readonly turnId: TurnId; readonly messageId: string; @@ -223,8 +222,6 @@ interface OpenCodePromptAdmission { idleDuringAdmission: { readonly turnId: TurnId; readonly raw: unknown } | undefined; idleObservedAfterMessage: boolean; messageObserved: boolean; - assistantResponseFinished: boolean; - responseReceipt: Deferred.Deferred; busyObserved: boolean; idleStatusConfirmations: number; accepted: boolean; @@ -900,26 +897,12 @@ const cancelPendingOpenCodePrompt = Effect.fn("cancelPendingOpenCodePrompt")(fun return; } admission.cancelled = true; - if (admission.recoveryFiber) { - yield* Fiber.interrupt(admission.recoveryFiber); - } if (admission.promptFiber) { yield* Fiber.interrupt(admission.promptFiber); } yield* Deferred.await(admission.submissionSettled); }); -const retirePromptAdmission = Effect.fnUntraced(function* ( - context: OpenCodeSessionContext, - admission: OpenCodePromptAdmission, -) { - admission.cancelled = true; - context.promptAdmission = undefined; - if (admission.recoveryFiber) { - yield* Fiber.interrupt(admission.recoveryFiber); - } -}); - const closeStartingOpenCodeContext = Effect.fn("closeStartingOpenCodeContext")(function* ( context: OpenCodeSessionContext, abortRemote: boolean, @@ -1398,7 +1381,6 @@ export function makeOpenCodeAdapter( if (!promptAdmission.requiresMessageReceipt) { yield* Deferred.await(promptAdmission.acceptance); } - let lastStatusIsIdle = false; for ( let retryCount = 0; retryCount < 5 || (promptAdmission.requiresMessageReceipt && !promptAdmission.accepted); @@ -1475,7 +1457,6 @@ export function makeOpenCodeAdapter( const status = statusData?.[context.openCodeSessionId]; const isIdle = statusData !== undefined && (status === undefined || status.type === "idle"); - lastStatusIsIdle = isIdle; const isBusy = status?.type === "busy" || status?.type === "retry"; if (isBusy) { promptAdmission.busyObserved = true; @@ -1489,14 +1470,14 @@ export function makeOpenCodeAdapter( if ( isIdle && idle !== undefined && - (promptAdmission.assistantResponseFinished || promptAdmission.busyObserved) + (promptAdmission.messageObserved || promptAdmission.busyObserved) ) { context.promptAdmission = undefined; context.awaitingBusyAfterInterruption = false; yield* scheduleIdleReconciliation(context, promptAdmission.turnId, idle.raw); return; } - if (isIdle && promptAdmission.assistantResponseFinished) { + if (isIdle && promptAdmission.messageObserved) { promptAdmission.idleStatusConfirmations += 1; if (promptAdmission.idleStatusConfirmations >= 2) { context.promptAdmission = undefined; @@ -1517,7 +1498,7 @@ export function makeOpenCodeAdapter( } if ( isIdle && - promptAdmission.assistantResponseFinished && + promptAdmission.messageObserved && promptAdmission.recoveryRaw !== undefined ) { context.promptAdmission = undefined; @@ -1533,61 +1514,6 @@ export function makeOpenCodeAdapter( const delayMs = Math.min(250 * 2 ** retryCount, 2_000); yield* Effect.sleep(`${delayMs} millis`); } - // An idle status can precede OpenCode's first assistant response. Keep - // the turn owned until that response arrives; its terminal update - // restarts reconciliation without polling an idle session indefinitely. - if (promptAdmission.messageObserved && !promptAdmission.assistantResponseFinished) { - // The terminal assistant event may have been lost during a reconnect. - // Confirm it from OpenCode's durable messages before waiting for the stream. - if (lastStatusIsIdle) { - const messages = yield* runOpenCodeSdk("session.messages", (signal) => - context.client.session.messages({ sessionID: context.openCodeSessionId }, { signal }), - ).pipe(Effect.timeout("1 second"), Effect.option); - if ( - (yield* Ref.get(context.stopped)) || - sessions.get(context.session.threadId) !== context || - context.promptAdmission !== promptAdmission || - context.activeTurnId !== promptAdmission.turnId || - context.promptGeneration !== promptAdmission.generation || - promptAdmission.cancelled - ) { - return; - } - const finished = - Option.isSome(messages) && - messages.value.data?.some( - (message) => - message.info.role === "assistant" && - message.info.parentID === promptAdmission.messageId && - (message.info.finish === "stop" || message.info.finish === "length"), - ); - if (finished) { - promptAdmission.assistantResponseFinished = true; - yield* Deferred.succeed(promptAdmission.responseReceipt, undefined); - } - } - const response = yield* Deferred.await(promptAdmission.responseReceipt).pipe( - Effect.timeoutOption("2 minutes"), - ); - if (Option.isNone(response)) { - yield* failPromptAdmissionRecovery(context, promptAdmission); - return; - } - if ( - context.promptAdmission === promptAdmission && - context.activeTurnId === promptAdmission.turnId && - context.promptGeneration === promptAdmission.generation && - !promptAdmission.cancelled - ) { - context.promptAdmission = undefined; - yield* scheduleIdleReconciliation( - context, - promptAdmission.turnId, - promptAdmission.recoveryRaw, - ); - } - return; - } yield* failPromptAdmissionRecovery(context, promptAdmission); }).pipe( Effect.ignoreCause, @@ -2489,7 +2415,15 @@ export function makeOpenCodeAdapter( promptAdmission.messageObserved = true; yield* Deferred.succeed(promptAdmission.messageReceipt, undefined); if (promptAdmission.accepted) { - yield* schedulePromptAdmissionRecovery(context, event); + const idle = promptAdmission.idleDuringAdmission; + context.awaitingBusyAfterInterruption = false; + context.promptAdmission = undefined; + if (promptAdmission.recoveryFiber) { + yield* Fiber.interrupt(promptAdmission.recoveryFiber); + } + if (idle) { + yield* scheduleIdleReconciliation(context, idle.turnId, idle.raw); + } } } context.messageRoleById.set(event.properties.info.id, event.properties.info.role); @@ -2497,11 +2431,6 @@ export function makeOpenCodeAdapter( context.textPartsByMessageId.delete(event.properties.info.id); } if (event.properties.info.role === "assistant") { - const finishesPromptResponse = - promptAdmission !== undefined && - promptAdmission.messageId === event.properties.info.parentID && - (event.properties.info.finish === "stop" || - event.properties.info.finish === "length"); const usage = context.turnTokenUsage; const parentMessageId = typeof event.properties.info.parentID === "string" && @@ -2545,18 +2474,6 @@ export function makeOpenCodeAdapter( yield* emitAssistantProgressCompletion(context, part, turnId, event); } } - if (finishesPromptResponse) { - promptAdmission.assistantResponseFinished = true; - yield* Deferred.succeed(promptAdmission.responseReceipt, undefined); - if (promptAdmission.accepted) { - if (promptAdmission.recoveryFiber) { - yield* Fiber.interrupt(promptAdmission.recoveryFiber); - } - context.promptAdmission = undefined; - context.awaitingBusyAfterInterruption = false; - yield* scheduleIdleReconciliation(context, promptAdmission.turnId, event); - } - } } break; } @@ -2763,13 +2680,8 @@ export function makeOpenCodeAdapter( yield* cancelIdleReconciliation(context); context.awaitingBusyAfterInterruption = false; if (context.promptAdmission?.turnId === turnId) { - const admission = context.promptAdmission; - admission.busyObserved = true; - if (admission.accepted && admission.messageObserved) { - yield* retirePromptAdmission(context, admission); - } else { - yield* schedulePromptAdmissionRecovery(context, event); - } + context.promptAdmission.busyObserved = true; + yield* schedulePromptAdmissionRecovery(context, event); } yield* updateProviderSession(context, { status: "running", @@ -2799,19 +2711,9 @@ export function makeOpenCodeAdapter( break; } if (context.promptAdmission?.turnId === turnId) { - const admission = context.promptAdmission; - const priorIdle = admission.idleDuringAdmission ?? admission.priorIdle; - admission.idleDuringAdmission = { turnId, raw: event }; - admission.idleObservedAfterMessage = admission.messageObserved; - if ( - admission.accepted && - admission.messageObserved && - (priorIdle || !context.awaitingBusyAfterInterruption) - ) { - yield* retirePromptAdmission(context, admission); - yield* scheduleIdleReconciliation(context, turnId, event); - break; - } + context.promptAdmission.idleDuringAdmission = { turnId, raw: event }; + context.promptAdmission.idleObservedAfterMessage = + context.promptAdmission.messageObserved; yield* schedulePromptAdmissionRecovery(context, event); break; } @@ -2846,10 +2748,6 @@ export function makeOpenCodeAdapter( break; } } - const promptAdmission = context.promptAdmission; - if (promptAdmission) { - yield* retirePromptAdmission(context, promptAdmission); - } yield* cancelIdleReconciliation(context); const terminalCancellation = activeTurnId !== undefined && cancellation?.turnId === activeTurnId @@ -3367,12 +3265,7 @@ export function makeOpenCodeAdapter( : undefined; context.pendingIdleReconciliation = undefined; const promptGeneration = context.promptGeneration + 1; - const previousAdmission = context.promptAdmission; - if (previousAdmission?.recoveryFiber) { - yield* Fiber.interrupt(previousAdmission.recoveryFiber); - } const promptAdmission: OpenCodePromptAdmission = { - steering: steeringTurnId !== undefined, generation: promptGeneration, turnId, messageId, @@ -3382,8 +3275,6 @@ export function makeOpenCodeAdapter( idleDuringAdmission: undefined, idleObservedAfterMessage: false, messageObserved: false, - assistantResponseFinished: false, - responseReceipt: Deferred.makeUnsafe(), busyObserved: false, idleStatusConfirmations: 0, accepted: false, @@ -3688,29 +3579,13 @@ export function makeOpenCodeAdapter( ) { context.awaitingBusyAfterInterruption = false; const idle = promptAdmission.idleDuringAdmission; - if (promptAdmission.busyObserved && promptAdmission.messageObserved) { - if (promptAdmission.recoveryFiber) { - yield* Fiber.interrupt(promptAdmission.recoveryFiber); - } - context.promptAdmission = undefined; - if (idle) { - yield* scheduleIdleReconciliation(context, turnId, idle.raw); - } - } else if (promptAdmission.assistantResponseFinished) { - if (promptAdmission.recoveryFiber) { - yield* Fiber.interrupt(promptAdmission.recoveryFiber); - } - context.promptAdmission = undefined; - yield* scheduleIdleReconciliation(context, turnId, promptAdmission.recoveryRaw); - } else if ( - idle && - promptAdmission.idleObservedAfterMessage && - promptAdmission.steering - ) { + if (idle && !promptAdmission.idleObservedAfterMessage) { + yield* schedulePromptAdmissionRecovery(context, idle.raw); + } else { context.promptAdmission = undefined; + } + if (idle && promptAdmission.idleObservedAfterMessage) { yield* scheduleIdleReconciliation(context, turnId, idle.raw); - } else { - yield* schedulePromptAdmissionRecovery(context, idle?.raw); } } else { yield* schedulePromptAdmissionRecovery(context, promptAdmission.recoveryRaw); @@ -3849,7 +3724,11 @@ export function makeOpenCodeAdapter( context.cancellation = cancellation; const promptAdmission = context.promptAdmission; if (promptAdmission !== undefined && promptAdmission.turnId === interruptedTurnId) { - yield* cancelPendingOpenCodePrompt(context); + promptAdmission.cancelled = true; + if (promptAdmission.promptFiber) { + yield* Fiber.interrupt(promptAdmission.promptFiber); + } + yield* Deferred.await(promptAdmission.submissionSettled); } yield* Effect.forEach([...context.commandFibers], Fiber.interrupt, { discard: true }); From 0fbe94b54b41534daf9251a4d55f4b06a744441a Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Fri, 2 Oct 2026 13:01:03 +0200 Subject: [PATCH 3/4] fix(server): settle OpenCode text parts when switching --- .../Layers/ProviderRuntimeIngestion.test.ts | 70 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 3 +- 2 files changed, 72 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 6e976d8a1d0d..7d53931f51b7 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1439,6 +1439,76 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + it.each([ + { mode: "token" as const, terminalType: "turn.completed" as const }, + { mode: "token" as const, terminalType: "turn.aborted" as const }, + { mode: "paragraph" as const, terminalType: "turn.completed" as const }, + { mode: "paragraph" as const, terminalType: "turn.aborted" as const }, + ])( + "settles the previous OpenCode part in $mode mode before $terminalType", + async ({ mode, terminalType }) => { + const harness = await createHarness({ serverSettings: { responseStreamingMode: mode } }); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-part-switch"); + const base = { + provider: ProviderDriverKind.make("opencode"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId, + turnId, + }; + const delta = (itemId: string, text: string) => ({ + ...base, + type: "content.delta" as const, + eventId: asEventId(`evt-${itemId}`), + itemId: asItemId(itemId), + payload: { streamKind: "assistant_text" as const, delta: text }, + }); + const readMessages = async () => + (await harness.readModel()).threads.find((entry) => entry.id === threadId)!.messages; + await harness.emitAndDrain([ + { ...base, type: "turn.started", eventId: asEventId("evt-part-switch-started") }, + delta("first-part", "First part.\n\n"), + ]); + expect(await readMessages()).toContainEqual( + expect.objectContaining({ + id: "assistant:first-part", + text: "First part.\n\n", + streaming: true, + }), + ); + + harness.advanceClock(1_000); + await harness.emitAndDrain([delta("second-part", "Second part.\n\n")]); + expect(await readMessages()).toEqual([ + expect.objectContaining({ id: "assistant:first-part", streaming: false }), + expect.objectContaining({ id: "assistant:second-part", streaming: true }), + ]); + await harness.emitAndDrain([ + { + ...base, + type: terminalType, + eventId: asEventId("evt-part-switch-terminal"), + payload: + terminalType === "turn.completed" + ? { state: "completed" } + : { reason: "Interrupted by user." }, + }, + ]); + expect(await readMessages()).toEqual([ + expect.objectContaining({ + id: "assistant:first-part", + text: "First part.\n\n", + streaming: false, + }), + expect.objectContaining({ + id: "assistant:second-part", + text: "Second part.\n\n", + streaming: false, + }), + ]); + }, + ); + it("keeps a completed OpenCode turn with reasoning but no answer visible", async () => { const harness = await createHarness(); const threadId = asThreadId("thread-1"); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 901209ed957e..28bee268cdca 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1299,7 +1299,8 @@ const make = Effect.gen(function* () { createdAt: input.event.createdAt, commandTag: "assistant-complete-on-new-part", finalDeltaCommandTag: "assistant-delta-finalize-on-new-part", - hasProjectedMessage: false, + hasProjectedMessage: + (yield* getThreadMessageById(input.threadId, activeMessageId)) !== undefined, }); } From 4b039fcd42cc22f2d5910ba132f3e44eeaa60782 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Fri, 2 Oct 2026 13:10:39 +0200 Subject: [PATCH 4/4] fix(server): settle OpenCode progress while awaiting input --- .../Layers/ProviderRuntimeIngestion.test.ts | 108 ++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 22 ++++ 2 files changed, 130 insertions(+) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 7d53931f51b7..bb0a0d0d0552 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1509,6 +1509,114 @@ describe("ProviderRuntimeIngestion", () => { }, ); + it.each( + (["token", "paragraph"] as const).flatMap((mode) => + (["approval", "question", "message"] as const).map((boundary) => ({ mode, boundary })), + ), + )( + "settles OpenCode progress at a $boundary boundary in $mode mode", + async ({ mode, boundary }) => { + const harness = await createHarness({ serverSettings: { responseStreamingMode: mode } }); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-progress-pause"); + const base = { + provider: ProviderDriverKind.make("opencode"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId, + turnId, + }; + const delta = (id: string, text: string, eventId = id) => ({ + ...base, + type: "content.delta" as const, + eventId: asEventId(`evt-${eventId}`), + itemId: asItemId(id), + payload: { streamKind: "assistant_progress_text" as const, delta: text }, + }); + const readMessages = async () => + (await harness.readModel()).threads.find((entry) => entry.id === threadId)!.messages; + await harness.emitAndDrain([ + { ...base, type: "turn.started", eventId: asEventId("evt-progress-pause-started") }, + delta("buffered-progress", "Checking "), + delta("projected-progress", "Waiting.\n\n"), + ]); + expect(await readMessages()).toMatchObject([ + { + id: "assistant:projected-progress", + role: "reasoning", + text: "Waiting.\n\n", + streaming: true, + }, + ]); + const request = + boundary === "approval" + ? { + ...base, + type: "request.opened" as const, + eventId: asEventId("evt-progress-pause-approval"), + requestId: RuntimeRequestId.make("progress-pause-approval"), + payload: { requestType: "command_execution_approval" as const, detail: "pwd" }, + } + : { + ...base, + type: "user-input.requested" as const, + eventId: asEventId("evt-progress-pause-question"), + requestId: RuntimeRequestId.make("progress-pause-question"), + payload: { + ...(boundary === "message" ? { responseMode: "message" as const } : {}), + questions: [{ id: "choice", header: "Choice", question: "Continue?", options: [] }], + }, + }; + await harness.emitAndDrain([request]); + if (boundary === "message") { + expect(await readMessages()).toMatchObject([ + { id: "assistant:projected-progress", streaming: true }, + ]); + } else { + expect(await readMessages()).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + id: "assistant:buffered-progress", + role: "reasoning", + text: "Checking ", + streaming: false, + }), + expect.objectContaining({ + id: "assistant:projected-progress", + role: "reasoning", + text: "Waiting.\n\n", + streaming: false, + }), + ]), + ); + } + await harness.emitAndDrain([ + delta("buffered-progress", "again.", "progress-resumed"), + { + ...base, + type: "turn.completed", + eventId: asEventId("evt-progress-pause-completed"), + payload: { state: "completed" }, + }, + ]); + expect(await readMessages()).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + id: "assistant:buffered-progress", + role: "reasoning", + text: "Checking again.", + streaming: false, + }), + expect.objectContaining({ + id: "assistant:projected-progress", + role: "reasoning", + text: "Waiting.\n\n", + streaming: false, + }), + ]), + ); + }, + ); + it("keeps a completed OpenCode turn with reasoning but no answer visible", async () => { const harness = await createHarness(); const threadId = asThreadId("thread-1"); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 28bee268cdca..3ae4e5ff764d 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2241,6 +2241,28 @@ const make = Effect.gen(function* () { hasProjectedMessage, flushedMessageIds, }); + const progressMessageIdsForTurn = yield* getAssistantMessageIdsForTurn( + thread.id, + pauseForUserTurnId, + ); + for (const messageId of progressMessageIdsForTurn) { + if (Option.isNone(yield* Cache.getOption(progressMessageIds, messageId))) { + continue; + } + const projectedMessage = yield* getThreadMessageById(thread.id, messageId); + yield* finalizeAssistantMessage({ + event, + threadId: thread.id, + messageId, + turnId: pauseForUserTurnId, + createdAt: now, + commandTag: "assistant-progress-complete-on-pause", + finalDeltaCommandTag: "assistant-progress-delta-finalize-on-pause", + hasProjectedMessage: projectedMessage !== undefined, + presentation: "progress", + }); + yield* forgetAssistantMessageId(thread.id, pauseForUserTurnId, messageId); + } } if (proposedPlanDelta && proposedPlanDelta.length > 0) {