Merge pull request #4 from eimexdev/t3code/fix-imported-chat-formatting

Fix Hermes imported history and native tool activity rendering

Author: eimexdev

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -242,6 +242,79 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
 it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-base-")))(
   "OrchestrationProjectionPipeline",
   (it) => {
+    it.effect("fully replaces imported history slots on refresh", () =>
+      Effect.gen(function* () {
+        const projectionPipeline = yield* OrchestrationProjectionPipeline;
+        const eventStore = yield* OrchestrationEventStore;
+        const sql = yield* SqlClient.SqlClient;
+        const threadId = ThreadId.make("thread-import-refresh");
+        const messageId = MessageId.make("message-import-refresh");
+        const originalAt = "2026-01-01T00:00:00.000Z";
+        const refreshedAt = "2026-01-01T00:01:00.000Z";
+
+        yield* eventStore.append({
+          type: "thread.message-sent",
+          eventId: EventId.make("evt-import-original"),
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          occurredAt: originalAt,
+          commandId: CommandId.make("cmd-import-original"),
+          causationEventId: null,
+          correlationId: CommandId.make("cmd-import-original"),
+          metadata: {},
+          payload: {
+            threadId,
+            messageId,
+            role: "user",
+            text: "legacy provider wrapper",
+            turnId: null,
+            streaming: false,
+            createdAt: originalAt,
+            updatedAt: originalAt,
+          },
+        });
+        yield* eventStore.append({
+          type: "thread.message-sent",
+          eventId: EventId.make("evt-import-refresh"),
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          occurredAt: refreshedAt,
+          commandId: CommandId.make("cmd-import-refresh"),
+          causationEventId: null,
+          correlationId: CommandId.make("cmd-import-refresh"),
+          metadata: {},
+          payload: {
+            threadId,
+            messageId,
+            role: "assistant",
+            text: "",
+            replaceText: true,
+            turnId: null,
+            streaming: false,
+            imported: true,
+            createdAt: refreshedAt,
+            updatedAt: refreshedAt,
+          },
+        });
+
+        yield* projectionPipeline.bootstrap;
+
+        const rows = yield* sql<{
+          readonly role: string;
+          readonly text: string;
+          readonly createdAt: string;
+        }>`
+          SELECT
+            role,
+            text,
+            created_at AS "createdAt"
+          FROM projection_thread_messages
+          WHERE message_id = ${messageId}
+        `;
+        assert.deepEqual(rows, [{ role: "assistant", text: "", createdAt: refreshedAt }]);
+      }),
+    );
+
     it.effect("stores message attachment references without mutating payloads", () =>
       Effect.gen(function* () {
         const projectionPipeline = yield* OrchestrationProjectionPipeline;

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -879,7 +879,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             text: nextText,
             ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),
             isStreaming: event.payload.streaming,
-            createdAt: previousMessage?.createdAt ?? event.payload.createdAt,
+            createdAt:
+              event.payload.imported || previousMessage === undefined
+                ? event.payload.createdAt
+                : previousMessage.createdAt,
             updatedAt: event.payload.updatedAt,
           });
           return;

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts

@@ -2866,9 +2866,16 @@ describe("ProviderRuntimeIngestion", () => {
       turnId: asTurnId("turn-9"),
       payload: {
         itemType: "command_execution",
-        status: "in_progress",
+        status: "inProgress",
         title: "Read file",
         detail: "/tmp/file.ts",
+        data: {
+          toolCallId: "call-read",
+          item: {
+            name: "read_file",
+            input: { path: "/tmp/file.ts" },
+          },
+        },
       },
     });
 
@@ -2883,11 +2890,22 @@ describe("ProviderRuntimeIngestion", () => {
     );
 
     expect(thread.session?.status).toBe("ready");
-    expect(
-      thread.activities.some(
-        (activity: ProviderRuntimeTestActivity) => activity.kind === "tool.started",
-      ),
-    ).toBe(true);
+    const toolStarted = thread.activities.find(
+      (activity: ProviderRuntimeTestActivity) => activity.kind === "tool.started",
+    );
+    expect(toolStarted?.payload).toEqual({
+      itemType: "command_execution",
+      status: "inProgress",
+      title: "Read file",
+      detail: "/tmp/file.ts",
+      data: {
+        toolCallId: "call-read",
+        item: {
+          name: "read_file",
+          input: { path: "/tmp/file.ts" },
+        },
+      },
+    });
   });
 
   it("consumes P1 runtime events into thread metadata, diff checkpoints, and activities", async () => {

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts

@@ -693,7 +693,10 @@ export function runtimeEventToActivities(
           summary: `${event.payload.title ?? "Tool"} started`,
           payload: {
             itemType: event.payload.itemType,
+            ...(event.payload.status ? { status: event.payload.status } : {}),
+            ...(event.payload.title ? { title: event.payload.title } : {}),
             ...(event.payload.detail ? { detail: truncateDetail(event.payload.detail) } : {}),
+            ...(event.payload.data !== undefined ? { data: event.payload.data } : {}),
           },
           turnId: toTurnId(event.turnId) ?? null,
           ...maybeSequence,

apps/server/src/orchestration/decider.import.test.ts

@@ -0,0 +1,96 @@
+import {
+  CommandId,
+  DEFAULT_PROVIDER_INTERACTION_MODE,
+  EventId,
+  MessageId,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+} from "@t3tools/contracts";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { expect, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+
+import { decideOrchestrationCommand } from "./decider.ts";
+import { createEmptyReadModel, projectEvent } from "./projector.ts";
+
+it.layer(NodeServices.layer)("decider imported messages", (it) => {
+  it.effect("replaces an existing history slot even when the imported text is empty", () =>
+    Effect.gen(function* () {
+      const createdAt = "2026-01-01T00:00:00.000Z";
+      const projectId = ProjectId.make("project-import");
+      const threadId = ThreadId.make("thread-import");
+      const initial = createEmptyReadModel(createdAt);
+      const withProject = yield* projectEvent(initial, {
+        sequence: 1,
+        eventId: EventId.make("event-project-import"),
+        aggregateKind: "project",
+        aggregateId: projectId,
+        type: "project.created",
+        occurredAt: createdAt,
+        commandId: CommandId.make("command-project-import"),
+        causationEventId: null,
+        correlationId: CommandId.make("command-project-import"),
+        metadata: {},
+        payload: {
+          projectId,
+          title: "Imports",
+          workspaceRoot: "/tmp/imports",
+          defaultModelSelection: null,
+          scripts: [],
+          createdAt,
+          updatedAt: createdAt,
+        },
+      });
+      const readModel = yield* projectEvent(withProject, {
+        sequence: 2,
+        eventId: EventId.make("event-thread-import"),
+        aggregateKind: "thread",
+        aggregateId: threadId,
+        type: "thread.created",
+        occurredAt: createdAt,
+        commandId: CommandId.make("command-thread-import"),
+        causationEventId: null,
+        correlationId: CommandId.make("command-thread-import"),
+        metadata: {},
+        payload: {
+          threadId,
+          projectId,
+          title: "Imported thread",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("hermes"),
+            model: "openai-codex::gpt-5.6-sol",
+          },
+          runtimeMode: "full-access",
+          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+          branch: null,
+          worktreePath: null,
+          createdAt,
+          updatedAt: createdAt,
+        },
+      });
+
+      const event = yield* decideOrchestrationCommand({
+        command: {
+          type: "thread.message.import",
+          commandId: CommandId.make("command-message-import"),
+          threadId,
+          messageId: MessageId.make("message-import"),
+          role: "assistant",
+          text: "",
+          createdAt: "2026-01-01T00:01:00.000Z",
+        },
+        readModel,
+      });
+
+      const importedEvent = Array.isArray(event) ? event[0] : event;
+      expect(importedEvent?.type).toBe("thread.message-sent");
+      if (importedEvent?.type !== "thread.message-sent") {
+        return;
+      }
+      expect(importedEvent.payload.imported).toBe(true);
+      expect(importedEvent.payload.replaceText).toBe(true);
+      expect(importedEvent.payload.text).toBe("");
+    }),
+  );
+});

apps/server/src/orchestration/decider.ts

@@ -921,7 +921,8 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
           messageId: command.messageId,
           role: command.role,
           text: command.text,
-          turnId: null,
+          replaceText: true,
+          turnId: command.turnId ?? null,
           streaming: false,
           imported: true,
           createdAt: command.createdAt,

apps/server/src/orchestration/projector.ts

@@ -443,6 +443,7 @@ export function projectEvent(
               entry.id === message.id
                 ? {
                     ...entry,
+                    role: message.role,
                     text: message.streaming
                       ? `${entry.text}${message.text}`
                       : payload.replaceText
@@ -451,6 +452,7 @@ export function projectEvent(
                           ? message.text
                           : entry.text,
                     streaming: message.streaming,
+                    createdAt: payload.imported ? message.createdAt : entry.createdAt,
                     updatedAt: message.updatedAt,
                     turnId: message.turnId,
                     ...(message.attachments !== undefined

apps/server/src/provider/Layers/HermesAdapter.test.ts

@@ -237,6 +237,77 @@ it.layer(testLayer)("HermesAdapter", (it) => {
     }),
   );
 
+  it.effect("emits correlated native tool lifecycle events with full tool data", () =>
+    Effect.gen(function* () {
+      const { adapter } = yield* HermesAdapterTestHarness;
+      const threadId = ThreadId.make("hermes-tool-thread");
+      yield* adapter.startSession({
+        provider: ProviderDriverKind.make("hermes"),
+        threadId,
+        runtimeMode: "full-access",
+      });
+
+      const eventsFiber = yield* adapter.streamEvents.pipe(
+        Stream.take(3),
+        Stream.runCollect,
+        Effect.forkChild,
+      );
+      yield* Effect.yieldNow;
+      const turn = yield* adapter.sendTurn({ threadId, input: "read the skill" });
+      const sourceMessageId = `hermes-user:${turn.turnId}`;
+      yield* adapter.receiveCallback({
+        protocolVersion: HERMES_BRIDGE_PROTOCOL_VERSION,
+        requestId: "callback-tool-start-request",
+        deliveryId: "callback-tool-start-delivery",
+        type: "tool.started",
+        chatId: "t3agent",
+        threadId,
+        sourceMessageId,
+        toolCallId: "call-skill",
+        name: "skill_view",
+        input: { name: "query" },
+      });
+      yield* adapter.receiveCallback({
+        protocolVersion: HERMES_BRIDGE_PROTOCOL_VERSION,
+        requestId: "callback-tool-complete-request",
+        deliveryId: "callback-tool-complete-delivery",
+        type: "tool.completed",
+        chatId: "t3agent",
+        threadId,
+        sourceMessageId,
+        toolCallId: "call-skill",
+        name: "skill_view",
+        input: { name: "query" },
+        result: "Skill loaded",
+        isError: false,
+      });
+
+      const events = Array.from(yield* Fiber.join(eventsFiber));
+      NodeAssert.deepEqual(
+        events.map((event) => event.type),
+        ["turn.started", "item.started", "item.completed"],
+      );
+      const started = events[1];
+      const completed = events[2];
+      NodeAssert.equal(started?.itemId, completed?.itemId);
+      NodeAssert.deepEqual(completed?.payload, {
+        itemType: "mcp_tool_call",
+        status: "completed",
+        title: "Read skill",
+        detail: "Skill loaded",
+        data: {
+          toolCallId: "call-skill",
+          item: {
+            toolCallId: "call-skill",
+            name: "skill_view",
+            input: { name: "query" },
+            result: { output: "Skill loaded" },
+          },
+        },
+      });
+    }),
+  );
+
   it.effect("acknowledges a repeated delivery id as a duplicate", () =>
     Effect.gen(function* () {
       const { adapter } = yield* HermesAdapterTestHarness;