import { describe, expect } from "bun:test"
import type { FileSystem } from "@opencode/core/filesystem"
import { DateTime, Effect, Fiber, Layer, LayerMap, Schema, Stream } from "effect"
import path from "path"
import { pathToFileURL } from "url"
import { eq } from "drizzle-orm"
import { Database } from "@opencode/core/database/database"
import { Agent } from "@opencode/core/agent"
import { App } from "@opencode/core/app"
import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
import { LayerNode } from "@opencode/util/effect/layer-node"
import { makeGlobalNode } from "@opencode/util/effect/app-node"
import { Bus } from "@opencode/core/bus"
import { EventTable } from "@opencode/core/event/sql"
import { Location } from "@opencode/schema/location"
import { SessionEvent } from "@opencode/core/session/event"
import { Model } from "@opencode/core/model"
import { Provider } from "@opencode/core/provider"
import { Project } from "@opencode/core/project"
import { ProjectTable } from "@opencode/core/project/sql"
import { AbsolutePath } from "@opencode/core/schema"
import { Session } from "@opencode/core/session"
import { SessionMessage } from "@opencode/core/session/message"
import { SessionProjector } from "@opencode/core/session/projector"
import { SessionExecution } from "@opencode/core/session/execution"
import { SessionInbox } from "@opencode/core/session/inbox"
import {
  BlobRefTable,
  BlobTable,
  SessionInboxTable,
  SessionMessageTable,
  SessionTable,
} from "@opencode/core/session/sql"
import { SessionStore } from "@opencode/core/session/store"
import { LocationServiceMap } from "@opencode/core/location-service-map"
import type { LocationServices } from "@opencode/core/location-services"
import { Image } from "@opencode/core/image"
import { Plugin } from "@opencode/core/plugin"
import { PluginHooks } from "@opencode/core/plugin/hooks"
import { Snapshot } from "@opencode/core/snapshot"
import { Skill } from "@opencode/core/skill"
import { tmpdirScoped } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
import { Config } from "@opencode/core/config"
import { Document, Info } from "@opencode/schema/config"
import { ConfigExperimental } from "@opencode/schema/config/experimental"
import { SessionStored } from "@opencode/core/session/stored"
import { SessionBlob } from "@opencode/core/session/blob"
import { SessionTransfer } from "@opencode/core/session/transfer"
import { PromptInput } from "@opencode/schema/prompt-input"

const executionCalls: Session.ID[] = []
const interruptCalls: Session.ID[] = []
const interruptContinuations: Array<boolean | undefined> = []
const wakeCalls: Session.ID[] = []
const activeSessions = new Set<Session.ID>()
const execution = Layer.succeed(
  SessionExecution.Service,
  SessionExecution.Service.of({
    active: Effect.sync(() => new Set(activeSessions)),
    isActive: (sessionID) => Effect.sync(() => activeSessions.has(sessionID)),
    resume: (sessionID) =>
      Effect.sync(() => {
        executionCalls.push(sessionID)
      }),
    interrupt: (sessionID, options) =>
      Effect.sync(() => {
        interruptCalls.push(sessionID)
        interruptContinuations.push(options?.resume)
        return activeSessions.delete(sessionID)
      }),
    wake: (sessionID) =>
      Effect.sync(() => {
        wakeCalls.push(sessionID)
      }),
    awaitIdle: () => Effect.void,
  }),
)
const mediaDirectory = "/media"
// A location whose instance fails to boot, the way a bad config or unreadable directory does: as a defect.
const brokenDirectory = "/broken"
const mediaBlob = Config.testLayer([
  new Document({
    type: "document",
    info: new Info({ experimental: new ConfigExperimental.Info({ media_blob: true }) }),
  }),
])
const locations = makeGlobalNode({
  service: LocationServiceMap.Service,
  layer: Layer.effect(
    LocationServiceMap.Service,
    Effect.gen(function* () {
      const bus = yield* Bus.Service
      return yield* LayerMap.make(
        (ref: Location.Ref) =>
          // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
          Layer.mergeAll(
            // Every real instance has its own config, so the caller's never shows through.
            ref.directory === mediaDirectory ? mediaBlob : Config.testLayer([]),
            ref.directory === brokenDirectory ? Layer.effectDiscard(Effect.die("instance boot failed")) : Layer.empty,
            LayerNode.compile(LayerNode.group([PluginHooks.node, Skill.node]), {
              replacements: [Bus.node.replace(Layer.succeed(Bus.Service, bus))],
            }),
            Layer.mock(Image.Service, {
              normalize: (resource, content) =>
                content.mime === "image/heic"
                  ? Effect.fail(new Image.DecodeError({ resource }))
                  : Effect.succeed(
                      content.content.length > 5 * 1024 * 1024 ? { ...content, content: "AA==" } : content,
                    ),
            }),
            Layer.mock(Snapshot.Service, {
              capture: () => Effect.undefined,
              restore: () => Effect.void,
            }),
            Layer.mock(Plugin.Service, { awaitActivation: Effect.void }),
          ).pipe(Layer.fresh) as unknown as Layer.Layer<LocationServices, FileSystem.DirectoryNotFoundError>,
      )
    }),
  ),
  deps: [Bus.node],
})
const harness = (channel?: string) =>
  testEffect(
    AppNodeBuilder.build(
      LayerNode.group([
        Database.node,
        Bus.node,
        SessionProjector.node,
        SessionStore.node,
        Session.node,
        SessionTransfer.node,
      ]),
      [
        Bus.node.replace(Bus.configured({ persist: true })),
        SessionExecution.node.replace(execution),
        LocationServiceMap.node.replace(locations),
        App.node.replace(App.configured(channel ? { channel } : {})),
      ],
    ),
  )
const it = harness()
const devIt = harness("dev")
const sessionID = Session.ID.make("ses_prompt_test")
const pngBase64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="
const messageID = SessionMessage.ID.create()

const setup = Effect.gen(function* () {
  const { db } = yield* Database.Service
  yield* db
    .insert(ProjectTable)
    .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
    .onConflictDoNothing()
    .run()
    .pipe(Effect.orDie)
  yield* db
    .insert(SessionTable)
    .values({
      id: sessionID,
      project_id: Project.ID.global,
      slug: "test",
      directory: "/project",
      title: "test",
      version: "test",
    })
    .onConflictDoNothing()
    .run()
    .pipe(Effect.orDie)
})

const admitted = (id: SessionMessage.ID) => Database.Service.use(({ db }) => SessionInbox.find(db, id))
const admittedCount = Database.Service.use(({ db }) =>
  db
    .select()
    .from(SessionInboxTable)
    .all()
    .pipe(
      Effect.orDie,
      Effect.map((rows) => rows.length),
    ),
)
const eventCount = (type: string) =>
  Database.Service.use(({ db }) =>
    db
      .select()
      .from(EventTable)
      .where(eq(EventTable.type, type))
      .all()
      .pipe(
        Effect.orDie,
        Effect.map((rows) => rows.length),
      ),
  )

const encodeMessage = Schema.encodeSync(SessionMessage.Info)
const assistantRow = (id: SessionMessage.ID, seq: number) => {
  const {
    id: _,
    type,
    ...data
  } = encodeMessage(
    SessionMessage.Assistant.make({
      id,
      type: "assistant",
      agent: Agent.ID.make("build"),
      model: { id: Model.ID.make("model"), providerID: Provider.ID.make("provider") },
      content: [],
      time: { created: DateTime.makeUnsafe(0) },
    }),
  )
  return { id, session_id: sessionID, type, seq, time_created: 0, data }
}

/** A user message with an inline image, then a tool result whose file is `toolUri` (inline by default). */
const mediaMessages = (prefix: string, toolUri = `data:image/png;base64,${pngBase64}`) =>
  [
    {
      id: `msg_${prefix}_user`,
      type: "user",
      text: "look",
      files: [{ data: pngBase64, mime: "image/png", source: { type: "inline" }, name: "shot.png" }],
      time: { created: 1 },
    },
    {
      id: `msg_${prefix}_assistant`,
      type: "assistant",
      agent: "build",
      model: { id: "model", providerID: "provider" },
      content: [
        {
          type: "tool",
          id: "call_shot",
          name: "screenshot",
          executed: true,
          state: {
            status: "completed",
            input: {},
            content: [{ type: "file", uri: toolUri, mime: "image/png" }],
          },
          time: { created: 2, ran: 2, completed: 2 },
        },
      ],
      time: { created: 2, completed: 3 },
    },
  ].map((message) => Schema.decodeUnknownSync(SessionMessage.Info)(message))

/** Forks a parent in `directory` holding one inline user image and returns the child's copied row. */
const forkInlineParent = Effect.fnUntraced(function* (id: string, directory: string) {
  const { db } = yield* Database.Service
  const parentID = Session.ID.make(id)
  yield* db
    .insert(SessionTable)
    .values({ id: parentID, project_id: Project.ID.global, slug: id, directory, title: id, version: "test" })
    .run()
    .pipe(Effect.orDie)
  yield* db
    .insert(SessionMessageTable)
    .values({
      id: SessionMessage.ID.make(`msg_${id}_user`),
      session_id: parentID,
      type: "user",
      seq: 1,
      time_created: 1,
      data: {
        text: "look",
        files: [{ data: pngBase64, mime: "image/png", source: { type: "inline" } }],
        time: { created: 1 },
      },
    })
    .run()
    .pipe(Effect.orDie)
  const child = yield* (yield* Session.Service).fork({ sessionID: parentID })
  const row = yield* db
    .select()
    .from(SessionMessageTable)
    .where(eq(SessionMessageTable.session_id, child.id))
    .get()
    .pipe(Effect.orDie)
  return row!
})

describe("Session.prompt", () => {
  it.effect("exposes the execution registry", () =>
    Effect.gen(function* () {
      const session = yield* Session.Service
      activeSessions.add(sessionID)
      expect(Array.from(yield* session.active)).toEqual([sessionID])
    }).pipe(Effect.ensuring(Effect.sync(() => activeSessions.clear()))),
  )

  it.effect("delegates execution continuation through SessionExecution", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      executionCalls.length = 0
      wakeCalls.length = 0
      yield* session.resume(sessionID)
      expect(executionCalls).toEqual([sessionID])
      expect(wakeCalls).toEqual([])
    }),
  )

  it.effect("delegates process-local interruption through SessionExecution", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      interruptCalls.length = 0
      wakeCalls.length = 0

      expect(yield* session.interrupt(sessionID)).toBeFalse()
      expect(interruptCalls).toEqual([sessionID])
      expect(wakeCalls).toEqual([])
      expect(yield* session.messages({ sessionID })).toEqual([])
    }),
  )

  it.effect("forwards interrupt continuation policy", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      interruptCalls.length = 0
      interruptContinuations.length = 0
      wakeCalls.length = 0

      yield* session.interrupt(sessionID, { resume: true })

      expect(interruptCalls).toEqual([sessionID])
      expect(interruptContinuations).toEqual([true])
      expect(wakeCalls).toEqual([])
    }),
  )

  it.effect("delegates interruption without requiring a recorded Session", () =>
    Effect.gen(function* () {
      const session = yield* Session.Service
      interruptCalls.length = 0

      yield* session.interrupt(Session.ID.make("ses_missing"))
      expect(interruptCalls).toEqual([Session.ID.make("ses_missing")])
    }),
  )

  it.effect("durably admits one user message before transcript promotion", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service

      const message = yield* session.prompt({
        sessionID,
        text: "Fix the failing tests",
        resume: false,
      })

      expect(message.payload.text).toBe("Fix the failing tests")
      expect(yield* session.messages({ sessionID })).toEqual([])
      expect(yield* admitted(message.id)).toMatchObject({
        id: message.id,
        sessionID,
        type: "user",
        payload: { text: "Fix the failing tests" },
        delivery: "steer",
      })
    }),
  )

  it.effect("commits a staged revert before admitting a new prompt", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service

      const boundary = yield* session.prompt({
        sessionID,
        text: "boundary",
        resume: false,
      })
      yield* SessionInbox.promote(db, bus, sessionID, "steer")
      const stale = SessionMessage.ID.make("msg_stale_assistant")
      yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie)
      yield* bus.publish(SessionEvent.RevertEvent.Staged, {
        sessionID,
        revert: { messageID: boundary.id, files: [] },
      })
      expect((yield* session.get(sessionID)).revert?.messageID).toBe(boundary.id)

      yield* session.prompt({ sessionID, text: "after revert", resume: false })

      expect((yield* session.get(sessionID)).revert).toBeUndefined()
      expect(
        (yield* db.select({ id: SessionMessageTable.id }).from(SessionMessageTable).all().pipe(Effect.orDie)).map(
          (row) => row.id,
        ),
      ).not.toContainAnyValues([boundary.id, stale])
      expect(yield* SessionInbox.find(db, boundary.id)).toBeUndefined()
    }),
  )

  it.effect("holds synthetic input behind a staged revert and discards it when committed", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service
      const boundary = yield* session.prompt({
        sessionID,
        text: "boundary",
        resume: false,
      })
      yield* SessionInbox.promote(db, bus, sessionID, "steer")
      yield* bus.publish(SessionEvent.RevertEvent.Staged, {
        sessionID,
        revert: { messageID: boundary.id, files: [] },
      })
      wakeCalls.length = 0

      const completion = yield* session.synthetic({ sessionID, text: "stale completion" })

      expect(wakeCalls).toEqual([])
      expect(yield* SessionInbox.find(db, completion.id)).toMatchObject({ type: "synthetic" })

      yield* session.revert.commit(sessionID)

      expect(yield* SessionInbox.find(db, completion.id)).toBeUndefined()
    }),
  )

  it.effect("resolves attachment MIME before admission", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const uri =
        "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="

      const message = yield* session.prompt({
        sessionID,
        text: "Inspect this image",
        files: [{ uri, name: "image.png", mention: { start: 8, end: 17, text: "[Image 1]" } }],
        resume: false,
      })

      expect(message.payload.files).toEqual([
        {
          data: uri.slice(uri.indexOf(",") + 1),
          mime: "image/png",
          source: { type: "inline" },
          name: "image.png",
          mention: { start: 8, end: 17, text: "[Image 1]" },
        },
      ])
      const stored = yield* admitted(message.id)
      expect(stored?.type).toBe("user")
      if (stored?.type === "user") expect(stored.payload.files).toEqual(message.payload.files)
    }),
  )

  it.effect("materializes selected source file content", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const directory = import.meta.dir
      const source = path.join(directory, "session-prompt.test.ts")
      const sourceUri = pathToFileURL(source)
      sourceUri.searchParams.set("start", "1")
      sourceUri.searchParams.set("end", "1")

      const message = yield* session.prompt({
        sessionID,
        text: "Inspect this",
        files: [{ uri: sourceUri.href, name: "main.ts" }],
        resume: false,
      })

      expect(message.payload.files).toHaveLength(1)
      expect(message.payload.files?.[0]).toMatchObject({
        mime: "text/plain",
        source: { type: "uri", uri: sourceUri.href },
        name: "main.ts",
      })
      expect(
        Buffer.from(message.payload.files?.[0]?.data ?? "", "base64")
          .toString("utf8")
          .replace(/\r$/, ""),
      ).toBe('import { describe, expect } from "bun:test"')
    }),
  )

  it.effect("materializes directories as directory attachments", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const uri = pathToFileURL(import.meta.dir).href

      const message = yield* session.prompt({
        sessionID,
        text: "Inspect this",
        files: [{ uri, name: "source" }],
        resume: false,
      })

      expect(message.payload.files).toHaveLength(1)
      expect(message.payload.files?.[0]).toMatchObject({
        mime: "application/x-directory",
        source: { type: "uri", uri },
        name: "source",
      })
      expect(Buffer.from(message.payload.files?.[0]?.data ?? "", "base64").toString("utf8")).toContain(
        "session-prompt.test.ts",
      )
    }),
  )

  it.effect("materializes local image content before admission", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const directory = yield* tmpdirScoped("opencode-session-prompt-")
      const source = path.join(directory.path, "image.png")
      const bytes = Buffer.from(
        "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=",
        "base64",
      )
      yield* Effect.promise(() => Bun.write(source, bytes))

      const message = yield* session.prompt({
        sessionID,
        text: "Inspect this image",
        files: [{ uri: pathToFileURL(source).href }],
        resume: false,
      })

      expect(message.payload.files).toEqual([
        {
          data: bytes.toString("base64"),
          mime: "image/png",
          source: { type: "uri", uri: pathToFileURL(source).href },
          name: "image.png",
        },
      ])
      const stored = yield* admitted(message.id)
      expect(stored?.type === "user" ? stored.payload.files : undefined).toEqual(message.payload.files)
    }),
  )

  it.live("admits captured image bytes with their source path after the source disappears", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const directory = yield* tmpdirScoped("opencode-session-prompt-")
      const source = path.join(directory.path, "screenshot.png")
      yield* Effect.promise(() => Bun.write(source, Buffer.from(pngBase64, "base64")))
      const bytes = yield* Effect.promise(() => Bun.file(source).arrayBuffer())
      yield* Effect.promise(() => Bun.file(source).delete())
      const file = {
        uri: `data:image/png;base64,${Buffer.from(bytes).toString("base64")}`,
        source: { type: "uri", uri: pathToFileURL(source).href },
        name: "screenshot.png",
        mention: { start: 0, end: 9, text: "[Image 1]" },
      } satisfies PromptInput.FileAttachment

      const message = yield* session.prompt({
        sessionID,
        text: "[Image 1] Inspect this image",
        files: [Schema.decodeUnknownSync(PromptInput.FileAttachment)(file)],
        resume: false,
      })

      expect(message.payload.files).toEqual([
        {
          data: pngBase64,
          mime: "image/png",
          source: file.source,
          name: file.name,
          mention: file.mention,
        },
      ])
      const stored = yield* admitted(message.id)
      expect(stored?.type === "user" ? stored.payload.files : undefined).toEqual(message.payload.files)
      expect(
        Schema.encodeSync(PromptInput.FileAttachment)(Schema.decodeUnknownSync(PromptInput.FileAttachment)(file)),
      ).toEqual(file)
    }),
  )

  it.effect("stores media refs at admission when the session's config enables media_blob", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const mediaSessionID = Session.ID.make("ses_prompt_media")
      yield* db
        .insert(SessionTable)
        .values({
          id: mediaSessionID,
          project_id: Project.ID.global,
          slug: "media",
          directory: mediaDirectory,
          title: "media",
          version: "test",
        })
        .run()
        .pipe(Effect.orDie)
      const bytes = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="

      const message = yield* (yield* Session.Service).prompt({
        sessionID: mediaSessionID,
        text: "Inspect this image",
        files: [{ uri: `data:image/png;base64,${bytes}` }],
        resume: false,
      })

      const row = yield* db
        .select()
        .from(SessionInboxTable)
        .where(eq(SessionInboxTable.id, message.id))
        .get()
        .pipe(Effect.orDie)
      const [hash] = (yield* db.select().from(BlobTable).all().pipe(Effect.orDie)).map((blob) => blob.hash)
      expect(hash).toBeDefined()
      expect(row?.payload).toMatchObject({ files: [{ uri: SessionStored.ref(hash!), mime: "image/png" }] })
      expect(yield* db.select().from(BlobRefTable).all().pipe(Effect.orDie)).toEqual([
        { session_id: mediaSessionID, message_id: message.id, seq: expect.any(Number), hash },
      ])
    }),
  )

  it.effect("forks a session's inline media into blobs when its location enables media_blob", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const child = yield* forkInlineParent("ses_fork_media", mediaDirectory)

      const hash = SessionBlob.hashBytes(Buffer.from(pngBase64, "base64"))
      expect(child.data).toMatchObject({ files: [{ uri: SessionStored.ref(hash), mime: "image/png" }] })
      expect(yield* db.select().from(BlobRefTable).all().pipe(Effect.orDie)).toEqual([
        { session_id: child.session_id, message_id: child.id, seq: 1, hash },
      ])
    }),
  )

  it.effect(
    "forks inline media inline when the parent's location doesn't enable media_blob, whatever the caller's",
    () =>
      Effect.gen(function* () {
        yield* setup
        const { db } = yield* Database.Service
        const child = yield* forkInlineParent("ses_fork_plain", "/project").pipe(Effect.provide(mediaBlob))

        expect(child.data).toMatchObject({ files: [{ data: pngBase64, mime: "image/png" }] })
        expect(yield* db.select().from(BlobTable).all().pipe(Effect.orDie)).toEqual([])
      }),
  )

  it.effect("forks and imports with media inline when the location's instance can't boot", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const session = yield* Session.Service
      const transfer = yield* SessionTransfer.Service
      const child = yield* forkInlineParent("ses_fork_broken", brokenDirectory)
      const imported = yield* transfer.import({
        data: {
          info: { ...(yield* session.get(sessionID)), id: Session.ID.make("ses_import_broken") },
          messages: mediaMessages("ses_import_broken"),
        },
        location: { directory: AbsolutePath.make(brokenDirectory) },
      })

      expect(child.data).toMatchObject({ files: [{ data: pngBase64, mime: "image/png" }] })
      expect(yield* session.messages({ sessionID: imported.id, order: "asc" })).toEqual(
        mediaMessages("ses_import_broken"),
      )
      expect(yield* db.select().from(BlobTable).all().pipe(Effect.orDie)).toEqual([])
    }),
  )

  it.effect("records refs on import only for the blobs it took out of its own inline media", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const session = yield* Session.Service
      const transfer = yield* SessionTransfer.Service
      // A blob no session references, awaiting sweep; the import names it but never held its bytes.
      const orphan = "d".repeat(64)
      yield* db
        .insert(BlobTable)
        .values({ hash: orphan, mime: "image/png", bytes: 3, data: Buffer.from("abc"), time_created: 1 })
        .run()
        .pipe(Effect.orDie)

      const imported = yield* transfer.import({
        data: {
          info: { ...(yield* session.get(sessionID)), id: Session.ID.make("ses_import_orphan") },
          messages: mediaMessages("ses_import_orphan", SessionStored.ref(orphan)),
        },
        location: { directory: AbsolutePath.make(mediaDirectory) },
      })

      const hash = SessionBlob.hashBytes(Buffer.from(pngBase64, "base64"))
      expect(
        yield* db
          .select({ session: BlobRefTable.session_id, hash: BlobRefTable.hash })
          .from(BlobRefTable)
          .all()
          .pipe(Effect.orDie),
      ).toEqual([{ session: imported.id, hash }])
    }),
  )

  devIt.effect("imports media into blobs by default on a dev-channel server with no config", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const session = yield* Session.Service
      const transfer = yield* SessionTransfer.Service
      const imported = yield* transfer.import({
        data: {
          info: { ...(yield* session.get(sessionID)), id: Session.ID.make("ses_import_dev") },
          messages: mediaMessages("ses_import_dev"),
        },
        location: { directory: AbsolutePath.make("/project") },
      })

      const hash = SessionBlob.hashBytes(Buffer.from(pngBase64, "base64"))
      expect(yield* db.select({ hash: BlobTable.hash }).from(BlobTable).all().pipe(Effect.orDie)).toEqual([{ hash }])
      expect(yield* session.messages({ sessionID: imported.id, order: "asc" })).toEqual(mediaMessages("ses_import_dev"))
    }),
  )

  it.effect("imports media into blobs only where the location enables media_blob", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const session = yield* Session.Service
      const transfer = yield* SessionTransfer.Service
      const template = yield* session.get(sessionID)
      const messages = (prefix: string) => mediaMessages(prefix)
      const imported = (id: string, directory: string) =>
        transfer.import({
          data: { info: { ...template, id: Session.ID.make(id) }, messages: messages(id) },
          location: { directory: AbsolutePath.make(directory) },
        })
      const rows = (id: Session.ID) =>
        db
          .select()
          .from(SessionMessageTable)
          .where(eq(SessionMessageTable.session_id, id))
          .orderBy(SessionMessageTable.seq)
          .all()
          .pipe(Effect.orDie)

      const media = yield* imported("ses_import_media", mediaDirectory)
      const plain = yield* imported("ses_import_plain", "/project")

      const hash = SessionBlob.hashBytes(Buffer.from(pngBase64, "base64"))
      const uri = SessionStored.ref(hash)
      const mediaRows = yield* rows(media.id)
      expect(mediaRows.map((row) => row.data)).toMatchObject([
        { files: [{ uri, mime: "image/png", name: "shot.png" }] },
        { content: [{ state: { content: [{ type: "file", uri, mime: "image/png" }] } }] },
      ])
      expect(JSON.stringify(mediaRows)).not.toContain(pngBase64)
      expect(JSON.stringify(yield* rows(plain.id))).toContain(pngBase64)
      expect(yield* db.select({ hash: BlobTable.hash }).from(BlobTable).all().pipe(Effect.orDie)).toEqual([{ hash }])
      expect(
        yield* db
          .select({ session: BlobRefTable.session_id, message: BlobRefTable.message_id, seq: BlobRefTable.seq })
          .from(BlobRefTable)
          .orderBy(BlobRefTable.seq)
          .all()
          .pipe(Effect.orDie),
      ).toEqual(mediaRows.map((row) => ({ session: media.id, message: row.id, seq: row.seq })))
      expect(yield* session.messages({ sessionID: media.id, order: "asc" })).toEqual(messages("ses_import_media"))
      expect(yield* session.messages({ sessionID: plain.id, order: "asc" })).toEqual(messages("ses_import_plain"))
    }),
  )

  it.effect("normalizes large image content before validating persisted Base64", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const pixel = Buffer.from(
        "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=",
        "base64",
      )
      const bytes = Buffer.concat([pixel, Buffer.alloc(4_323_030 - pixel.length)])
      const data = bytes.toString("base64")
      expect(data).toHaveLength(5_764_040)

      const message = yield* session.prompt({
        sessionID,
        text: "Inspect this image",
        files: [{ uri: `data:image/png;base64,${data}` }],
        resume: false,
      })

      expect(message.payload.files).toEqual([
        {
          data: "AA==",
          mime: "image/png",
          source: { type: "inline" },
        },
      ])
    }),
  )

  it.effect("passes images the resizer cannot decode through unchanged", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const data = Buffer.from("000000186674797068656963000000006d69663168656963", "hex").toString("base64")

      const message = yield* session.prompt({
        sessionID,
        text: "Inspect this photo",
        files: [{ uri: `data:image/heic;base64,${data}` }],
        resume: false,
      })

      expect(message.payload.files).toEqual([{ data, mime: "image/heic", source: { type: "inline" } }])
    }),
  )

  it.effect("sniffs data URL content instead of trusting its declared MIME", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const uri = `data:video/mp2t;base64,${Buffer.from("export const value = 1\n").toString("base64")}`

      const message = yield* session.prompt({
        sessionID,
        text: "Inspect this",
        files: [{ uri, name: "main.ts" }],
        resume: false,
      })

      expect(message.payload.files).toEqual([
        {
          data: Buffer.from("export const value = 1\n").toString("base64"),
          mime: "text/plain",
          source: { type: "inline" },
          name: "main.ts",
        },
      ])
    }),
  )

  it.effect("rejects malformed base64 data URLs", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const uri = "data:image/png;base64,not-base64"

      const error = yield* session
        .prompt({
          sessionID,
          text: "Inspect this",
          files: [{ uri, name: "image.png" }],
          resume: false,
        })
        .pipe(Effect.flip)

      expect(error).toMatchObject({
        _tag: "Session.AttachmentError",
        uri: "image.png",
        message: "Invalid attachment data URL: image.png",
      })
    }),
  )

  it.effect("rejects oversized inline attachments without echoing their bytes", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const uri = `data:application/octet-stream;base64,${Buffer.alloc(20 * 1024 * 1024 + 1).toString("base64")}`

      const error = yield* session
        .prompt({
          sessionID,
          text: "Inspect this",
          files: [{ uri }],
          resume: false,
        })
        .pipe(Effect.flip)

      expect(error).toMatchObject({
        _tag: "Session.AttachmentError",
        uri: "inline attachment",
        message: "Attachment exceeds the 20971520 byte limit: inline attachment",
      })
    }),
  )

  it.effect("streams durable Session events after an aggregate sequence", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service
      const publicEvents = (input: { sessionID: Session.ID; after?: number }) =>
        session
          .log({ ...input, follow: true })
          .pipe(Stream.filter((item): item is SessionEvent.DurableEvent => !Bus.isSynced(item)))
      const fiber = yield* publicEvents({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
      yield* Effect.yieldNow

      yield* session.prompt({ sessionID, text: "First", resume: false })
      yield* session.prompt({ sessionID, text: "Second", resume: false })
      yield* SessionInbox.promote(db, bus, sessionID, "steer")
      const streamed = Array.from(yield* Fiber.join(fiber))

      expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([
        [0, "session.inbox.enqueued"],
        [1, "session.inbox.enqueued"],
        [2, "session.inbox.delivered"],
        [3, "session.inbox.delivered"],
      ])
      expect(
        Array.from(
          yield* publicEvents({ sessionID, after: streamed[0].durable?.seq }).pipe(Stream.take(1), Stream.runCollect),
        ).map((event): [number | undefined, string] => [event.durable?.seq, event.type]),
      ).toEqual([[1, "session.inbox.enqueued"]])
    }),
  )

  it.effect("resumes through a recorded message without appending another prompt", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const message = yield* session.prompt({
        sessionID,
        text: "Fix the failing tests",
        resume: false,
      })

      executionCalls.length = 0
      wakeCalls.length = 0
      yield* session.resume(sessionID)

      expect(yield* session.messages({ sessionID })).toEqual([])
      expect((yield* session.inbox(sessionID)).map((item) => item.id)).toEqual([message.id])
      expect(executionCalls).toEqual([sessionID])
      expect(wakeCalls).toEqual([])
    }),
  )

  it.effect("records distinct messages when the ID is omitted", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const input = { sessionID, text: "Fix the failing tests", resume: false }

      const first = yield* session.prompt(input)
      const second = yield* session.prompt(input)

      expect(second.id).not.toBe(first.id)
      expect(yield* session.messages({ sessionID })).toEqual([])
      expect(yield* admittedCount).toBe(2)
    }),
  )

  it.effect("returns the original recorded message when the ID is retried", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const input = {
        sessionID,
        id: messageID,
        text: "Fix the failing tests",
        resume: false,
      }

      const first = yield* session.prompt(input)
      const retried = yield* session.prompt(input)

      expect(retried).toEqual(first)
      expect(yield* session.messages({ sessionID })).toEqual([])
      expect(yield* admittedCount).toBe(1)
    }),
  )

  it.effect("reconciles an exact retry from the promoted message without admission history", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service
      const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
      const first = yield* session.prompt(input)
      yield* SessionInbox.promote(db, bus, sessionID, "steer")
      yield* db.delete(EventTable).where(eq(EventTable.aggregate_id, sessionID)).run().pipe(Effect.orDie)

      const retried = yield* session.prompt(input)

      expect(retried).toMatchObject({ id: first.id, type: "user", payload: { text: first.payload.text } })
      expect(yield* session.messages({ sessionID })).toMatchObject([
        { id: messageID, type: "user", text: "Fix the failing tests" },
      ])
    }),
  )

  it.effect("ignores delivery when retrying a promoted message", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service
      const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
      yield* session.prompt(input)
      yield* SessionInbox.promote(db, bus, sessionID, "steer")

      const retried = yield* session.prompt({ ...input, delivery: "queue" })

      expect(retried).toMatchObject({ id: messageID, type: "user", payload: { text: input.text } })
      expect(yield* admitted(messageID)).toBeUndefined()
    }),
  )

  it.effect("wakes execution when an exact prompt retry recovers a committed message", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const input = {
        sessionID,
        id: messageID,
        text: "Recover committed prompt",
        resume: false,
      }
      const first = yield* session.prompt(input)
      wakeCalls.length = 0

      const retried = yield* session.prompt({ ...input, resume: true })

      expect(retried).toEqual(first)
      expect(wakeCalls).toEqual([sessionID])
    }),
  )

  it.effect("keeps the first admission when one ID is reused with a different prompt", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service

      const first = yield* session.prompt({
        sessionID,
        id: messageID,
        text: "Fix the failing tests",
      })
      const retried = yield* session.prompt({
        sessionID,
        id: messageID,
        text: "Delete the failing tests",
        resume: false,
      })

      expect(retried).toEqual(first)
      expect(retried.payload.text).toBe("Fix the failing tests")
      expect(yield* session.messages({ sessionID })).toHaveLength(0)
      expect(yield* admittedCount).toBe(1)
    }),
  )

  it.effect("keeps the first admission's delivery mode when one ID is reused with another", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service

      const first = yield* session.prompt({
        id: messageID,
        sessionID,
        text: "Fix the failing tests",
        resume: false,
      })
      const retried = yield* session.prompt({
        id: messageID,
        sessionID,
        text: "Fix the failing tests",
        delivery: "queue",
        resume: false,
      })

      expect(retried).toEqual(first)
      expect(retried.delivery).toBe("steer")
      expect(yield* admittedCount).toBe(1)
    }),
  )

  it.effect("returns one recorded message to concurrent exact retries", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const input = {
        sessionID,
        id: messageID,
        text: "Fix the failing tests",
        resume: false,
      }

      const messages = yield* Effect.all([session.prompt(input), session.prompt(input)], { concurrency: "unbounded" })

      expect(messages[1]).toEqual(messages[0])
      expect(yield* session.messages({ sessionID })).toEqual([])
      expect(yield* admittedCount).toBe(1)
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxEnqueued.type, 1))).toBe(1)
    }),
  )

  it.effect("promotes one message once under concurrent promotion attempts", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      yield* session.prompt({
        id: messageID,
        sessionID,
        text: "Promote once",
        resume: false,
      })

      yield* Effect.all(
        [SessionInbox.promote(db, bus, sessionID, "steer"), SessionInbox.promote(db, bus, sessionID, "steer")],
        { concurrency: "unbounded" },
      )

      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxDelivered.type, 1))).toBe(1)
      expect(yield* admitted(messageID)).toBeUndefined()
      expect(yield* session.messages({ sessionID })).toMatchObject([
        { id: messageID, type: "user", text: "Promote once" },
      ])
    }),
  )

  it.effect("reprojects pending inbox input without scheduling execution", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      wakeCalls.length = 0
      yield* session.prompt({
        id: messageID,
        sessionID,
        text: "Replay pending",
        resume: false,
      })
      const syntheticID = SessionMessage.ID.create()
      yield* session.synthetic({ id: syntheticID, sessionID, text: "Replay synthetic", resume: false })
      const recorded = yield* db
        .select()
        .from(EventTable)
        .where(eq(EventTable.aggregate_id, sessionID))
        .all()
        .pipe(Effect.orDie)

      yield* bus.remove(sessionID)
      yield* db.delete(SessionInboxTable).where(eq(SessionInboxTable.session_id, sessionID)).run().pipe(Effect.orDie)
      yield* db
        .delete(SessionMessageTable)
        .where(eq(SessionMessageTable.session_id, sessionID))
        .run()
        .pipe(Effect.orDie)
      yield* Effect.forEach(
        recorded.map((event) => ({
          id: event.id,
          created: event.created,
          aggregateID: event.aggregate_id,
          seq: event.seq,
          type: event.type,
          data: event.data,
        })),
        (event) => bus.replay(event),
        { discard: true },
      )

      expect(yield* admitted(messageID)).toMatchObject({
        id: messageID,
        type: "user",
        payload: { text: "Replay pending" },
      })
      expect(yield* admitted(syntheticID)).toMatchObject({
        id: syntheticID,
        type: "synthetic",
        payload: { text: "Replay synthetic" },
      })
      expect(yield* session.messages({ sessionID })).toEqual([])
      expect(wakeCalls).toEqual([])
    }),
  )

  it.effect("rejects reuse of one globally unique message ID across sessions", () =>
    Effect.gen(function* () {
      yield* setup
      const { db } = yield* Database.Service
      const session = yield* Session.Service
      const other = Session.ID.make("ses_prompt_other")
      yield* db
        .insert(SessionTable)
        .values({
          id: other,
          project_id: Project.ID.global,
          slug: "other",
          directory: "/project",
          title: "other",
          version: "test",
        })
        .onConflictDoNothing()
        .run()
        .pipe(Effect.orDie)
      yield* session.prompt({ id: messageID, sessionID, text: "Fix the failing tests", resume: false })
      const failure = yield* session
        .prompt({ id: messageID, sessionID: other, text: "Fix the failing tests", resume: false })
        .pipe(Effect.flip)

      expect(failure).toMatchObject({ _tag: "Session.PromptConflictError", sessionID: other, messageID })
    }),
  )

  it.effect("rejects a prompt ID already used by visible Session history", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const { db } = yield* Database.Service
      const {
        id: _,
        type,
        ...data
      } = encodeMessage({
        id: messageID,
        type: "synthetic",
        text: "Existing history",
        time: { created: DateTime.makeUnsafe(0) },
      })
      yield* db
        .insert(SessionMessageTable)
        .values({ id: messageID, session_id: sessionID, type, seq: 0, time_created: 0, data })
        .run()
        .pipe(Effect.orDie)

      const failure = yield* session
        .prompt({
          id: messageID,
          sessionID,
          text: "Conflicting prompt",
          resume: false,
        })
        .pipe(Effect.flip)

      expect(failure).toMatchObject({ _tag: "Session.PromptConflictError", sessionID, messageID })
      expect(yield* admitted(messageID)).toBeUndefined()
    }),
  )

  it.effect("starts execution by default after recording the prompt", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      executionCalls.length = 0
      wakeCalls.length = 0

      yield* session.prompt({ sessionID, text: "Run by default" })

      expect(executionCalls).toEqual([])
      expect(wakeCalls).toEqual([sessionID])
    }),
  )

  it.effect("starts execution when resume is explicitly true", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      executionCalls.length = 0
      wakeCalls.length = 0

      yield* session.prompt({
        sessionID,
        text: "Run explicitly",
        resume: true,
      })

      expect(executionCalls).toEqual([])
      expect(wakeCalls).toEqual([sessionID])
    }),
  )

  it.effect("only records the prompt when resume is false", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      executionCalls.length = 0
      wakeCalls.length = 0

      yield* session.prompt({ sessionID, text: "Do not run", resume: false })

      expect(executionCalls).toEqual([])
      expect(wakeCalls).toEqual([])
    }),
  )

  it.effect("keeps the first admission's metadata when one ID is reused with other metadata", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const input = {
        id: messageID,
        sessionID,
        text: "Deploy",
        metadata: { source: "api" },
        resume: false,
      }

      const first = yield* session.prompt(input)
      const retried = yield* session.prompt(input)
      const differing = yield* session.prompt({ ...input, metadata: { source: "plugin" } })

      expect(retried).toEqual(first)
      expect(differing).toEqual(first)
      expect(first.payload.metadata).toEqual({ source: "api" })
      expect(yield* admittedCount).toBe(1)
    }),
  )

  it.effect("durably admits synthetic input before transcript promotion", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service

      const input = yield* session.synthetic({
        id: messageID,
        sessionID,
        text: "Background work completed",
        description: "shell completion",
        metadata: { job: "shell" },
        resume: false,
      })

      expect(yield* session.messages({ sessionID })).toEqual([])
      expect(yield* admitted(input.id)).toMatchObject({
        type: "synthetic",
        sessionID,
        delivery: "steer",
        payload: {
          text: "Background work completed",
          description: "shell completion",
          metadata: { job: "shell" },
        },
      })

      yield* SessionInbox.promote(db, bus, sessionID, "steer")

      expect(yield* session.messages({ sessionID })).toMatchObject([
        {
          id: messageID,
          type: "synthetic",
          text: "Background work completed",
          description: "shell completion",
          metadata: { job: "shell" },
        },
      ])
    }),
  )

  it.effect("reconciles synthetic retries from the promoted message regardless of payload", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const database = yield* Database.Service
      const input = { id: messageID, sessionID, text: "Completed", resume: false }

      const entries = yield* Effect.all([session.synthetic(input), session.synthetic(input)], {
        concurrency: "unbounded",
      })
      yield* SessionInbox.promote(database.db, bus, sessionID, "steer")
      const promotedRetry = yield* session.synthetic(input)
      const differing = yield* session.synthetic({ ...input, text: "Different completion" })

      expect(entries[1]).toEqual(entries[0])
      expect(promotedRetry).toMatchObject({ id: messageID, type: "synthetic", payload: { text: "Completed" } })
      expect(differing).toMatchObject({ id: messageID, type: "synthetic", payload: { text: "Completed" } })
      expect(yield* admittedCount).toBe(0)
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxEnqueued.type, 1))).toBe(1)
    }),
  )

  it.effect("keeps queued input pending until the idle boundary", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service

      const input = yield* session.synthetic({
        sessionID,
        text: "Queued completion",
        delivery: "queue",
        resume: false,
      })

      expect(input.delivery).toBe("queue")
      expect(yield* SessionInbox.has(db, sessionID, "input")).toBe(true)
      expect(yield* SessionInbox.promote(db, bus, sessionID, "steer")).toBe(0)
      expect(yield* session.messages({ sessionID })).toEqual([])
      expect(yield* SessionInbox.promote(db, bus, sessionID, "input")).toBe(1)
      expect(yield* SessionInbox.has(db, sessionID, "input")).toBe(false)
      expect(yield* session.messages({ sessionID })).toMatchObject([
        { id: input.id, type: "synthetic", text: "Queued completion" },
      ])
    }),
  )

  it.effect("promotes prompt and synthetic steers in admission order", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service

      yield* session.prompt({
        sessionID,
        text: "First prompt",
        resume: false,
      })
      yield* session.synthetic({ sessionID, text: "Background completion", resume: false })
      yield* session.prompt({
        sessionID,
        text: "Second prompt",
        resume: false,
      })

      yield* SessionInbox.promote(db, bus, sessionID, "steer")

      expect(
        (yield* session.messages({ sessionID, order: "asc" })).map((message) =>
          message.type === "user" || message.type === "synthetic" ? message.text : message.type,
        ),
      ).toEqual(["First prompt", "Background completion", "Second prompt"])
    }),
  )
})

describe("Session.inbox", () => {
  it.effect("fails for an unknown session", () =>
    Effect.gen(function* () {
      const session = yield* Session.Service
      expect(yield* session.inbox(Session.ID.make("ses_missing")).pipe(Effect.flip)).toMatchObject({
        _tag: "Session.NotFoundError",
      })
    }),
  )

  it.effect("lists admitted work in admission order until promotion", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const bus = yield* Bus.Service
      const { db } = yield* Database.Service

      const first = yield* session.prompt({ sessionID, text: "First steer", resume: false })
      const queued = yield* session.synthetic({
        sessionID,
        text: "Queued completion",
        delivery: "queue",
        resume: false,
      })
      const second = yield* session.prompt({ sessionID, text: "Second steer", resume: false })

      expect(yield* session.inbox(sessionID)).toMatchObject([
        { id: first.id, type: "user", delivery: "steer" },
        { id: queued.id, type: "synthetic", delivery: "queue" },
        { id: second.id, type: "user", delivery: "steer" },
      ])

      expect(yield* SessionInbox.promote(db, bus, sessionID, "input")).toBe(2)
      expect(yield* session.inbox(sessionID)).toMatchObject([{ id: queued.id, type: "synthetic" }])

      expect(yield* SessionInbox.promote(db, bus, sessionID, "input")).toBe(1)
      expect(yield* session.inbox(sessionID)).toEqual([])
    }),
  )

  it.effect("lists an unhandled compaction until it is cancelled", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const { db } = yield* Database.Service

      const barrier = yield* session.compact({ sessionID })
      expect(yield* SessionInbox.has(db, sessionID, "input")).toBe(true)
      expect(yield* session.inbox(sessionID)).toMatchObject([{ id: barrier.id, type: "compaction" }])

      yield* session.cancelInbox({ sessionID, inboxID: barrier.id })
      expect(yield* SessionInbox.has(db, sessionID, "input")).toBe(false)
      expect(yield* session.inbox(sessionID)).toEqual([])
    }),
  )

  it.effect("cancels pending input and allows its ID to be admitted again", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const inputID = SessionMessage.ID.make("msg_cancelled_queue")
      yield* session.prompt({
        id: inputID,
        sessionID,
        text: "Queue this",
        delivery: "queue",
        resume: false,
      })

      yield* session.cancelInbox({ sessionID, inboxID: inputID })

      expect(yield* session.inbox(sessionID)).toEqual([])
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxCancelled.type, 1))).toBe(1)
      expect(yield* session.cancelInbox({ sessionID, inboxID: inputID }).pipe(Effect.flip)).toMatchObject({
        _tag: "Session.InboxConflictError",
        sessionID,
        inboxID: inputID,
      })
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxCancelled.type, 1))).toBe(1)

      const retried = yield* session.prompt({
        id: inputID,
        sessionID,
        text: "Queue this",
        delivery: "queue",
        resume: false,
      })
      expect(retried).toMatchObject({ id: inputID, delivery: "queue" })
    }),
  )

  it.effect("moves pending input between steer and queue delivery", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* Session.Service
      const queued = yield* session.synthetic({
        sessionID,
        text: "Steer this",
        delivery: "queue",
        resume: false,
      })
      const alreadySteered = yield* session.prompt({ sessionID, text: "Already steer", resume: false })
      wakeCalls.length = 0

      yield* session.steerInbox({ sessionID, inboxID: queued.id })

      expect(yield* session.inbox(sessionID)).toMatchObject([
        { id: queued.id, delivery: "steer" },
        { id: alreadySteered.id, delivery: "steer" },
      ])
      expect(wakeCalls).toEqual([sessionID])
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxDeliveryChanged.type, 1))).toBe(1)

      wakeCalls.length = 0
      yield* session.queueInbox({ sessionID, inboxID: queued.id })
      expect(yield* session.inbox(sessionID)).toMatchObject([
        { id: queued.id, delivery: "queue" },
        { id: alreadySteered.id, delivery: "steer" },
      ])
      expect(wakeCalls).toEqual([])
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxDeliveryChanged.type, 1))).toBe(2)

      expect(yield* session.steerInbox({ sessionID, inboxID: alreadySteered.id }).pipe(Effect.flip)).toMatchObject({
        _tag: "Session.InboxConflictError",
        sessionID,
        inboxID: alreadySteered.id,
      })
      yield* session.cancelInbox({ sessionID, inboxID: alreadySteered.id })
      expect(wakeCalls).toEqual([])
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxDeliveryChanged.type, 1))).toBe(2)
      expect(yield* eventCount(Bus.versionedType(SessionEvent.InboxCancelled.type, 1))).toBe(1)
    }),
  )
})