Julius MarmingeMerge main and use source proposed plans in the sidebar268d4f77 months agoHistory
1940 lines · 69.6 KB · TypeScriptRawDownload
import {  CheckpointRef,  CommandId,  CorrelationId,  EventId,  MessageId,  ProjectId,  ThreadId,  TurnId,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest";import { Effect, Layer, ManagedRuntime } 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";import { ServerConfig } from "../../config.ts"; const makeProjectionPipelineTestLayer = (stateDir: string) =>  OrchestrationProjectionPipelineLive.pipe(    Layer.provideMerge(OrchestrationEventStoreLive),    Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)),    Layer.provideMerge(SqlitePersistenceMemory),    Layer.provideMerge(NodeServices.layer),  ); const runWithProjectionPipelineLayer = <A, E>(  stateDir: string,  effect: Effect.Effect<    A,    E,    OrchestrationProjectionPipeline | OrchestrationEventStore | SqlClient.SqlClient  >,) =>  Effect.acquireUseRelease(    Effect.sync(() => ManagedRuntime.make(makeProjectionPipelineTestLayer(stateDir))),    (runtime) => Effect.promise(() => runtime.runPromise(effect)),    (runtime) => Effect.promise(() => runtime.dispose()),  ); const projectionLayer = it.layer(makeProjectionPipelineTestLayer(process.cwd())); 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",          runtimeMode: "full-access",          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("stores message attachment references without mutating payloads", () =>    Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(      Effect.flatMap((stateDir) =>        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: "thread.message-sent",            eventId: EventId.makeUnsafe("evt-attachments"),            aggregateKind: "thread",            aggregateId: ThreadId.makeUnsafe("thread-attachments"),            occurredAt: now,            commandId: CommandId.makeUnsafe("cmd-attachments"),            causationEventId: null,            correlationId: CommandId.makeUnsafe("cmd-attachments"),            metadata: {},            payload: {              threadId: ThreadId.makeUnsafe("thread-attachments"),              messageId: MessageId.makeUnsafe("message-attachments"),              role: "user",              text: "Inspect this",              attachments: [                {                  type: "image",                  id: "thread-attachments-att-1",                  name: "example.png",                  mimeType: "image/png",                  sizeBytes: 5,                },              ],              turnId: null,              streaming: false,              createdAt: now,              updatedAt: now,            },          });           yield* projectionPipeline.bootstrap;           const rows = yield* sql<{            readonly attachmentsJson: string | null;          }>`            SELECT              attachments_json AS "attachmentsJson"            FROM projection_thread_messages            WHERE message_id = 'message-attachments'          `;          assert.equal(rows.length, 1);          assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [            {              type: "image",              id: "thread-attachments-att-1",              name: "example.png",              mimeType: "image/png",              sizeBytes: 5,            },          ]);        }).pipe(          (effect) => runWithProjectionPipelineLayer(stateDir, effect),          Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),        ),      ),    ),  );   it.effect("preserves mixed image attachment metadata as-is", () =>    Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(      Effect.flatMap((stateDir) =>        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: "thread.message-sent",            eventId: EventId.makeUnsafe("evt-attachments-safe"),            aggregateKind: "thread",            aggregateId: ThreadId.makeUnsafe("thread-attachments-safe"),            occurredAt: now,            commandId: CommandId.makeUnsafe("cmd-attachments-safe"),            causationEventId: null,            correlationId: CommandId.makeUnsafe("cmd-attachments-safe"),            metadata: {},            payload: {              threadId: ThreadId.makeUnsafe("thread-attachments-safe"),              messageId: MessageId.makeUnsafe("message-attachments-safe"),              role: "user",              text: "Inspect this",              attachments: [                {                  type: "image",                  id: "thread-attachments-safe-att-1",                  name: "untrusted.exe",                  mimeType: "image/x-unknown",                  sizeBytes: 5,                },                {                  type: "image",                  id: "thread-attachments-safe-att-2",                  name: "not-image.png",                  mimeType: "image/png",                  sizeBytes: 5,                },              ],              turnId: null,              streaming: false,              createdAt: now,              updatedAt: now,            },          });           yield* projectionPipeline.bootstrap;           const rows = yield* sql<{            readonly attachmentsJson: string | null;          }>`            SELECT              attachments_json AS "attachmentsJson"            FROM projection_thread_messages            WHERE message_id = 'message-attachments-safe'          `;          assert.equal(rows.length, 1);          assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [            {              type: "image",              id: "thread-attachments-safe-att-1",              name: "untrusted.exe",              mimeType: "image/x-unknown",              sizeBytes: 5,            },            {              type: "image",              id: "thread-attachments-safe-att-2",              name: "not-image.png",              mimeType: "image/png",              sizeBytes: 5,            },          ]);        }).pipe(          (effect) => runWithProjectionPipelineLayer(stateDir, effect),          Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),        ),      ),    ),  );   it.effect(    "passes explicit empty attachment arrays through the projection pipeline to clear attachments",    () =>      Effect.gen(function* () {        const projectionPipeline = yield* OrchestrationProjectionPipeline;        const eventStore = yield* OrchestrationEventStore;        const sql = yield* SqlClient.SqlClient;        const now = new Date().toISOString();        const later = new Date(Date.now() + 1_000).toISOString();         yield* eventStore.append({          type: "project.created",          eventId: EventId.makeUnsafe("evt-clear-attachments-1"),          aggregateKind: "project",          aggregateId: ProjectId.makeUnsafe("project-clear-attachments"),          occurredAt: now,          commandId: CommandId.makeUnsafe("cmd-clear-attachments-1"),          causationEventId: null,          correlationId: CommandId.makeUnsafe("cmd-clear-attachments-1"),          metadata: {},          payload: {            projectId: ProjectId.makeUnsafe("project-clear-attachments"),            title: "Project Clear Attachments",            workspaceRoot: "/tmp/project-clear-attachments",            defaultModel: null,            scripts: [],            createdAt: now,            updatedAt: now,          },        });         yield* eventStore.append({          type: "thread.created",          eventId: EventId.makeUnsafe("evt-clear-attachments-2"),          aggregateKind: "thread",          aggregateId: ThreadId.makeUnsafe("thread-clear-attachments"),          occurredAt: now,          commandId: CommandId.makeUnsafe("cmd-clear-attachments-2"),          causationEventId: null,          correlationId: CommandId.makeUnsafe("cmd-clear-attachments-2"),          metadata: {},          payload: {            threadId: ThreadId.makeUnsafe("thread-clear-attachments"),            projectId: ProjectId.makeUnsafe("project-clear-attachments"),            title: "Thread Clear Attachments",            model: "gpt-5-codex",            runtimeMode: "full-access",            branch: null,            worktreePath: null,            createdAt: now,            updatedAt: now,          },        });         yield* eventStore.append({          type: "thread.message-sent",          eventId: EventId.makeUnsafe("evt-clear-attachments-3"),          aggregateKind: "thread",          aggregateId: ThreadId.makeUnsafe("thread-clear-attachments"),          occurredAt: now,          commandId: CommandId.makeUnsafe("cmd-clear-attachments-3"),          causationEventId: null,          correlationId: CommandId.makeUnsafe("cmd-clear-attachments-3"),          metadata: {},          payload: {            threadId: ThreadId.makeUnsafe("thread-clear-attachments"),            messageId: MessageId.makeUnsafe("message-clear-attachments"),            role: "user",            text: "Has attachments",            attachments: [              {                type: "image",                id: "thread-clear-attachments-att-1",                name: "clear.png",                mimeType: "image/png",                sizeBytes: 5,              },            ],            turnId: null,            streaming: false,            createdAt: now,            updatedAt: now,          },        });         yield* eventStore.append({          type: "thread.message-sent",          eventId: EventId.makeUnsafe("evt-clear-attachments-4"),          aggregateKind: "thread",          aggregateId: ThreadId.makeUnsafe("thread-clear-attachments"),          occurredAt: later,          commandId: CommandId.makeUnsafe("cmd-clear-attachments-4"),          causationEventId: null,          correlationId: CommandId.makeUnsafe("cmd-clear-attachments-4"),          metadata: {},          payload: {            threadId: ThreadId.makeUnsafe("thread-clear-attachments"),            messageId: MessageId.makeUnsafe("message-clear-attachments"),            role: "user",            text: "",            attachments: [],            turnId: null,            streaming: false,            createdAt: now,            updatedAt: later,          },        });         yield* projectionPipeline.bootstrap;         const rows = yield* sql<{          readonly attachmentsJson: string | null;        }>`          SELECT            attachments_json AS "attachmentsJson"          FROM projection_thread_messages          WHERE message_id = 'message-clear-attachments'        `;        assert.equal(rows.length, 1);        assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), []);      }),  );   it.effect("overwrites stored attachment references when a message updates attachments", () =>    Effect.sync(() =>      fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-")),    ).pipe(      Effect.flatMap((stateDir) =>        Effect.gen(function* () {          const projectionPipeline = yield* OrchestrationProjectionPipeline;          const eventStore = yield* OrchestrationEventStore;          const sql = yield* SqlClient.SqlClient;          const now = new Date().toISOString();          const later = new Date(Date.now() + 1_000).toISOString();           yield* eventStore.append({            type: "project.created",            eventId: EventId.makeUnsafe("evt-overwrite-1"),            aggregateKind: "project",            aggregateId: ProjectId.makeUnsafe("project-overwrite"),            occurredAt: now,            commandId: CommandId.makeUnsafe("cmd-overwrite-1"),            causationEventId: null,            correlationId: CommandId.makeUnsafe("cmd-overwrite-1"),            metadata: {},            payload: {              projectId: ProjectId.makeUnsafe("project-overwrite"),              title: "Project Overwrite",              workspaceRoot: "/tmp/project-overwrite",              defaultModel: null,              scripts: [],              createdAt: now,              updatedAt: now,            },          });           yield* eventStore.append({            type: "thread.created",            eventId: EventId.makeUnsafe("evt-overwrite-2"),            aggregateKind: "thread",            aggregateId: ThreadId.makeUnsafe("thread-overwrite"),            occurredAt: now,            commandId: CommandId.makeUnsafe("cmd-overwrite-2"),            causationEventId: null,            correlationId: CommandId.makeUnsafe("cmd-overwrite-2"),            metadata: {},            payload: {              threadId: ThreadId.makeUnsafe("thread-overwrite"),              projectId: ProjectId.makeUnsafe("project-overwrite"),              title: "Thread Overwrite",              model: "gpt-5-codex",              runtimeMode: "full-access",              branch: null,

Lines 1–500 of 1940.