import Foundation
import Testing

@testable import NucleicCore

/// `NashEventLog` is the Nash Viewer's source: it projects every nash emission kind onto one flat
/// row shape and keeps a per-session ring, so a chatty session can't evict another's history.
@Suite struct NashEventLogTests {
    private let session = SessionID(rawValue: "s1")
    private let at = Date(timeIntervalSince1970: 1_700_000_000)

    private func project(
        _ kind: MCPApprovalServer.ShellReportCall.Kind, shellID: String = "sh-1"
    ) -> NashEvent {
        NashEventLog.project(seq: 1, at: at, sessionID: session, shellID: shellID, kind: kind)
    }

    @Test func projectsExecWithCwdExitAndDuration() {
        let ok = project(
            .exec(argv: ["npm", "test"], cwd: "/w", exitCode: 0, durationMs: 1500, pipeline: nil))
        #expect(ok.kind == .exec)
        #expect(ok.summary == "npm test")
        #expect(ok.detail == "in /w · 1.5s")  // exit omitted when it succeeded
        #expect(ok.preview == nil)
        #expect(ok.details == .command(argv: ["npm", "test"], cwd: "/w", exitCode: 0, durationMs: 1500, pipeline: nil))
        #expect(ok.pipelineKey == nil)

        let failed = project(
            .exec(argv: ["ls", "nope"], cwd: "/w", exitCode: 2, durationMs: 12, pipeline: nil))
        #expect(failed.detail == "in /w · exit 2 · 12ms")
    }

    /// A stage's exec event keeps the slot that joins it to the pipe links either side, keyed by the
    /// emitting shell — nash's pipeline counter restarts at 1 in every process.
    @Test func projectsExecPipelineSlot() {
        let slot = NashPipelineSlot(id: 4, index: 1, length: 3)
        let staged = project(
            .exec(argv: ["grep", "x"], cwd: "/w", exitCode: 0, durationMs: 1, pipeline: slot),
            shellID: "sh-9")
        #expect(staged.details == .command(argv: ["grep", "x"], cwd: "/w", exitCode: 0, durationMs: 1, pipeline: slot))
        #expect(staged.pipelineKey == NashPipelineKey(shellID: "sh-9", pipelineID: 4))
    }

    @Test func projectsDataFlowKindsWithDecodedPreview() {
        let payload = Data("hello world".utf8).base64EncodedString()
        let redirect = project(
            .redirect(
                op: ">", fd: 1, target: "out.txt", bytes: 2048, truncated: true,
                hash: "abcdef0123456789", previewB64: payload))
        #expect(redirect.summary == "1> out.txt")
        #expect(redirect.detail == "2.0 KB")
        #expect(redirect.details == .redirect(op: ">", fd: 1, target: "out.txt"))
        #expect(redirect.hash == "abcdef0123456789")
        #expect(redirect.preview == "hello world")
        #expect(redirect.bytes == 2048)
        #expect(redirect.truncated)

        // nash sends stderr as fd 2 *and* op "2>"; the summary must not print the fd twice.
        let stderr = project(
            .redirect(
                op: "2>", fd: 2, target: "/tmp/err.log", bytes: 0, truncated: false,
                hash: "", previewB64: ""))
        #expect(stderr.summary == "2> /tmp/err.log")

        let pipe = project(
            .pipe(
                pipelineID: 7, fromIndex: 0, toIndex: 1, fromText: "curl -sL https://x.sh",
                toText: "sh", bytes: 12, truncated: false, hash: "", previewB64: payload))
        // The link names its own ends, so the flat line reads as the pipeline you wrote.
        #expect(pipe.summary == "curl -sL https://x.sh | sh → 12 B")
        #expect(pipe.detail == "pipeline 7")
        #expect(
            pipe.details
                == .pipe(
                    pipelineID: 7, fromIndex: 0, toIndex: 1, fromText: "curl -sL https://x.sh",
                    toText: "sh"))
        #expect(pipe.hash == nil)  // nash sent none; not the string "no hash"
        #expect(pipe.pipelineKey == NashPipelineKey(shellID: "sh-1", pipelineID: 7))

        // A nash that predates stage text falls back to 1-based stage numbers, matching the row.
        let unnamed = project(
            .pipe(
                pipelineID: 7, fromIndex: 0, toIndex: 1, fromText: "", toText: "", bytes: 12,
                truncated: false, hash: "", previewB64: payload))
        #expect(unnamed.summary == "stage 1 | stage 2 → 12 B")
    }

    /// A binary payload (or `/dev/null`'s empty preview) must not become mojibake — the row falls
    /// back to metadata only, and the viewer simply shows no disclosure.
    @Test func dropsUndecodablePreviews() {
        let binary = Data([0xff, 0xfe, 0xfd]).base64EncodedString()
        #expect(project(.cmdsub(bytes: 3, truncated: false, hash: "h", previewB64: binary)).preview == nil)
        #expect(project(.cmdsub(bytes: 0, truncated: false, hash: "h", previewB64: "")).preview == nil)
    }

    @Test func projectsMetadataKinds() {
        #expect(project(.cd(from: "/a", to: "/b")).summary == "cd /b")
        #expect(project(.export(op: "export", names: ["PATH", "HOME"])).summary == "export PATH HOME")
        #expect(project(.fallback(reason: "parse", input: "for x in")).kind == .fallback)
        #expect(project(.dropped(count: 1)).summary == "dropped 1 event under back-pressure")
        #expect(project(.dropped(count: 4)).summary == "dropped 4 events under back-pressure")
    }

    @MainActor
    @Test func ringIsPerSessionAndTotalsSurviveTrim() {
        let log = NashEventLog(capacity: 2)
        let other = SessionID(rawValue: "s2")
        for i in 0..<5 {
            log.record(
                sessionID: session,
                call: .init(kind: .cd(from: "/", to: "/\(i)"), sessionID: session.rawValue))
        }
        log.record(sessionID: other, call: .init(kind: .dropped(count: 3), sessionID: other.rawValue))

        // s1 trimmed to the newest 2; s2 untouched by s1's churn.
        #expect(log.events(for: session).map(\.summary) == ["cd /3", "cd /4"])
        #expect(log.events(for: other).count == 1)
        // Lifetime totals keep counting past the trim.
        #expect(log.total(for: session) == 5)
        #expect(log.total(for: session, kind: .cd) == 5)
        #expect(log.total(for: other, kind: .dropped) == 1)

        log.clear(sessionID: session)
        #expect(log.events(for: session).isEmpty)
        #expect(log.total(for: session) == 5)  // clear drops rows, not history

        log.forget(sessionID: session)
        #expect(log.total(for: session) == 0)
    }

    @MainActor
    @Test func retainsAndIndexesEmitterEnvironment() {
        let log = NashEventLog()
        let vm = NashEnvironment.linuxVM(name: "nucleic-lin-s1")
        let container = NashEnvironment.container(name: "scratch-build")
        log.record(
            sessionID: session,
            call: .init(
                kind: .exec(
                    argv: ["swift", "test"], cwd: "/w", exitCode: 0, durationMs: 1, pipeline: nil),
                sessionID: session.rawValue, environment: vm))
        log.record(
            sessionID: session,
            call: .init(
                kind: .cd(from: "/w", to: "/tmp"), sessionID: session.rawValue),
            environment: container)

        #expect(log.events(for: session).map(\.environment) == [vm, container])
        #expect(log.environments(for: session) == [vm, container])
        log.clear(sessionID: session)
        #expect(log.environments(for: session) == [vm, container])
        log.forget(sessionID: session)
        #expect(log.environments(for: session).isEmpty)
    }

    /// The join a pipe row is built on: a link names its producer and consumer by looking up the
    /// exec events that carried the same pipeline slot.
    @MainActor
    @Test func indexesPipelineStagesFromExecEvents() {
        let log = NashEventLog()
        func stage(_ argv: [String], index: Int, exitCode: Int = 0, shell: String = "sh-1", id: Int = 3) {
            log.record(
                sessionID: session,
                call: .init(
                    kind: .exec(
                        argv: argv, cwd: "/w", exitCode: exitCode, durationMs: 1,
                        pipeline: NashPipelineSlot(id: id, index: index, length: 2)),
                    sessionID: session.rawValue, shellID: shell))
        }
        stage(["curl", "-sL", "https://x.sh"], index: 0)
        stage(["sh"], index: 1, exitCode: 1)

        let stages = log.stages(for: session, pipeline: NashPipelineKey(shellID: "sh-1", pipelineID: 3))
        #expect(stages[0]?.command == "curl -sL https://x.sh")
        #expect(stages[1] == NashPipelineStage(command: "sh", exitCode: 1, length: 2))

        // The counter restarts in every nash process, so another shell's pipeline 3 is a different
        // pipeline — it must not borrow these stage names.
        #expect(log.stages(for: session, pipeline: NashPipelineKey(shellID: "sh-2", pipelineID: 3)).isEmpty)

        // A stage that is a compound command reports one exec per command inside it; the first is
        // the one that opened the stage and the one the link is named for.
        stage(["echo", "second"], index: 0)
        #expect(stages[0]?.command == log.stages(for: session, pipeline: .init(shellID: "sh-1", pipelineID: 3))[0]?.command)

        // Stage names outlive the rows: a pipe row's whole value is naming its ends, and the exec
        // rows that name them trim on the same schedule as the pipe row itself.
        log.clear(sessionID: session)
        #expect(log.stages(for: session, pipeline: .init(shellID: "sh-1", pipelineID: 3))[0] != nil)
        log.forget(sessionID: session)
        #expect(log.stages(for: session, pipeline: .init(shellID: "sh-1", pipelineID: 3)).isEmpty)
    }

    /// A command still running is projected as an `exec` row with nothing it can't know yet: no
    /// duration, and `isRunning` for the spinner the panel draws instead of a metric.
    @Test func projectsExecStartAsARunningCommand() {
        let running = project(.execStart(argv: ["make", "-j8"], cwd: "/w", elapsedMs: 450, pipeline: nil))
        #expect(running.kind == .exec)
        #expect(running.isRunning)
        #expect(running.summary == "make -j8")
        #expect(running.detail == "in /w · running")
        #expect(running.details == .command(argv: ["make", "-j8"], cwd: "/w", exitCode: 0, durationMs: nil, pipeline: nil))
    }

    /// The lifecycle the panel is built on: an announced command shows as one row that *becomes*
    /// the finished row — same id, same place, same clock — rather than appearing twice.
    @MainActor
    @Test func completionRewritesTheRunningRowInPlace() throws {
        let log = NashEventLog()
        func report(
            _ kind: MCPApprovalServer.ShellReportCall.Kind, token: Int?, shell: String = "sh-1",
            at: Date = self.at
        ) {
            log.record(
                sessionID: session,
                call: .init(
                    kind: kind, sessionID: session.rawValue, shellID: shell, execToken: token),
                at: at)
        }
        report(.execStart(argv: ["make"], cwd: "/w", elapsedMs: 450, pipeline: nil), token: 7)
        let started = try #require(log.events(for: session).last)
        #expect(started.isRunning)
        // The row counts elapsed time from its own clock, so that clock is the moment the command
        // started — not the moment we heard about it, 450ms of announce delay and batching later.
        #expect(started.at == at - 0.45)
        // While it runs it is a promise of a row, not an emission: the count is the completion's.
        #expect(log.total(for: session, kind: .exec) == 0)

        report(
            .exec(argv: ["make"], cwd: "/w", exitCode: 2, durationMs: 9000, pipeline: nil),
            token: 7, at: at + 9)
        let rows = log.events(for: session)
        #expect(rows.count == 1)
        #expect(rows[0].id == started.id)  // the row the panel is already showing, updated in place
        #expect(rows[0].at == started.at)  // stamped when it started, which is where it sits
        #expect(!rows[0].isRunning)
        #expect(rows[0].detail == "in /w · exit 2 · 9.0s")
        #expect(log.total(for: session, kind: .exec) == 1)
    }

    /// Every way the join can fail has to leave an ordinary two-row-free feed: the completion is
    /// never lost, it just lands as its own row.
    @MainActor
    @Test func completionAppendsWhenNoRunningRowIsWaiting() {
        let log = NashEventLog()
        func report(_ kind: MCPApprovalServer.ShellReportCall.Kind, token: Int?, shell: String = "sh-1") {
            log.record(
                sessionID: session,
                call: .init(
                    kind: kind, sessionID: session.rawValue, shellID: shell, execToken: token))
        }
        report(.execStart(argv: ["make"], cwd: "/w", elapsedMs: 450, pipeline: nil), token: 7)
        // Another nash process counts its own tokens from 1, so token 7 there is a different
        // command — it must not retire this one's row.
        report(.exec(argv: ["ls"], cwd: "/w", exitCode: 0, durationMs: 1, pipeline: nil), token: 7, shell: "sh-2")
        // And a nash that predates the announcement sends no token at all.
        report(.exec(argv: ["pwd"], cwd: "/w", exitCode: 0, durationMs: 1, pipeline: nil), token: nil)
        #expect(log.events(for: session).map(\.summary) == ["make", "ls", "pwd"])
        #expect(log.events(for: session).map(\.isRunning) == [true, false, false])

        // Clearing mid-command drops the row the completion would have landed on; it appends.
        log.clear(sessionID: session)
        report(.exec(argv: ["make"], cwd: "/w", exitCode: 0, durationMs: 5, pipeline: nil), token: 7)
        #expect(log.events(for: session).map(\.summary) == ["make"])
        #expect(log.events(for: session).map(\.isRunning) == [false])
    }

    /// A running row knows neither its exit code nor which stage it opened, so it must not seed the
    /// stage index the pipe rows are named from — the completion does that.
    @MainActor
    @Test func runningRowDoesNotClaimItsPipelineStage() {
        let log = NashEventLog()
        let slot = NashPipelineSlot(id: 3, index: 0, length: 2)
        let key = NashPipelineKey(shellID: "sh-1", pipelineID: 3)
        log.record(
            sessionID: session,
            call: .init(
                kind: .execStart(argv: ["curl", "-s", "u"], cwd: "/w", elapsedMs: 450, pipeline: slot),
                sessionID: session.rawValue, shellID: "sh-1", execToken: 1))
        #expect(log.stages(for: session, pipeline: key).isEmpty)

        log.record(
            sessionID: session,
            call: .init(
                kind: .exec(argv: ["curl", "-s", "u"], cwd: "/w", exitCode: 6, durationMs: 30, pipeline: slot),
                sessionID: session.rawValue, shellID: "sh-1", execToken: 1))
        #expect(log.stages(for: session, pipeline: key)[0] == NashPipelineStage(command: "curl -s u", exitCode: 6, length: 2))
    }

    /// The stage index is unbounded input (one entry per pipeline the agent ever runs), so it evicts
    /// whole pipelines oldest-first rather than growing for the life of the session.
    @MainActor
    @Test func pipelineIndexEvictsOldestPipelines() {
        let log = NashEventLog(pipelineCapacity: 2)
        for id in 1...3 {
            log.record(
                sessionID: session,
                call: .init(
                    kind: .exec(
                        argv: ["stage\(id)"], cwd: "/w", exitCode: 0, durationMs: 1,
                        pipeline: NashPipelineSlot(id: id, index: 0, length: 2)),
                    sessionID: session.rawValue, shellID: "sh-1"))
        }
        #expect(log.stages(for: session, pipeline: .init(shellID: "sh-1", pipelineID: 1)).isEmpty)
        #expect(log.stages(for: session, pipeline: .init(shellID: "sh-1", pipelineID: 3))[0] != nil)
    }

    /// A whole nash batch lands in one pass (docs/NASH_STREAM_PERF_PLAN.md §P1) and has to be
    /// indistinguishable from the same events recorded one at a time: order preserved, the trim
    /// applied once at the end, and a completion still folding into the announcement it retires —
    /// including when both arrive in the *same* batch, which is what a 500 ms flush makes common.
    @MainActor
    @Test func recordsAWholeBatchInOnePass() {
        let log = NashEventLog(capacity: 3)
        func call(_ kind: MCPApprovalServer.ShellReportCall.Kind, token: Int? = nil)
            -> MCPApprovalServer.ShellReportCall
        {
            .init(kind: kind, sessionID: session.rawValue, shellID: "sh-1", execToken: token)
        }
        let landed = log.record(
            sessionID: session,
            calls: [
                call(.cd(from: "/", to: "/a")),
                call(.execStart(argv: ["make"], cwd: "/a", elapsedMs: 450, pipeline: nil), token: 7),
                call(.cd(from: "/a", to: "/b")),
                call(
                    .exec(argv: ["make"], cwd: "/a", exitCode: 0, durationMs: 900, pipeline: nil),
                    token: 7),
                call(.dropped(count: 2)),
            ], at: at)

        // Five calls in, five rows back — but the completion folded into the announcement, so the
        // ring holds four, and the trim to `capacity` ran once over the finished batch.
        #expect(landed.count == 5)
        #expect(landed[1].id == landed[3].id)
        #expect(log.events(for: session).map(\.summary) == ["make", "cd /b", "dropped 2 events under back-pressure"])
        #expect(log.events(for: session).map(\.isRunning) == [false, false, false])
        #expect(log.events(for: session)[0].detail == "in /a · 900ms")
        // The trimmed `cd /a` still counts, and the announcement still counts only once.
        #expect(log.total(for: session, kind: .cd) == 2)
        #expect(log.total(for: session, kind: .exec) == 1)
        #expect(log.total(for: session, kind: .dropped) == 1)
    }

    /// An announcement whose completion arrives in a *later* batch still finds its row — the ring
    /// the batch mutates is the same one the previous batch appended to.
    @MainActor
    @Test func batchesJoinAcrossDeliveries() {
        let log = NashEventLog()
        log.record(
            sessionID: session,
            calls: [
                .init(
                    kind: .execStart(argv: ["make"], cwd: "/w", elapsedMs: 450, pipeline: nil),
                    sessionID: session.rawValue, shellID: "sh-1", execToken: 7)
            ], at: at)
        log.record(
            sessionID: session,
            calls: [
                .init(
                    kind: .exec(argv: ["make"], cwd: "/w", exitCode: 2, durationMs: 9000, pipeline: nil),
                    sessionID: session.rawValue, shellID: "sh-1", execToken: 7)
            ], at: at + 9)
        #expect(log.events(for: session).count == 1)
        #expect(log.events(for: session)[0].detail == "in /w · exit 2 · 9.0s")
    }
}

/// The durable half of the feed (docs/NASH.md §9.4): rows written through `NashEventPersisting`
/// and restored into a later launch's ring, so a chat opened after a relaunch shows the shell
/// history it recorded before it.
@MainActor
@Suite struct NashEventPersistenceTests {
    private let session = SessionID(rawValue: "s1")

    private func store() throws -> GRDBMetadataStore { try GRDBMetadataStore(path: nil) }

    /// One launch's log, wired to `store`. Distinct `launchID`s are what make a second log stand
    /// in for the next run of the app.
    private func log(
        _ store: GRDBMetadataStore, launch: String, capacity: Int = 1000,
        retention: NashRetention = NashRetention()
    ) -> NashEventLog {
        NashEventLog(capacity: capacity, store: store, retention: retention, launchID: launch)
    }

    private func cd(_ log: NashEventLog, _ to: String, session: SessionID? = nil) {
        let id = session ?? self.session
        log.record(
            sessionID: id, call: .init(kind: .cd(from: "/", to: to), sessionID: id.rawValue))
    }

    @Test func restoresPreviousLaunchesIntoTheRing() async throws {
        let db = try store()
        let first = log(db, launch: "run-1")
        cd(first, "/a")
        cd(first, "/b")
        await first.flush()

        let second = log(db, launch: "run-2")
        cd(second, "/c")  // this run's own event, recorded before the chat was opened
        await second.hydrate(sessionID: session)

        // History first, live after — the panel reads the ring chronologically.
        #expect(second.events(for: session).map(\.summary) == ["cd /a", "cd /b", "cd /c"])
        // And the ids stay unique and ascending, which is what the list and its expansion set need.
        let ids = second.events(for: session).map(\.id)
        #expect(ids == ids.sorted() && Set(ids).count == ids.count)
        // The header counts restored emissions alongside this run's.
        #expect(second.total(for: session) == 3)
        #expect(second.total(for: session, kind: .cd) == 3)
    }

    /// The case that makes `launchID` necessary: a session emits while its chat is closed, so its
    /// rows are already on disk by the time hydration runs. Re-reading them would double the feed.
    @Test func doesNotRestoreThisLaunchesOwnRows() async throws {
        let db = try store()
        let live = log(db, launch: "run-1")
        cd(live, "/a")
        await live.flush()
        await live.hydrate(sessionID: session)

        #expect(live.events(for: session).map(\.summary) == ["cd /a"])
        #expect(live.total(for: session) == 1)
        // Hydration runs once per session, so a second open can't duplicate history either.
        await live.hydrate(sessionID: session)
        #expect(live.events(for: session).count == 1)
    }

    /// Restored rows have to carry everything a live row draws from, including the payload and the
    /// stage names a `pipe` row is built on — which come from the exec rows loaded beside it.
    @Test func restoresPayloadsAndPipelineStages() async throws {
        let db = try store()
        let first = log(db, launch: "run-1")
        let payload = Data("hello".utf8).base64EncodedString()
        for (index, argv) in [["curl", "-s", "u"], ["sh"]].enumerated() {
            first.record(
                sessionID: session,
                call: .init(
                    kind: .exec(
                        argv: argv, cwd: "/w", exitCode: 0, durationMs: 2,
                        pipeline: NashPipelineSlot(id: 3, index: index, length: 2)),
                    sessionID: session.rawValue, shellID: "sh-1"))
        }
        first.record(
            sessionID: session,
            call: .init(
                kind: .pipe(
                    pipelineID: 3, fromIndex: 0, toIndex: 1, fromText: "curl -s u", toText: "sh",
                    bytes: 5, truncated: true, hash: "abc", previewB64: payload),
                sessionID: session.rawValue, shellID: "sh-1"))
        await first.flush()

        let second = log(db, launch: "run-2")
        await second.hydrate(sessionID: session)
        let restored = try #require(second.events(for: session).last)
        #expect(restored.kind == .pipe)
        #expect(restored.summary == "curl -s u | sh → 5 B")
        #expect(restored.preview == "hello")
        #expect(restored.bytes == 5)
        #expect(restored.truncated)
        #expect(restored.hash == "abc")
        #expect(
            restored.details
                == .pipe(pipelineID: 3, fromIndex: 0, toIndex: 1, fromText: "curl -s u", toText: "sh"))
        // The pipe row can still name its ends: the stage index was rebuilt from the restored execs.
        let key = try #require(restored.pipelineKey)
        #expect(second.stages(for: session, pipeline: key)[0]?.command == "curl -s u")
        #expect(second.stages(for: session, pipeline: key)[1]?.command == "sh")
    }

    /// Emitter attribution survives the round trip, so a VM or container stays selectable in the
    /// picker after a relaunch.
    @Test func restoresEmitterEnvironments() async throws {
        let db = try store()
        let vm = NashEnvironment.macVM(name: "nucleic-mac-s1")
        let first = log(db, launch: "run-1")
        first.record(
            sessionID: session,
            call: .init(
                kind: .cd(from: "/", to: "/tmp"), sessionID: session.rawValue, environment: vm))
        await first.flush()

        let second = log(db, launch: "run-2")
        await second.hydrate(sessionID: session)
        #expect(second.events(for: session).map(\.environment) == [vm])
        #expect(second.environments(for: session) == [vm])
    }

    /// A command in flight is a fact about *this* run. It is written only once it returns, so a
    /// relaunch can never restore a row that claims to be running a process that died with the
    /// last launch — and the row that is finally stored is the whole command, from its start time.
    @Test func doesNotStoreCommandsThatHaveNotReturned() async throws {
        let db = try store()
        let first = log(db, launch: "run-1")
        first.record(
            sessionID: session,
            call: .init(
                kind: .execStart(argv: ["make"], cwd: "/w", elapsedMs: 450, pipeline: nil),
                sessionID: session.rawValue, shellID: "sh-1", execToken: 4))
        await first.flush()
        #expect(try await db.loadNashEvents(
            sessionID: session, excludingLaunch: "run-2", limit: 10).isEmpty)

        first.record(
            sessionID: session,
            call: .init(
                kind: .exec(argv: ["make"], cwd: "/w", exitCode: 0, durationMs: 61_000, pipeline: nil),
                sessionID: session.rawValue, shellID: "sh-1", execToken: 4))
        await first.flush()

        let second = log(db, launch: "run-2")
        await second.hydrate(sessionID: session)
        let restored = try #require(second.events(for: session).last)
        #expect(restored.summary == "make")
        #expect(!restored.isRunning)
        #expect(restored.detail == "in /w · 61.0s")
        #expect(second.total(for: session, kind: .exec) == 1)  // counted once, not twice
    }

    /// A ring smaller than the stored history takes the newest rows — the same tail it would have
    /// held had the app never quit.
    @Test func restoresNewestRowsWhenHistoryExceedsTheRing() async throws {
        let db = try store()
        let first = log(db, launch: "run-1")
        for i in 0..<5 { cd(first, "/\(i)") }
        await first.flush()

        let second = log(db, launch: "run-2", capacity: 2)
        await second.hydrate(sessionID: session)
        #expect(second.events(for: session).map(\.summary) == ["cd /3", "cd /4"])
        // Rows the ring couldn't hold are still counted — they happened, and they're still on disk.
        #expect(second.total(for: session) == 5)
    }

    /// Clearing is a decision about the history, not just the screen: rows the operator dropped
    /// must not come back at the next launch.
    @Test func clearAndForgetDropStoredRows() async throws {
        let db = try store()
        let other = SessionID(rawValue: "s2")
        let first = log(db, launch: "run-1")
        cd(first, "/a")
        cd(first, "/b", session: other)
        await first.flush()
        first.clear(sessionID: session)
        await first.flush()

        // The delete is ordered behind the flush, so let it land before reading back.
        var remaining = try await db.loadNashEvents(
            sessionID: session, excludingLaunch: "run-2", limit: 10)
        for _ in 0..<50 where !remaining.isEmpty {
            try await Task.sleep(for: .milliseconds(10))
            remaining = try await db.loadNashEvents(
                sessionID: session, excludingLaunch: "run-2", limit: 10)
        }
        #expect(remaining.isEmpty)
        // Another session's history is untouched by one session's clear.
        #expect(try await db.loadNashEvents(
            sessionID: other, excludingLaunch: "run-2", limit: 10).count == 1)

        first.forget(sessionID: other)
        var others = try await db.loadNashEvents(
            sessionID: other, excludingLaunch: "run-2", limit: 10)
        for _ in 0..<50 where !others.isEmpty {
            try await Task.sleep(for: .milliseconds(10))
            others = try await db.loadNashEvents(
                sessionID: other, excludingLaunch: "run-2", limit: 10)
        }
        #expect(others.isEmpty)
    }

    /// Disk retention, both bounds: the per-session row cap and the global payload budget, each
    /// evicting oldest-first.
    @Test func retentionBoundsRowsAndPayloadBytes() async throws {
        let db = try store()
        let rows = (0..<5).map {
            NashEventRecord(
                event: NashEventLog.project(
                    seq: UInt64($0), at: Date(timeIntervalSince1970: 1_700_000_000),
                    sessionID: session, kind: .cd(from: "/", to: "/\($0)")),
                launchID: "run-1")
        }
        try await db.saveNashEvents(
            rows, retention: NashRetention(rowsPerSession: 3, previewByteBudget: nil))
        #expect(
            try await db.loadNashEvents(sessionID: session, excludingLaunch: "x", limit: 10)
                .map(\.summary) == ["cd /2", "cd /3", "cd /4"])

        // Three 10-byte payloads against a 25-byte budget keeps the newest two.
        let payloads = (0..<3).map { i in
            NashEventRecord(
                event: NashEventLog.project(
                    seq: UInt64(100 + i), at: Date(timeIntervalSince1970: 1_700_000_000),
                    sessionID: session,
                    kind: .cmdsub(
                        bytes: 10, truncated: false, hash: "h",
                        previewB64: Data(String(repeating: "\(i)", count: 10).utf8)
                            .base64EncodedString())),
                launchID: "run-1")
        }
        try await db.saveNashEvents(
            payloads, retention: NashRetention(rowsPerSession: 100, previewByteBudget: 25))
        let kept = try await db.loadNashEvents(sessionID: session, excludingLaunch: "x", limit: 10)
        #expect(kept.compactMap(\.preview) == ["1111111111", "2222222222"])
        // The payload sweep is a global budget, so it also took the payload-free rows below them.
        #expect(kept.count == 2)
    }

    /// No store — the pre-existing behavior, unchanged: a live ring and nothing on disk.
    @Test func withoutAStoreTheLogStaysInMemory() async {
        let log = NashEventLog()
        cd(log, "/a")
        await log.flush()
        await log.hydrate(sessionID: session)
        #expect(log.events(for: session).map(\.summary) == ["cd /a"])
    }
}
