import {  CheckpointRef,  CommandId,  CorrelationId,  EventId,  MessageId,  ProjectId,  ThreadId,  TurnId,  ProviderInstanceId,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest";import * as Effect from "effect/Effect";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Path from "effect/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 { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";import { ProjectionThreadMessageRepository } from "../../persistence/Services/ProjectionThreadMessages.ts";import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import {  ORCHESTRATION_PROJECTOR_NAMES,  OrchestrationProjectionPipelineLive,} from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";import { ServerConfig } from "../../config.ts"; const makeProjectionPipelinePrefixedTestLayer = (prefix: string) =>  OrchestrationProjectionPipelineLive.pipe(    Layer.provideMerge(OrchestrationEventStoreLive),    Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix })),    Layer.provideMerge(SqlitePersistenceMemory),    Layer.provideMerge(NodeServices.layer),  ); const exists = (filePath: string) =>  Effect.gen(function* () {    const fileSystem = yield* FileSystem.FileSystem;    const fileInfo = yield* Effect.result(fileSystem.stat(filePath));    return fileInfo._tag === "Success";  }); const BaseTestLayer = makeProjectionPipelinePrefixedTestLayer("t3-projection-pipeline-test-"); it.layer(BaseTestLayer)("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 = "2026-01-01T00:00:00.000Z";       yield* eventStore.append({        type: "project.created",        eventId: EventId.make("evt-1"),        aggregateKind: "project",        aggregateId: ProjectId.make("project-1"),        occurredAt: now,        commandId: CommandId.make("cmd-1"),        causationEventId: null,        correlationId: CommandId.make("cmd-1"),        metadata: {},        payload: {          projectId: ProjectId.make("project-1"),          title: "Project 1",          workspaceRoot: "/tmp/project-1",          defaultModelSelection: null,          scripts: [],          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.created",        eventId: EventId.make("evt-2"),        aggregateKind: "thread",        aggregateId: ThreadId.make("thread-1"),        occurredAt: now,        commandId: CommandId.make("cmd-2"),        causationEventId: null,        correlationId: CommandId.make("cmd-2"),        metadata: {},        payload: {          threadId: ThreadId.make("thread-1"),          projectId: ProjectId.make("project-1"),          title: "Thread 1",          modelSelection: {            instanceId: ProviderInstanceId.make("codex"),            model: "gpt-5-codex",          },          runtimeMode: "full-access",          branch: null,          worktreePath: null,          createdAt: now,          updatedAt: now,        },      });       yield* eventStore.append({        type: "thread.message-sent",        eventId: EventId.make("evt-3"),        aggregateKind: "thread",        aggregateId: ThreadId.make("thread-1"),        occurredAt: now,        commandId: CommandId.make("cmd-3"),        causationEventId: null,        correlationId: CommandId.make("cmd-3"),        metadata: {},        payload: {          threadId: ThreadId.make("thread-1"),          messageId: MessageId.make("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);      }       // Settled lifecycle through the DB pipeline: thread.settled writes the      // override + timestamp, thread.unsettled(user) flips to the active pin.      yield* eventStore.append({        type: "thread.settled",        eventId: EventId.make("evt-settle-1"),        aggregateKind: "thread",        aggregateId: ThreadId.make("thread-1"),        occurredAt: "2026-01-01T00:00:01.000Z",        commandId: CommandId.make("cmd-settle-1"),        causationEventId: null,        correlationId: CommandId.make("cmd-settle-1"),        metadata: {},        payload: {          threadId: ThreadId.make("thread-1"),          settledAt: "2026-01-01T00:00:01.000Z",          updatedAt: "2026-01-01T00:00:01.000Z",        },      });      yield* projectionPipeline.bootstrap;       const settledRows = yield* sql<{        readonly settledOverride: string | null;        readonly settledAt: string | null;      }>`        SELECT          settled_override AS "settledOverride",          settled_at AS "settledAt"        FROM projection_threads        WHERE thread_id = 'thread-1'      `;      assert.deepEqual(settledRows, [        { settledOverride: "settled", settledAt: "2026-01-01T00:00:01.000Z" },      ]);       yield* eventStore.append({        type: "thread.unsettled",        eventId: EventId.make("evt-unsettle-1"),        aggregateKind: "thread",        aggregateId: ThreadId.make("thread-1"),        occurredAt: "2026-01-01T00:00:02.000Z",        commandId: CommandId.make("cmd-unsettle-1"),        causationEventId: null,        correlationId: CommandId.make("cmd-unsettle-1"),        metadata: {},        payload: {          threadId: ThreadId.make("thread-1"),          reason: "user",          updatedAt: "2026-01-01T00:00:02.000Z",        },      });      yield* projectionPipeline.bootstrap;       const unsettledRows = yield* sql<{        readonly settledOverride: string | null;        readonly settledAt: string | null;      }>`        SELECT          settled_override AS "settledOverride",          settled_at AS "settledAt"        FROM projection_threads        WHERE thread_id = 'thread-1'      `;      assert.deepEqual(unsettledRows, [{ settledOverride: "active", settledAt: null }]);    }),  );}); it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shell-summary-cost-")))(  "OrchestrationProjectionPipeline shell-summary cost",  (it) => {    it.effect("does not hydrate large activity history for shell-only updates", () =>      Effect.gen(function* () {        const projectionPipeline = yield* OrchestrationProjectionPipeline;        const eventStore = yield* OrchestrationEventStore;        const sql = yield* SqlClient.SqlClient;        const projectId = ProjectId.make("project-shell-summary-cost");        const threadId = ThreadId.make("thread-shell-summary-cost");        const createdAt = "2026-08-19T00:00:00.000Z";        const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>          eventStore            .append(event)            .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));        const readPendingUserInputCount = Effect.gen(function* () {          const rows = yield* sql<{ readonly count: number }>`          SELECT pending_user_input_count AS "count"          FROM projection_threads          WHERE thread_id = ${threadId}        `;          return rows[0]?.count;        });        const readLatestUserMessageAt = Effect.gen(function* () {          const rows = yield* sql<{ readonly latestUserMessageAt: string | null }>`          SELECT latest_user_message_at AS "latestUserMessageAt"          FROM projection_threads          WHERE thread_id = ${threadId}        `;          return rows[0]?.latestUserMessageAt;        });         yield* appendAndProject({          type: "project.created",          eventId: EventId.make("evt-shell-summary-cost-project"),          aggregateKind: "project",          aggregateId: projectId,          occurredAt: createdAt,          commandId: CommandId.make("cmd-shell-summary-cost-project"),          causationEventId: null,          correlationId: CorrelationId.make("cmd-shell-summary-cost-project"),          metadata: {},          payload: {            projectId,            title: "Shell summary cost",            workspaceRoot: "/tmp/project-shell-summary-cost",            defaultModelSelection: null,            scripts: [],            createdAt,            updatedAt: createdAt,          },        });        yield* appendAndProject({          type: "thread.created",          eventId: EventId.make("evt-shell-summary-cost-thread"),          aggregateKind: "thread",          aggregateId: threadId,          occurredAt: createdAt,          commandId: CommandId.make("cmd-shell-summary-cost-thread"),          causationEventId: null,          correlationId: CorrelationId.make("cmd-shell-summary-cost-thread"),          metadata: {},          payload: {            threadId,            projectId,            title: "Shell summary cost",            modelSelection: {              instanceId: ProviderInstanceId.make("codex"),              model: "gpt-5-codex",            },            runtimeMode: "full-access",            interactionMode: "default",            branch: null,            worktreePath: null,            createdAt,            updatedAt: createdAt,          },        });         yield* sql`        WITH RECURSIVE history(index_value) AS (          SELECT 1          UNION ALL          SELECT index_value + 1          FROM history          WHERE index_value < 10000        )        INSERT INTO projection_thread_activities (          activity_id,          thread_id,          turn_id,          tone,          kind,          summary,          payload_json,          sequence,          created_at,          user_input_request_id,          user_input_state        )        SELECT          'activity-shell-summary-history-' || printf('%05d', index_value),          ${threadId},          NULL,          'info',          'user-input.resolved',          'Historical user input',          json_object(            'requestId',            'request-shell-summary-history-' || printf('%05d', index_value),            'blob',            replace(hex(zeroblob(2048)), '00', 'x')          ),          index_value,          ${createdAt},          'request-shell-summary-history-' || printf('%05d', index_value),          'cleared'        FROM history      `;        // If the shell update reads the collection, this sentinel makes the        // accidental JSON decode fail instead of hiding behind timing variance.        yield* sql`        INSERT INTO projection_thread_activities (          activity_id,          thread_id,          turn_id,          tone,          kind,          summary,          payload_json,          sequence,          created_at        ) VALUES (          'activity-shell-summary-invalid-sentinel',          ${threadId},          NULL,          'info',          'user-input.requested',          'Must not be decoded',          '{',          NULL,          ${createdAt}        )      `;         yield* sql`          INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)          VALUES (${threadId}, 0)          ON CONFLICT (thread_id)          DO UPDATE SET pending_count = 0        `;         yield* appendAndProject({          type: "thread.message-sent",          eventId: EventId.make("evt-shell-summary-assistant-message"),          aggregateKind: "thread",          aggregateId: threadId,          occurredAt: "2026-08-19T00:00:01.000Z",          commandId: CommandId.make("cmd-shell-summary-assistant-message"),          causationEventId: null,          correlationId: CorrelationId.make("cmd-shell-summary-assistant-message"),          metadata: {},          payload: {            threadId,            messageId: MessageId.make("message-shell-summary-assistant"),            role: "assistant",            text: "Still working",            attachments: [],            turnId: null,            streaming: false,            createdAt: "2026-08-19T00:00:01.000Z",            updatedAt: "2026-08-19T00:00:01.000Z",          },        });        assert.equal(yield* readPendingUserInputCount, 0);         yield* appendAndProject({          type: "thread.message-sent",          eventId: EventId.make("evt-shell-summary-user-message-collision"),          aggregateKind: "thread",          aggregateId: threadId,          occurredAt: "2026-08-19T00:00:01.100Z",          commandId: CommandId.make("cmd-shell-summary-user-message-collision"),          causationEventId: null,          correlationId: CorrelationId.make("cmd-shell-summary-user-message-collision"),          metadata: {},          payload: {            threadId,            messageId: MessageId.make("message-shell-summary-collision"),            role: "user",            text: "A user message",            attachments: [],            turnId: null,            streaming: false,            createdAt: "2026-08-19T00:00:01.100Z",            updatedAt: "2026-08-19T00:00:01.100Z",          },        });        assert.equal(yield* readLatestUserMessageAt, "2026-08-19T00:00:01.100Z");         yield* appendAndProject({          type: "thread.message-sent",          eventId: EventId.make("evt-shell-summary-assistant-message-collision"),          aggregateKind: "thread",          aggregateId: threadId,          occurredAt: "2026-08-19T00:00:01.200Z",          commandId: CommandId.make("cmd-shell-summary-assistant-message-collision"),          causationEventId: null,          correlationId: CorrelationId.make("cmd-shell-summary-assistant-message-collision"),          metadata: {},          payload: {            threadId,            messageId: MessageId.make("message-shell-summary-collision"),            role: "assistant",            text: "The row was replaced",            attachments: [],            turnId: null,            streaming: false,            createdAt: "2026-08-19T00:00:01.200Z",            updatedAt: "2026-08-19T00:00:01.200Z",          },        });        assert.equal(yield* readLatestUserMessageAt, null);         yield* appendAndProject({          type: "thread.activity-appended",          eventId: EventId.make("evt-shell-summary-routine-activity"),          aggregateKind: "thread",          aggregateId: threadId,          occurredAt: "2026-08-19T00:00:01.000Z",          commandId: CommandId.make("cmd-shell-summary-routine-activity"),          causationEventId: null,          correlationId: CorrelationId.make("cmd-shell-summary-routine-activity"),          metadata: {},          payload: {            threadId,            activity: {              id: EventId.make("activity-shell-summary-routine"),              tone: "info",              kind: "task.progress",              summary: "Running tests",              payload: { taskId: "task-1", status: "running" },              turnId: null,              createdAt: "2026-08-19T00:00:01.000Z",            },          },        });        assert.equal(yield* readPendingUserInputCount, 0);         yield* appendAndProject({          type: "thread.activity-appended",          eventId: EventId.make("evt-shell-summary-user-input-requested"),          aggregateKind: "thread",          aggregateId: threadId,          occurredAt: "2026-08-19T00:00:02.000Z",

Lines 1–500 of 3363. Load more