import {  CommandId,  EventId,  ProjectId,  ProviderInstanceId,  ThreadId,  TurnItemId,  type OrchestrationV2DomainEvent,} from "@t3tools/contracts";import { assert, it } from "@effect/vitest";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Fiber from "effect/Fiber";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as SqlClient from "effect/unstable/sql/SqlClient";import * as Statement from "effect/unstable/sql/Statement"; import {  LIVE_STREAM_MAX_ITEMS,  LiveStreamBufferError,} from "../../orchestration/LiveStreamBudget.ts";import { PersistenceDecodeError } from "../Errors.ts";import { toShellApplicationEvent } from "../../orchestration-v2/ShellStream.ts";import { OrchestrationEventStore } from "../Services/OrchestrationEventStore.ts";import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts";import { SqlitePersistenceMemory } from "./Sqlite.ts";const isPersistenceDecodeError = Schema.is(PersistenceDecodeError);const isLiveStreamBufferError = Schema.is(LiveStreamBufferError); const TestLayer = OrchestrationEventStoreLive.pipe(Layer.provideMerge(SqlitePersistenceMemory));const layer = it.layer(TestLayer); layer("OrchestrationEventStore", (it) => {  it.effect("retains only shell metadata from oversized replay and live application events", () =>    Effect.scoped(      Effect.gen(function* () {        const store = yield* OrchestrationEventStore;        const afterSequence = yield* store.latestApplicationSequence;        const now = yield* DateTime.now;        const threadId = ThreadId.make("thread:large-shell-body");        const input = "x".repeat(9 * 1024 * 1024);        const event: OrchestrationV2DomainEvent = {          id: EventId.make("event:large-shell-body:replay"),          type: "turn-item.updated",          threadId,          occurredAt: now,          payload: {            id: TurnItemId.make("tool:large-shell-body"),            type: "command_execution",            threadId,            runId: null,            nodeId: null,            providerThreadId: null,            providerTurnId: null,            nativeItemRef: null,            parentItemId: null,            ordinal: 1,            status: "running",            title: "Large command",            input,            startedAt: now,            completedAt: null,            updatedAt: now,          },        };        const [replayed] = yield* store.appendAgentEvents({ events: [event] });        const pull = yield* Stream.toPull(          store.streamProjectedApplicationEvents({            afterSequence,            project: toShellApplicationEvent,          }),        );        assert.deepEqual(yield* pull, [{ sequence: replayed!.sequence, event: { threadId } }]);        const [live] = yield* store.appendAgentEvents({          events: [            {              ...event,              id: EventId.make("event:large-shell-body:live"),              payload: { ...event.payload, status: "completed", completedAt: now },            },          ],        });        yield* store.publishCommitted([live!]);        assert.deepEqual(yield* pull, [{ sequence: live!.sequence, event: { threadId } }]);        const durable = yield* store          .readAgentEvents({ threadId, afterSequence })          .pipe(Stream.runCollect);        assert.lengthOf(durable, 2);        for (const stored of durable) {          assert.equal(stored.event.type, "turn-item.updated");          if (stored.event.type !== "turn-item.updated") return;          assert.equal(stored.event.payload.type, "command_execution");          if (stored.event.payload.type !== "command_execution") return;          assert.equal(stored.event.payload.input, input);        }      }),    ),  );   it.effect("stores json columns as strings and replays CLI-origin events", () =>    Effect.gen(function* () {      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const now = "2026-01-01T00:00:00.000Z";       const appended = yield* eventStore.append({        type: "project.created",        eventId: EventId.make("evt-store-roundtrip"),        aggregateKind: "project",        aggregateId: ProjectId.make("project-roundtrip"),        occurredAt: now,        commandId: CommandId.make("cmd-store-roundtrip"),        causationEventId: null,        correlationId: CommandId.make("cmd-store-roundtrip"),        metadata: {          adapterKey: "codex",          origin: {            surface: "cli",          },        },        payload: {          projectId: ProjectId.make("project-roundtrip"),          title: "Roundtrip Project",          workspaceRoot: "/tmp/project-roundtrip",          defaultModelSelection: null,          scripts: [],          createdAt: now,          updatedAt: now,        },      });       const storedRows = yield* sql<{        readonly payloadJson: string;        readonly metadataJson: string;      }>`        SELECT          payload_json AS "payloadJson",          metadata_json AS "metadataJson"        FROM orchestration_events        WHERE event_id = ${appended.eventId}      `;      assert.equal(storedRows.length, 1);      assert.equal(typeof storedRows[0]?.payloadJson, "string");      assert.equal(typeof storedRows[0]?.metadataJson, "string");       const replayed = yield* Stream.runCollect(eventStore.readFromSequence(0, 10)).pipe(        Effect.map((chunk) => Array.from(chunk)),      );      assert.equal(replayed.length, 1);      assert.equal(replayed[0]?.type, "project.created");      assert.equal(replayed[0]?.metadata.adapterKey, "codex");      assert.deepEqual(replayed[0]?.metadata.origin, { surface: "cli" });    }),  );   it.effect("fails with PersistenceDecodeError when stored json is invalid", () =>    Effect.gen(function* () {      const eventStore = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const now = "2026-01-01T00:00:00.000Z";       yield* sql`        INSERT INTO orchestration_events (          event_id,          aggregate_kind,          stream_id,          stream_version,          event_type,          occurred_at,          command_id,          causation_event_id,          correlation_id,          actor_kind,          payload_json,          metadata_json        )        VALUES (          ${EventId.make("evt-store-invalid-json")},          ${"project"},          ${ProjectId.make("project-invalid-json")},          ${0},          ${"project.created"},          ${now},          ${CommandId.make("cmd-store-invalid-json")},          ${null},          ${null},          ${"server"},          ${"{"},          ${"{}"}        )      `;       const replayResult = yield* Effect.result(        Stream.runCollect(eventStore.readFromSequence(0, 10)),      );      assert.equal(replayResult._tag, "Failure");      if (replayResult._tag === "Failure") {        assert.ok(isPersistenceDecodeError(replayResult.failure));        assert.ok(          replayResult.failure.operation.includes(            "OrchestrationEventStore.readFromSequence:decodeRows",          ),        );      }    }),  );   it.effect("orders project and V2 agent events in the retained application event source", () =>    Effect.gen(function* () {      const eventStore = yield* OrchestrationEventStore;      const projectId = ProjectId.make("project-shared-stream");      const threadId = ThreadId.make("thread-shared-stream");      const providerInstanceId = ProviderInstanceId.make("codex");      const occurredAt = DateTime.makeUnsafe("2026-01-02T00:00:00.000Z");      const now = DateTime.formatIso(occurredAt);      const baselineSequence = yield* eventStore.latestApplicationSequence;       const projectEvent = yield* eventStore.append({        type: "project.created",        eventId: EventId.make("event-project-shared-stream"),        aggregateKind: "project",        aggregateId: projectId,        occurredAt: now,        commandId: CommandId.make("command-project-shared-stream"),        causationEventId: null,        correlationId: CommandId.make("command-project-shared-stream"),        metadata: {},        payload: {          projectId,          title: "Shared stream",          workspaceRoot: "/tmp/shared-stream",          defaultModelSelection: null,          scripts: [],          createdAt: now,          updatedAt: now,        },      });      const [threadEvent] = yield* eventStore.appendAgentEvents({        commandId: CommandId.make("command-thread-shared-stream"),        events: [          {            id: EventId.make("event-thread-shared-stream"),            type: "thread.created",            threadId,            providerInstanceId,            occurredAt,            payload: {              id: threadId,              projectId,              title: "Thread",              providerInstanceId,              modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },              runtimeMode: "full-access",              interactionMode: "default",              branch: null,              worktreePath: null,              activeProviderThreadId: null,              lineage: {                rootThreadId: threadId,                parentThreadId: null,                relationshipToParent: null,              },              forkedFrom: null,              createdBy: "user",              creationSource: "web",              createdAt: occurredAt,              updatedAt: occurredAt,              archivedAt: null,              settledOverride: null,              settledAt: null,              lastVisitedAt: null,              deletedAt: null,            },          },        ],      });       const applicationEvents = yield* eventStore        .streamApplicationEvents({ afterSequence: baselineSequence })        .pipe(          Stream.take(2),          Stream.runCollect,          Effect.map((chunk) => Array.from(chunk)),        );      assert.deepEqual(        applicationEvents.map((event) => event.sequence),        [projectEvent.sequence, threadEvent!.sequence],      );      assert.isTrue("aggregateKind" in applicationEvents[0]!);      assert.isTrue("event" in applicationEvents[1]!);       const finiteReplay = yield* eventStore        .readApplicationEvents({          afterSequence: baselineSequence,          throughSequence: threadEvent!.sequence,        })        .pipe(          Stream.runCollect,          Effect.map((chunk) => Array.from(chunk)),        );      assert.deepEqual(        finiteReplay.map((event) => event.sequence),        [projectEvent.sequence, threadEvent!.sequence],      );       const legacyReplay = yield* eventStore.readFromSequence(projectEvent.sequence - 1).pipe(        Stream.runCollect,        Effect.map((chunk) => Array.from(chunk)),      );      assert.deepEqual(        legacyReplay.map((event) => event.type),        ["project.created"],      );    }),  );  it.effect("measures only a bounded thread replay before decoding its payloads", () =>    Effect.gen(function* () {      const store = yield* OrchestrationEventStore;      const sql = yield* SqlClient.SqlClient;      const threadId = ThreadId.make("measured-replay-thread");      const baseline = yield* store.latestApplicationSequence;      const unicodePayload = '{"text":"🦊"}';      const oversizedPayload = "{" + "x".repeat(1_048_576);      let streamVersion = 0;      const insert = Effect.fn(function* (input: {        readonly id: string;        readonly payload: string;        readonly type?: string;        readonly threadId?: string;        readonly version?: number;      }) {        const rows = yield* sql<{ readonly sequence: number }>`          INSERT INTO orchestration_events (            event_id, aggregate_kind, stream_id, stream_version, event_type,            occurred_at, actor_kind, payload_json, metadata_json, application_event_version          ) VALUES (            ${input.id}, 'thread', ${input.threadId ?? threadId},            ${streamVersion++}, ${input.type ?? "thread.metadata-updated"},            '2026-09-04T00:00:00.000Z', 'server', ${input.payload}, '{}', ${input.version ?? 2}          ) RETURNING sequence        `;        return rows[0]!.sequence;      });      const first = yield* insert({        id: "measured-replay-1",        payload: unicodePayload,        type: "thread.created",      });      yield* insert({        id: "measured-unrelated",        threadId: "other-replay-thread",        payload: oversizedPayload,      });      yield* insert({ id: "measured-legacy", payload: oversizedPayload, version: 1 });      const head = yield* insert({ id: "measured-replay-02", payload: oversizedPayload });      const later = yield* insert({ id: "measured-replay-003", payload: "{}" });      const range = { threadId, afterSequence: baseline, throughSequence: head, maxEvents: 128 };      assert.deepEqual(yield* store.getAgentReplayStats(range), {        eventCount: 2,        payloadBytes: Buffer.byteLength(unicodePayload) + Buffer.byteLength(oversizedPayload),        hasCreateEvent: true,      });      assert.deepEqual(yield* store.getAgentReplayStats({ ...range, afterSequence: first }), {        eventCount: 1,        payloadBytes: Buffer.byteLength(oversizedPayload),        hasCreateEvent: false,      });      assert.equal(        (yield* store.getAgentReplayStats({ ...range, throughSequence: later, maxEvents: 1 }))          .eventCount,        2,      );      assert.deepEqual(yield* store.getAgentReplayStats({ ...range, afterSequence: later }), {        eventCount: 0,        payloadBytes: 0,        hasCreateEvent: false,      });    }),  );}); for (const phase of ["high-water", "replay"] as const) {  it.effect(`bounds application live events while the ${phase} query is blocked`, () =>    Effect.scoped(      Effect.gen(function* () {        const store = yield* OrchestrationEventStore;        const now = "2026-01-03T00:00:00.000Z";        const projectId = ProjectId.make(`project:blocked-${phase}`);        const projectEvent = yield* store.append({          type: "project.created",          eventId: EventId.make(`event:blocked-${phase}:created`),          aggregateKind: "project",          aggregateId: projectId,          occurredAt: now,          commandId: CommandId.make(`command:blocked-${phase}`),          causationEventId: null,          correlationId: null,          metadata: {},          payload: {            projectId,            title: "Blocked stream",            workspaceRoot: "/tmp/blocked-stream",            defaultModelSelection: null,            scripts: [],            createdAt: now,            updatedAt: now,          },        });        if (projectEvent.type !== "project.created")          return yield* Effect.die("Expected project.created");        const readStarted = yield* Deferred.make<void>();        const readClosed = yield* Deferred.make<void>();        const blockRead: Statement.Transformer = (statement) => {          const [query] = statement.compile();          if (            query.includes("FROM orchestration_events") &&            query.includes("MAX(sequence)") === (phase === "high-water")          ) {            return Deferred.succeed(readStarted, undefined).pipe(              Effect.andThen(Effect.never),              Effect.ensuring(Deferred.succeed(readClosed, undefined)),              Effect.as(statement),            );          }          return Effect.succeed(statement);        };        const reader = yield* store          .streamApplicationEvents({ afterSequence: projectEvent.sequence })          .pipe(            Stream.provideService(Statement.CurrentTransformer, blockRead),            Stream.runDrain,            Effect.result,            Effect.forkScoped,          );        yield* Deferred.await(readStarted);        yield* store.publishCommitted(          Array.from({ length: LIVE_STREAM_MAX_ITEMS + 1 }, (_, index) => ({            ...projectEvent,            eventId: EventId.make(`event:blocked-${phase}:${index}`),            sequence: projectEvent.sequence + index + 1,          })),        );        yield* Deferred.await(readClosed);        const result = yield* Fiber.join(reader);        assert.equal(result._tag, "Failure");        if (result._tag === "Failure") {          assert.equal(result.failure._tag, "PersistenceSqlError");          assert.isTrue(isLiveStreamBufferError(result.failure.cause));        }      }),    ).pipe(Effect.provide(Layer.fresh(TestLayer))),  );}