Julius Marmingemigrate to single turn store1d975ad8 months agoHistory
969 lines · 33.0 KB · TypeScriptRawDownload
import {  CheckpointRef,  CommandId,  CorrelationId,  EventId,  MessageId,  ProviderSessionId,  ProviderThreadId,  ProjectId,  ThreadId,  TurnId,} from "@t3tools/contracts";import { NodeServices } from "@effect/platform-node";import { assert, it } from "@effect/vitest";import { Effect, Layer } from "effect";import fs from "node:fs";import os from "node:os";import path from "node:path";import * as SqlClient from "effect/unstable/sql/SqlClient"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";import {  makeSqlitePersistenceLive,  SqlitePersistenceMemory,} from "../../persistence/Layers/Sqlite.ts";import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import {  ORCHESTRATION_PROJECTOR_NAMES,  OrchestrationProjectionPipelineLive,} from "./ProjectionPipeline.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts"; const projectionLayer = it.layer(  OrchestrationProjectionPipelineLive.pipe(    Layer.provideMerge(OrchestrationEventStoreLive),    Layer.provideMerge(SqlitePersistenceMemory),  ),); projectionLayer("OrchestrationProjectionPipeline", (it) => {  it.effect("bootstraps all projection states and writes projection rows", () =>    Effect.gen(function* () {      const projectionPipeline = yield* OrchestrationProjectionPipeline;      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const now = new Date().toISOString();       yield* eventStore.append({        type: "project.created",        eventId: EventId.makeUnsafe("evt-1"),        aggregateKind: "project",        aggregateId: ProjectId.makeUnsafe("project-1"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-1"),        causationEventId: null,        correlationId: CommandId.makeUnsafe("cmd-1"),        metadata: {},        payload: {          projectId: ProjectId.makeUnsafe("project-1"),          title: "Project 1",          workspaceRoot: "/tmp/project-1",          defaultModel: null,          scripts: [],          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.created",        eventId: EventId.makeUnsafe("evt-2"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-1"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-2"),        causationEventId: null,        correlationId: CommandId.makeUnsafe("cmd-2"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-1"),          projectId: ProjectId.makeUnsafe("project-1"),          title: "Thread 1",          model: "gpt-5-codex",          branch: null,          worktreePath: null,          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.message-sent",        eventId: EventId.makeUnsafe("evt-3"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-1"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-3"),        causationEventId: null,        correlationId: CommandId.makeUnsafe("cmd-3"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-1"),          messageId: MessageId.makeUnsafe("message-1"),          role: "assistant",          text: "hello",          turnId: null,          streaming: false,          createdAt: now,          updatedAt: now,        },      });       yield* projectionPipeline.bootstrap;       const projectRows = yield* sql<{        readonly projectId: string;        readonly title: string;        readonly scriptsJson: string;      }>`        SELECT          project_id AS "projectId",          title,          scripts_json AS "scriptsJson"        FROM projection_projects      `;      assert.deepEqual(projectRows, [        { projectId: "project-1", title: "Project 1", scriptsJson: "[]" },      ]);       const messageRows = yield* sql<{        readonly messageId: string;        readonly text: string;      }>`        SELECT          message_id AS "messageId",          text        FROM projection_thread_messages      `;      assert.deepEqual(messageRows, [{ messageId: "message-1", text: "hello" }]);       const stateRows = yield* sql<{        readonly projector: string;        readonly lastAppliedSequence: number;      }>`        SELECT          projector,          last_applied_sequence AS "lastAppliedSequence"        FROM projection_state        ORDER BY projector ASC      `;      assert.equal(stateRows.length, Object.keys(ORCHESTRATION_PROJECTOR_NAMES).length);      for (const row of stateRows) {        assert.equal(row.lastAppliedSequence, 3);      }    }),  );   it.effect("resumes from projector last_applied_sequence without replaying older events", () =>    Effect.gen(function* () {      const projectionPipeline = yield* OrchestrationProjectionPipeline;      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const now = new Date().toISOString();       yield* eventStore.append({        type: "project.created",        eventId: EventId.makeUnsafe("evt-a1"),        aggregateKind: "project",        aggregateId: ProjectId.makeUnsafe("project-a"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-a1"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-a1"),        metadata: {},        payload: {          projectId: ProjectId.makeUnsafe("project-a"),          title: "Project A",          workspaceRoot: "/tmp/project-a",          defaultModel: null,          scripts: [],          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.created",        eventId: EventId.makeUnsafe("evt-a2"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-a"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-a2"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-a2"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-a"),          projectId: ProjectId.makeUnsafe("project-a"),          title: "Thread A",          model: "gpt-5-codex",          branch: null,          worktreePath: null,          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.message-sent",        eventId: EventId.makeUnsafe("evt-a3"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-a"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-a3"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-a3"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-a"),          messageId: MessageId.makeUnsafe("message-a"),          role: "assistant",          text: "hello",          turnId: null,          streaming: false,          createdAt: now,          updatedAt: now,        },      });       yield* projectionPipeline.bootstrap;       yield* eventStore.append({        type: "thread.message-sent",        eventId: EventId.makeUnsafe("evt-a4"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-a"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-a4"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-a4"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-a"),          messageId: MessageId.makeUnsafe("message-a"),          role: "assistant",          text: " world",          turnId: null,          streaming: true,          createdAt: now,          updatedAt: now,        },      });       yield* projectionPipeline.bootstrap;      yield* projectionPipeline.bootstrap;       const messageRows = yield* sql<{ readonly text: string }>`        SELECT text FROM projection_thread_messages WHERE message_id = 'message-a'      `;      assert.deepEqual(messageRows, [{ text: "hello world" }]);       const stateRows = yield* sql<{        readonly projector: string;        readonly lastAppliedSequence: number;      }>`        SELECT          projector,          last_applied_sequence AS "lastAppliedSequence"        FROM projection_state      `;      const maxSequenceRows = yield* sql<{ readonly maxSequence: number }>`        SELECT MAX(sequence) AS "maxSequence" FROM orchestration_events      `;      const maxSequence = maxSequenceRows[0]?.maxSequence ?? 0;      for (const row of stateRows) {        assert.equal(row.lastAppliedSequence, maxSequence);      }    }),  );   it.effect("keeps accumulated assistant text when completion payload text is empty", () =>    Effect.gen(function* () {      const projectionPipeline = yield* OrchestrationProjectionPipeline;      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const now = new Date().toISOString();       yield* eventStore.append({        type: "project.created",        eventId: EventId.makeUnsafe("evt-empty-1"),        aggregateKind: "project",        aggregateId: ProjectId.makeUnsafe("project-empty"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-empty-1"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-empty-1"),        metadata: {},        payload: {          projectId: ProjectId.makeUnsafe("project-empty"),          title: "Project Empty",          workspaceRoot: "/tmp/project-empty",          defaultModel: null,          scripts: [],          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.created",        eventId: EventId.makeUnsafe("evt-empty-2"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-empty"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-empty-2"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-empty-2"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-empty"),          projectId: ProjectId.makeUnsafe("project-empty"),          title: "Thread Empty",          model: "gpt-5-codex",          branch: null,          worktreePath: null,          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.message-sent",        eventId: EventId.makeUnsafe("evt-empty-3"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-empty"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-empty-3"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-empty-3"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-empty"),          messageId: MessageId.makeUnsafe("assistant-empty"),          role: "assistant",          text: "Hello",          turnId: null,          streaming: true,          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.message-sent",        eventId: EventId.makeUnsafe("evt-empty-4"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-empty"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-empty-4"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-empty-4"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-empty"),          messageId: MessageId.makeUnsafe("assistant-empty"),          role: "assistant",          text: " world",          turnId: null,          streaming: true,          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.message-sent",        eventId: EventId.makeUnsafe("evt-empty-5"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-empty"),        occurredAt: now,        commandId: CommandId.makeUnsafe("cmd-empty-5"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-empty-5"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-empty"),          messageId: MessageId.makeUnsafe("assistant-empty"),          role: "assistant",          text: "",          turnId: null,          streaming: false,          createdAt: now,          updatedAt: now,        },      });       yield* projectionPipeline.bootstrap;       const messageRows = yield* sql<{ readonly text: string; readonly isStreaming: unknown }>`        SELECT          text,          is_streaming AS "isStreaming"        FROM projection_thread_messages        WHERE message_id = 'assistant-empty'      `;      assert.equal(messageRows.length, 1);      assert.equal(messageRows[0]?.text, "Hello world");      assert.isFalse(Boolean(messageRows[0]?.isStreaming));    }),  );   it.effect("resolves turn-count conflicts when checkpoint completion rewrites provisional turns", () =>    Effect.gen(function* () {      const projectionPipeline = yield* OrchestrationProjectionPipeline;      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>        eventStore          .append(event)          .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));       yield* appendAndProject({        type: "project.created",        eventId: EventId.makeUnsafe("evt-conflict-1"),        aggregateKind: "project",        aggregateId: ProjectId.makeUnsafe("project-conflict"),        occurredAt: "2026-02-26T13:00:00.000Z",        commandId: CommandId.makeUnsafe("cmd-conflict-1"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-conflict-1"),        metadata: {},        payload: {          projectId: ProjectId.makeUnsafe("project-conflict"),          title: "Project Conflict",          workspaceRoot: "/tmp/project-conflict",          defaultModel: null,          scripts: [],          createdAt: "2026-02-26T13:00:00.000Z",          updatedAt: "2026-02-26T13:00:00.000Z",        },      });       yield* appendAndProject({        type: "thread.created",        eventId: EventId.makeUnsafe("evt-conflict-2"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-conflict"),        occurredAt: "2026-02-26T13:00:01.000Z",        commandId: CommandId.makeUnsafe("cmd-conflict-2"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-conflict-2"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-conflict"),          projectId: ProjectId.makeUnsafe("project-conflict"),          title: "Thread Conflict",          model: "gpt-5-codex",          branch: null,          worktreePath: null,          createdAt: "2026-02-26T13:00:01.000Z",          updatedAt: "2026-02-26T13:00:01.000Z",        },      });       yield* appendAndProject({        type: "thread.turn-interrupt-requested",        eventId: EventId.makeUnsafe("evt-conflict-3"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-conflict"),        occurredAt: "2026-02-26T13:00:02.000Z",        commandId: CommandId.makeUnsafe("cmd-conflict-3"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-conflict-3"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-conflict"),          turnId: TurnId.makeUnsafe("turn-interrupted"),          createdAt: "2026-02-26T13:00:02.000Z",        },      });       yield* appendAndProject({        type: "thread.message-sent",        eventId: EventId.makeUnsafe("evt-conflict-4"),        aggregateKind: "thread",        aggregateId: ThreadId.makeUnsafe("thread-conflict"),        occurredAt: "2026-02-26T13:00:03.000Z",        commandId: CommandId.makeUnsafe("cmd-conflict-4"),        causationEventId: null,        correlationId: CorrelationId.makeUnsafe("cmd-conflict-4"),        metadata: {},        payload: {          threadId: ThreadId.makeUnsafe("thread-conflict"),          messageId: MessageId.makeUnsafe("assistant-conflict"),          role: "assistant",          text: "done",          turnId: TurnId.makeUnsafe("turn-completed"),

Lines 1–500 of 969.