fix(opencode): keep tool-call progress out of assistant answers

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." },
+    ]);
+  }),
+);

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") {

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<typeof eventStore.append>[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) => {

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