Skip to content
Merged
73 changes: 73 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
5 changes: 4 additions & 1 deletion apps/server/src/orchestration/Layers/ProjectionPipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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" },
},
},
},
});

Expand All @@ -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 () => {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
96 changes: 96 additions & 0 deletions apps/server/src/orchestration/decider.import.test.ts
Original file line number Diff line number Diff line change
@@ -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("");
}),
);
});
3 changes: 2 additions & 1 deletion apps/server/src/orchestration/decider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration/projector.ts
Original file line number Diff line number Diff line change
Expand Up @@ -443,6 +443,7 @@ export function projectEvent(
entry.id === message.id
? {
...entry,
role: message.role,
text: message.streaming
? `${entry.text}${message.text}`
: payload.replaceText
Expand All @@ -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
Expand Down
71 changes: 71 additions & 0 deletions apps/server/src/provider/Layers/HermesAdapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading