import {  CheckpointRef,  CommandId,  CorrelationId,  EventId,  MessageId,  ProviderSessionId,  ProviderThreadId,  ProjectId,  ThreadId,  TurnId,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";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";import { ServerConfig, type ServerConfigShape } from "../../config.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("materializes message image attachments into stateDir and stores URL references", () =>    Effect.gen(function* () {      const projectionPipeline = yield* OrchestrationProjectionPipeline;      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const now = new Date().toISOString();      const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"));       const serverConfig = {        mode: "web",        port: 0,        host: undefined,        cwd: "/tmp/project-attachments",        keybindingsConfigPath: path.join(stateDir, "keybindings.json"),        stateDir,        staticDir: undefined,        devUrl: undefined,        noBrowser: true,        authToken: undefined,        autoBootstrapProjectFromCwd: false,        logWebSocketEvents: false,      } satisfies ServerConfigShape;       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",              name: "example.png",              mimeType: "image/png",              sizeBytes: 5,              dataUrl: "data:image/png;base64,SGVsbG8=",            },          ],          turnId: null,          streaming: false,          createdAt: now,          updatedAt: now,        },      });       yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));       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",          name: "example.png",          mimeType: "image/png",          sizeBytes: 5,          dataUrl: "/attachments/thread-attachments/message-attachments-0.png",        },      ]);       const attachmentPath = path.join(        stateDir,        "attachments",        "thread-attachments",        "message-attachments-0.png",      );      assert.equal(fs.existsSync(attachmentPath), true);      assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("SGVsbG8=", "base64"));      fs.rmSync(stateDir, { recursive: true, force: true });    }),  );   it.effect("materializes only image data URLs and uses a safe extension whitelist", () =>    Effect.gen(function* () {      const projectionPipeline = yield* OrchestrationProjectionPipeline;      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const now = new Date().toISOString();      const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"));       const serverConfig = {        mode: "web",        port: 0,        host: undefined,        cwd: "/tmp/project-attachments",        keybindingsConfigPath: path.join(stateDir, "keybindings.json"),        stateDir,        staticDir: undefined,        devUrl: undefined,        noBrowser: true,        authToken: undefined,        autoBootstrapProjectFromCwd: false,        logWebSocketEvents: false,      } satisfies ServerConfigShape;       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",              name: "untrusted.exe",              mimeType: "image/x-unknown",              sizeBytes: 5,              dataUrl: "data:image/x-unknown;base64,SGVsbG8=",            },            {              type: "image",              name: "not-image.png",              mimeType: "image/png",              sizeBytes: 5,              dataUrl: "data:text/plain;base64,SGVsbG8=",            },          ],          turnId: null,          streaming: false,          createdAt: now,          updatedAt: now,        },      });       yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));       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",          name: "untrusted.exe",          mimeType: "image/x-unknown",          sizeBytes: 5,          dataUrl: "/attachments/thread-attachments-safe/message-attachments-safe-0.bin",        },        {          type: "image",          name: "not-image.png",          mimeType: "image/png",          sizeBytes: 5,          dataUrl: "data:text/plain;base64,SGVsbG8=",        },      ]);       const firstAttachmentPath = path.join(        stateDir,        "attachments",        "thread-attachments-safe",        "message-attachments-safe-0.bin",      );      assert.equal(fs.existsSync(firstAttachmentPath), true);      assert.deepEqual(fs.readFileSync(firstAttachmentPath), Buffer.from("SGVsbG8=", "base64"));       const secondAttachmentPath = path.join(        stateDir,        "attachments",        "thread-attachments-safe",        "message-attachments-safe-1.png",      );      assert.equal(fs.existsSync(secondAttachmentPath), false);      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",            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",                name: "clear.png",                mimeType: "image/png",                sizeBytes: 5,                dataUrl: "data:image/png;base64,SGVsbG8=",              },            ],            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 attachment file bytes when a message updates the same attachment index",    () =>      Effect.gen(function* () {        const projectionPipeline = yield* OrchestrationProjectionPipeline;        const eventStore = yield* OrchestrationEventStore;        const stateDir = fs.mkdtempSync(          path.join(os.tmpdir(), "t3-projection-attachments-overwrite-"),        );        const now = new Date().toISOString();        const later = new Date(Date.now() + 1_000).toISOString();         const serverConfig = {          mode: "web",          port: 0,          host: undefined,          cwd: "/tmp/project-attachments",          keybindingsConfigPath: path.join(stateDir, "keybindings.json"),          stateDir,          staticDir: undefined,          devUrl: undefined,

Lines 1–500 of 1970. Load more