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..bb0a0d0d0552 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1439,6 +1439,366 @@ 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.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"); + 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..3ae4e5ff764d 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,39 @@ 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: + (yield* getThreadMessageById(input.threadId, activeMessageId)) !== undefined, + }); } return yield* startAssistantSegmentForTurn({ threadId: input.threadId, turnId: input.turnId, - baseKey: assistantSegmentBaseKeyFromEvent(input.event), + baseKey, }); }); @@ -1423,6 +1448,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 +1472,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 +1532,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 +1544,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 +1816,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 +1995,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 +2134,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") @@ -2149,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) { @@ -2256,6 +2370,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 +2399,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 +2422,7 @@ const make = Effect.gen(function* () { !existingAssistantMessage || existingAssistantMessage.text.length === 0; const shouldSkipRedundantCompletion = + assistantCompletion.presentation !== "progress" && Option.isNone(activeAssistantMessageId) && turnId !== undefined && hasAssistantMessagesForTurn && @@ -2317,10 +2438,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 +2458,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 +2538,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..28281fab00d8 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -7024,6 +7024,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 +7253,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; diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 34ebd1b544fb..9914c29e293c 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -332,6 +332,7 @@ type OpenCodeTextPartState = Pick, "id" | "tokens">; @@ -350,6 +351,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 +599,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 +624,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 +696,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": @@ -1613,6 +1637,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 +2316,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: { @@ -2387,6 +2466,13 @@ 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); + } } } break; @@ -3018,6 +3104,7 @@ export function makeOpenCodeAdapter( pendingQuestions: new Map(), textPartsByMessageId: new Map(), messageRoleById: new Map(), + messageTurnIdById: new Map(), turnTokenUsage: undefined, activeTurnId: undefined, activeAgent: undefined, @@ -3201,6 +3288,7 @@ export function makeOpenCodeAdapter( context.promptAdmission = promptAdmission; context.activeTurnId = turnId; + rememberMessageTurnId(context, messageId, turnId); if (steeringTurnId === undefined) { context.turnTokenUsage = makeOpenCodeTurnTokenUsageAccumulator(); } @@ -3987,6 +4075,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),