import { CommandId, EventId, ProjectId, ThreadId, type OrchestrationEvent,} from "@t3tools/contracts";import { Effect } from "effect";import { describe, expect, it } from "vitest"; import { createEmptyReadModel, projectEvent } from "./projector.ts"; function makeEvent(input: { sequence: number; type: OrchestrationEvent["type"]; occurredAt: string; aggregateKind: OrchestrationEvent["aggregateKind"]; aggregateId: string; commandId: string | null; payload: unknown;}): OrchestrationEvent { return { sequence: input.sequence, eventId: EventId.makeUnsafe(`event-${input.sequence}`), type: input.type, aggregateKind: input.aggregateKind, aggregateId: input.aggregateKind === "project" ? ProjectId.makeUnsafe(input.aggregateId) : ThreadId.makeUnsafe(input.aggregateId), occurredAt: input.occurredAt, commandId: input.commandId === null ? null : CommandId.makeUnsafe(input.commandId), causationEventId: null, correlationId: null, metadata: {}, payload: input.payload as never, } as OrchestrationEvent;} describe("orchestration projector", () => { it("applies thread.created events", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); const next = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: now, commandId: "cmd-thread-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5-codex", branch: null, worktreePath: null, createdAt: now, updatedAt: now, }, }), ), ); expect(next.snapshotSequence).toBe(1); expect(next.threads).toEqual([ { id: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5-codex", branch: null, worktreePath: null, latestTurnId: null, createdAt: now, updatedAt: now, deletedAt: null, messages: [], activities: [], checkpoints: [], session: null, }, ]); }); it("fails when event payload cannot be decoded by runtime schema", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); await expect( Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: now, commandId: "cmd-invalid", payload: { // missing required threadId projectId: "project-1", title: "demo", model: "gpt-5-codex", branch: null, worktreePath: null, createdAt: now, updatedAt: now, }, }), ), ), ).rejects.toBeDefined(); }); it("keeps projector forward-compatible for unhandled event types", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); const next = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 7, type: "thread.turn-start-requested", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-01-01T00:00:00.000Z", commandId: "cmd-unhandled", payload: { threadId: "thread-1", messageId: "message-1", createdAt: "2026-01-01T00:00:00.000Z", }, }), ), ); expect(next.snapshotSequence).toBe(7); expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z"); expect(next.threads).toEqual([]); }); it("tracks latest turn id from session lifecycle events", async () => { const createdAt = "2026-02-23T08:00:00.000Z"; const startedAt = "2026-02-23T08:00:05.000Z"; const model = createEmptyReadModel(createdAt); const afterCreate = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5.3-codex", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }), ), ); const afterRunning = await Effect.runPromise( projectEvent( afterCreate, makeEvent({ sequence: 2, type: "thread.session-set", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: startedAt, commandId: "cmd-running", payload: { threadId: "thread-1", session: { threadId: "thread-1", status: "running", providerName: "codex", providerSessionId: "session-1", providerThreadId: "provider-thread-1", activeTurnId: "turn-1", lastError: null, updatedAt: startedAt, }, }, }), ), ); const thread = afterRunning.threads[0]; expect(thread?.latestTurnId).toBe("turn-1"); expect(thread?.session?.status).toBe("running"); }); it("marks assistant messages completed with non-streaming updates", async () => { const createdAt = "2026-02-23T09:00:00.000Z"; const deltaAt = "2026-02-23T09:00:01.000Z"; const completeAt = "2026-02-23T09:00:03.500Z"; const model = createEmptyReadModel(createdAt); const afterCreate = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5.3-codex", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }), ), ); const afterDelta = await Effect.runPromise( projectEvent( afterCreate, makeEvent({ sequence: 2, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: deltaAt, commandId: "cmd-delta", payload: { threadId: "thread-1", messageId: "assistant:msg-1", role: "assistant", text: "hello", turnId: "turn-1", streaming: true, createdAt: deltaAt, updatedAt: deltaAt, }, }), ), ); const afterComplete = await Effect.runPromise( projectEvent( afterDelta, makeEvent({ sequence: 3, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: completeAt, commandId: "cmd-complete", payload: { threadId: "thread-1", messageId: "assistant:msg-1", role: "assistant", text: "", turnId: "turn-1", streaming: false, createdAt: completeAt, updatedAt: completeAt, }, }), ), ); const message = afterComplete.threads[0]?.messages[0]; expect(message?.id).toBe("assistant:msg-1"); expect(message?.text).toBe("hello"); expect(message?.streaming).toBe(false); expect(message?.updatedAt).toBe(completeAt); }); it("prunes reverted turn messages from in-memory thread snapshot", async () => { const createdAt = "2026-02-23T10:00:00.000Z"; const model = createEmptyReadModel(createdAt); const afterCreate = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5.3-codex", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }), ), ); const events: ReadonlyArray<OrchestrationEvent> = [ makeEvent({ sequence: 2, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:01.000Z", commandId: "cmd-user-1", payload: { threadId: "thread-1", messageId: "user-msg-1", role: "user", text: "First edit", turnId: null, streaming: false, createdAt: "2026-02-23T10:00:01.000Z", updatedAt: "2026-02-23T10:00:01.000Z", }, }), makeEvent({ sequence: 3, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.000Z", commandId: "cmd-assistant-1", payload: { threadId: "thread-1", messageId: "assistant-msg-1", role: "assistant", text: "Updated README to v2.\n", turnId: "turn-1", streaming: false, createdAt: "2026-02-23T10:00:02.000Z", updatedAt: "2026-02-23T10:00:02.000Z", }, }), makeEvent({ sequence: 4, type: "thread.turn-diff-completed", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.500Z", commandId: "cmd-turn-1-complete", payload: { threadId: "thread-1", turnId: "turn-1", checkpointTurnCount: 1, checkpointRef: "refs/t3/checkpoints/thread-1/turn/1", status: "ready", files: [], assistantMessageId: "assistant-msg-1", completedAt: "2026-02-23T10:00:02.500Z", }, }), makeEvent({ sequence: 5, type: "thread.activity-appended", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.750Z", commandId: "cmd-activity-1", payload: { threadId: "thread-1", activity: { id: "activity-1", tone: "tool", kind: "tool.started", summary: "Edit file started", payload: { toolKind: "command" }, turnId: "turn-1", createdAt: "2026-02-23T10:00:02.750Z", }, }, }), makeEvent({ sequence: 6, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:03.000Z", commandId: "cmd-user-2", payload: { threadId: "thread-1", messageId: "user-msg-2", role: "user", text: "Second edit", turnId: null, streaming: false, createdAt: "2026-02-23T10:00:03.000Z", updatedAt: "2026-02-23T10:00:03.000Z", }, }), makeEvent({ sequence: 7, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.000Z", commandId: "cmd-assistant-2", payload: { threadId: "thread-1", messageId: "assistant-msg-2", role: "assistant", text: "Updated README to v3.\n", turnId: "turn-2", streaming: false, createdAt: "2026-02-23T10:00:04.000Z", updatedAt: "2026-02-23T10:00:04.000Z", }, }), makeEvent({ sequence: 8, type: "thread.turn-diff-completed", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.500Z", commandId: "cmd-turn-2-complete", payload: { threadId: "thread-1", turnId: "turn-2", checkpointTurnCount: 2, checkpointRef: "refs/t3/checkpoints/thread-1/turn/2", status: "ready", files: [], assistantMessageId: "assistant-msg-2", completedAt: "2026-02-23T10:00:04.500Z", }, }), makeEvent({ sequence: 9, type: "thread.activity-appended", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.750Z", commandId: "cmd-activity-2", payload: { threadId: "thread-1", activity: { id: "activity-2", tone: "tool", kind: "tool.completed", summary: "Edit file complete", payload: { toolKind: "command" }, turnId: "turn-2", createdAt: "2026-02-23T10:00:04.750Z", }, }, }), makeEvent({ sequence: 10, type: "thread.reverted", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:05.000Z", commandId: "cmd-revert", payload: { threadId: "thread-1", turnCount: 1, }, }), ]; const afterRevert = await events.reduce<Promise<ReturnType<typeof createEmptyReadModel>>>( (statePromise, event) => statePromise.then((state) => Effect.runPromise(projectEvent(state, event))), Promise.resolve(afterCreate), ); const thread = afterRevert.threads[0]; expect(thread?.messages.map((message) => ({ role: message.role, text: message.text }))).toEqual( [ { role: "user", text: "First edit" }, { role: "assistant", text: "Updated README to v2.\n" }, ], ); expect( thread?.activities.map((activity) => ({ id: activity.id, turnId: activity.turnId })),Lines 1–500 of 747. Load more
import { CommandId, EventId, ProjectId, ThreadId, type OrchestrationEvent, } from "@t3tools/contracts"; import { Effect } from "effect"; import { describe, expect, it } from "vitest"; import { createEmptyReadModel, projectEvent } from "./projector.ts"; function makeEvent(input: { sequence: number; type: OrchestrationEvent["type"]; occurredAt: string; aggregateKind: OrchestrationEvent["aggregateKind"]; aggregateId: string; commandId: string | null; payload: unknown; }): OrchestrationEvent { return { sequence: input.sequence, eventId: EventId.makeUnsafe(`event-${input.sequence}`), type: input.type, aggregateKind: input.aggregateKind, aggregateId: input.aggregateKind === "project" ? ProjectId.makeUnsafe(input.aggregateId) : ThreadId.makeUnsafe(input.aggregateId), occurredAt: input.occurredAt, commandId: input.commandId === null ? null : CommandId.makeUnsafe(input.commandId), causationEventId: null, correlationId: null, metadata: {}, payload: input.payload as never, } as OrchestrationEvent; } describe("orchestration projector", () => { it("applies thread.created events", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); const next = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: now, commandId: "cmd-thread-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5-codex", branch: null, worktreePath: null, createdAt: now, updatedAt: now, }, }), ), ); expect(next.snapshotSequence).toBe(1); expect(next.threads).toEqual([ { id: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5-codex", branch: null, worktreePath: null, latestTurnId: null, createdAt: now, updatedAt: now, deletedAt: null, messages: [], activities: [], checkpoints: [], session: null, }, ]); }); it("fails when event payload cannot be decoded by runtime schema", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); await expect( Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: now, commandId: "cmd-invalid", payload: { // missing required threadId projectId: "project-1", title: "demo", model: "gpt-5-codex", branch: null, worktreePath: null, createdAt: now, updatedAt: now, }, }), ), ), ).rejects.toBeDefined(); }); it("keeps projector forward-compatible for unhandled event types", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); const next = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 7, type: "thread.turn-start-requested", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-01-01T00:00:00.000Z", commandId: "cmd-unhandled", payload: { threadId: "thread-1", messageId: "message-1", createdAt: "2026-01-01T00:00:00.000Z", }, }), ), ); expect(next.snapshotSequence).toBe(7); expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z"); expect(next.threads).toEqual([]); }); it("tracks latest turn id from session lifecycle events", async () => { const createdAt = "2026-02-23T08:00:00.000Z"; const startedAt = "2026-02-23T08:00:05.000Z"; const model = createEmptyReadModel(createdAt); const afterCreate = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5.3-codex", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }), ), ); const afterRunning = await Effect.runPromise( projectEvent( afterCreate, makeEvent({ sequence: 2, type: "thread.session-set", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: startedAt, commandId: "cmd-running", payload: { threadId: "thread-1", session: { threadId: "thread-1", status: "running", providerName: "codex", providerSessionId: "session-1", providerThreadId: "provider-thread-1", activeTurnId: "turn-1", lastError: null, updatedAt: startedAt, }, }, }), ), ); const thread = afterRunning.threads[0]; expect(thread?.latestTurnId).toBe("turn-1"); expect(thread?.session?.status).toBe("running"); }); it("marks assistant messages completed with non-streaming updates", async () => { const createdAt = "2026-02-23T09:00:00.000Z"; const deltaAt = "2026-02-23T09:00:01.000Z"; const completeAt = "2026-02-23T09:00:03.500Z"; const model = createEmptyReadModel(createdAt); const afterCreate = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5.3-codex", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }), ), ); const afterDelta = await Effect.runPromise( projectEvent( afterCreate, makeEvent({ sequence: 2, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: deltaAt, commandId: "cmd-delta", payload: { threadId: "thread-1", messageId: "assistant:msg-1", role: "assistant", text: "hello", turnId: "turn-1", streaming: true, createdAt: deltaAt, updatedAt: deltaAt, }, }), ), ); const afterComplete = await Effect.runPromise( projectEvent( afterDelta, makeEvent({ sequence: 3, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: completeAt, commandId: "cmd-complete", payload: { threadId: "thread-1", messageId: "assistant:msg-1", role: "assistant", text: "", turnId: "turn-1", streaming: false, createdAt: completeAt, updatedAt: completeAt, }, }), ), ); const message = afterComplete.threads[0]?.messages[0]; expect(message?.id).toBe("assistant:msg-1"); expect(message?.text).toBe("hello"); expect(message?.streaming).toBe(false); expect(message?.updatedAt).toBe(completeAt); }); it("prunes reverted turn messages from in-memory thread snapshot", async () => { const createdAt = "2026-02-23T10:00:00.000Z"; const model = createEmptyReadModel(createdAt); const afterCreate = await Effect.runPromise( projectEvent( model, makeEvent({ sequence: 1, type: "thread.created", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", payload: { threadId: "thread-1", projectId: "project-1", title: "demo", model: "gpt-5.3-codex", branch: null, worktreePath: null, createdAt, updatedAt: createdAt, }, }), ), ); const events: ReadonlyArray<OrchestrationEvent> = [ makeEvent({ sequence: 2, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:01.000Z", commandId: "cmd-user-1", payload: { threadId: "thread-1", messageId: "user-msg-1", role: "user", text: "First edit", turnId: null, streaming: false, createdAt: "2026-02-23T10:00:01.000Z", updatedAt: "2026-02-23T10:00:01.000Z", }, }), makeEvent({ sequence: 3, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.000Z", commandId: "cmd-assistant-1", payload: { threadId: "thread-1", messageId: "assistant-msg-1", role: "assistant", text: "Updated README to v2.\n", turnId: "turn-1", streaming: false, createdAt: "2026-02-23T10:00:02.000Z", updatedAt: "2026-02-23T10:00:02.000Z", }, }), makeEvent({ sequence: 4, type: "thread.turn-diff-completed", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.500Z", commandId: "cmd-turn-1-complete", payload: { threadId: "thread-1", turnId: "turn-1", checkpointTurnCount: 1, checkpointRef: "refs/t3/checkpoints/thread-1/turn/1", status: "ready", files: [], assistantMessageId: "assistant-msg-1", completedAt: "2026-02-23T10:00:02.500Z", }, }), makeEvent({ sequence: 5, type: "thread.activity-appended", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.750Z", commandId: "cmd-activity-1", payload: { threadId: "thread-1", activity: { id: "activity-1", tone: "tool", kind: "tool.started", summary: "Edit file started", payload: { toolKind: "command" }, turnId: "turn-1", createdAt: "2026-02-23T10:00:02.750Z", }, }, }), makeEvent({ sequence: 6, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:03.000Z", commandId: "cmd-user-2", payload: { threadId: "thread-1", messageId: "user-msg-2", role: "user", text: "Second edit", turnId: null, streaming: false, createdAt: "2026-02-23T10:00:03.000Z", updatedAt: "2026-02-23T10:00:03.000Z", }, }), makeEvent({ sequence: 7, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.000Z", commandId: "cmd-assistant-2", payload: { threadId: "thread-1", messageId: "assistant-msg-2", role: "assistant", text: "Updated README to v3.\n", turnId: "turn-2", streaming: false, createdAt: "2026-02-23T10:00:04.000Z", updatedAt: "2026-02-23T10:00:04.000Z", }, }), makeEvent({ sequence: 8, type: "thread.turn-diff-completed", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.500Z", commandId: "cmd-turn-2-complete", payload: { threadId: "thread-1", turnId: "turn-2", checkpointTurnCount: 2, checkpointRef: "refs/t3/checkpoints/thread-1/turn/2", status: "ready", files: [], assistantMessageId: "assistant-msg-2", completedAt: "2026-02-23T10:00:04.500Z", }, }), makeEvent({ sequence: 9, type: "thread.activity-appended", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.750Z", commandId: "cmd-activity-2", payload: { threadId: "thread-1", activity: { id: "activity-2", tone: "tool", kind: "tool.completed", summary: "Edit file complete", payload: { toolKind: "command" }, turnId: "turn-2", createdAt: "2026-02-23T10:00:04.750Z", }, }, }), makeEvent({ sequence: 10, type: "thread.reverted", aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:05.000Z", commandId: "cmd-revert", payload: { threadId: "thread-1", turnCount: 1, }, }), ]; const afterRevert = await events.reduce<Promise<ReturnType<typeof createEmptyReadModel>>>( (statePromise, event) => statePromise.then((state) => Effect.runPromise(projectEvent(state, event))), Promise.resolve(afterCreate), ); const thread = afterRevert.threads[0]; expect(thread?.messages.map((message) => ({ role: message.role, text: message.text }))).toEqual( [ { role: "user", text: "First edit" }, { role: "assistant", text: "Updated README to v2.\n" }, ], ); expect( thread?.activities.map((activity) => ({ id: activity.id, turnId: activity.turnId })),