import Foundation
import Observation
import NucleicProtocol

/// The execution environment whose nash process emitted an event.
///
/// `id` is stable within a session and backs the Nash Viewer's picker. `label` is deliberately
/// operator-facing: the session's primary process is always "Agent" even when that process itself
/// happens to be sandboxed in Nucleic's managed container.
public struct NashEnvironment: Sendable, Hashable {
    public enum Kind: String, Sendable, Hashable, CaseIterable {
        case agent
        case host
        case macVM = "mac-vm"
        case linuxVM = "linux-vm"
        case container
        case app
    }

    public let kind: Kind
    public let id: String
    public let label: String

    public init(kind: Kind, id: String, label: String) {
        self.kind = kind
        self.id = id
        self.label = label
    }

    public static let agent = NashEnvironment(kind: .agent, id: "agent", label: "Agent")
    public static let host = NashEnvironment(kind: .host, id: "host", label: "Host")

    public static func macVM(name: String) -> NashEnvironment {
        NashEnvironment(kind: .macVM, id: "mac-vm:\(name)", label: "macOS VM")
    }

    public static func linuxVM(name: String) -> NashEnvironment {
        NashEnvironment(kind: .linuxVM, id: "linux-vm:\(name)", label: "Linux VM")
    }

    public static func container(name: String) -> NashEnvironment {
        NashEnvironment(kind: .container, id: "container:\(name)", label: name)
    }

    public static func app(origin: String, label: String = "App") -> NashEnvironment {
        NashEnvironment(kind: .app, id: "app:\(origin)", label: label)
    }

    /// Rebuild attribution supplied by nash's batch envelope. Unknown future kinds are ignored so
    /// an updated guest cannot make an older host reject otherwise-valid shell events.
    static func fromWire(kind: String?, id: String?, label: String?) -> NashEnvironment? {
        guard let rawKind = kind, let kind = Kind(rawValue: rawKind),
            let id, !id.isEmpty, let label, !label.isEmpty
        else { return nil }
        return NashEnvironment(kind: kind, id: id, label: label)
    }

    public var sortKey: String {
        let rank: Int = switch kind {
        case .agent: 0
        case .host: 1
        case .macVM: 2
        case .linuxVM: 3
        case .container: 4
        case .app: 5
        }
        return "\(rank)-\(label.lowercased())-\(id)"
    }
}

/// Where a command sat in the pipeline it ran as part of (`brush_core::gate::PipelineSlot`).
///
/// The `id` is shared with the `pipe` events for the same pipeline, which is what lets the viewer
/// name a link's producer and consumer (`curl … ▸ sh`) instead of showing bare stage indices.
public struct NashPipelineSlot: Sendable, Hashable, Codable {
    public let id: Int
    /// Zero-based position of this stage within the pipeline.
    public let index: Int
    /// How many stages the pipeline has.
    public let length: Int

    public init(id: Int, index: Int, length: Int) {
        self.id = id
        self.index = index
        self.length = length
    }
}

/// Identity of one pipeline invocation within a session.
///
/// nash's pipeline counter is **per shell process**, so two concurrent shells in the same session
/// both count from 1. The emitting shell is therefore part of the key — without it, one shell's
/// `a | b` would borrow another's stage names.
public struct NashPipelineKey: Sendable, Hashable {
    public let shellID: String
    public let pipelineID: Int

    public init(shellID: String, pipelineID: Int) {
        self.shellID = shellID
        self.pipelineID = pipelineID
    }
}

/// What nash saw running as one stage of a pipeline — the label a pipe row shows for the command
/// on either end of the link.
public struct NashPipelineStage: Sendable, Equatable {
    /// The stage's argv, joined (`grep -n .`).
    public let command: String
    public let exitCode: Int
    /// How many stages the pipeline has, as that stage reported it.
    public let length: Int

    public init(command: String, exitCode: Int, length: Int) {
        self.command = command
        self.exitCode = exitCode
        self.length = length
    }
}

/// One emission from the nash agent shell, flattened for display (docs/NASH.md §5).
///
/// `MCPApprovalServer.ShellReportCall.Kind` is the wire shape — an associated-value enum, one
/// payload per kind. This is its viewer projection, in two layers:
///
/// - ``details`` keeps the structured facts of each kind (a redirect's operator/fd/target, a pipe's
///   link indices, a command's argv and exit code), because the viewer draws a genuinely different
///   layout per kind — a redirect renders as a labelled flow, a pipe as a stage chain — and cannot
///   do that from a pre-formatted string.
/// - ``summary``/``detail`` are the *flat* projection of the same facts: one line each, identical
///   shape for every kind. They are what the panel searches and what a compact/accessible reading
///   falls back to, so the heterogeneous stream still filters as one list.
///
/// **Redaction**: previews are already bounded (64 KB) and redacted in-guest by `nash-observe`
/// before they ever reach the host, so nothing is re-redacted here. Everything else is structural.
public struct NashEvent: Identifiable, Sendable, Equatable {
    /// The emission kinds nash reports — the viewer's filter axis. Mirrors
    /// ``MCPApprovalServer/ShellReportCall/Kind`` one-for-one.
    public enum Kind: String, CaseIterable, Sendable {
        case exec
        case cd
        case export
        case redirect
        case cmdsub
        case pipe
        case fallback
        case dropped
        case policy

        /// How the filter menu groups the kinds. The split is what an operator is actually looking
        /// for: the data nash intercepted, the shell state it watched change, or a problem with the
        /// observation itself.
        public enum Group: String, CaseIterable, Sendable {
            case dataFlow
            case shell
            case problems

            public var label: String {
                switch self {
                case .dataFlow: "Data flow"
                case .shell: "Shell activity"
                case .problems: "Problems"
                }
            }
        }

        public var group: Group {
            switch self {
            case .redirect, .cmdsub, .pipe: .dataFlow
            case .exec, .cd, .export: .shell
            case .fallback, .dropped, .policy: .problems
            }
        }

        /// Full human label — the filter menu's wording. Deliberately not the wire spelling:
        /// "Cmdsub" is nash's term, "Substitution" is the operator's.
        public var label: String {
            switch self {
            case .exec: "Commands"
            case .cd: "Directory changes"
            case .export: "Variables"
            case .redirect: "Redirects"
            case .cmdsub: "Substitutions"
            case .pipe: "Pipes"
            case .fallback: "Bash fallbacks"
            case .dropped: "Dropped events"
            case .policy: "Shell policy"
            }
        }

        /// The row's compact tag. Lowercase and short so a column of them reads as a quiet gutter
        /// rather than eight shouting badges.
        public var badge: String {
            switch self {
            case .exec: "run"
            case .cd: "cd"
            case .export: "env"
            case .redirect: "file"
            case .cmdsub: "subst"
            case .pipe: "pipe"
            case .fallback: "bash"
            case .dropped: "lost"
            case .policy: "policy"
            }
        }

        /// Whether nash intercepted actual bytes for this kind — the kinds whose rows carry a
        /// payload worth opening.
        public var carriesPayload: Bool { group == .dataFlow }
    }

    /// The structured facts behind a row, one case per wire kind. The viewer renders each case with
    /// its own layout; ``NashEvent/summary`` is the same information flattened to one line.
    ///
    /// `Codable` because this is the one part of a row that has no flat column shape: the durable
    /// feed (``NashEventPersisting``) stores it as JSON. That makes the case names and their
    /// labels an on-disk format — adding a case is safe, renaming one drops the rows that used it
    /// (they decode to nil and are skipped, never failing the whole load).
    public enum Details: Sendable, Equatable, Codable {
        case command(argv: [String], cwd: String, exitCode: Int, durationMs: Int?, pipeline: NashPipelineSlot?)
        case directory(from: String, to: String)
        case environment(op: String, names: [String])
        /// `op` is the operator as written (`>`, `>>`, `<`, `2>`, `<<<`); `target` is nil for
        /// heredocs and here-strings, whose body has no path.
        case redirect(op: String, fd: Int, target: String?)
        case substitution
        /// `fromText`/`toText` are the stages either side of the link as written, rendered from the
        /// AST at spawn. Empty only for a batch from a nash that predates them, in which case the
        /// viewer falls back to the stage names it can derive from exec events.
        case pipe(pipelineID: Int, fromIndex: Int, toIndex: Int, fromText: String, toText: String)
        case fallback(reason: String, input: String)
        case dropped(count: Int)
        /// An operator lever nash saw (docs/NASH.md §4.3). `action` is `ignored` for one that came
        /// from the environment — the agent's own writable channel — and `honored` for one the
        /// trusted policy file set.
        case policy(lever: String, value: String, action: String, note: String)
    }

    /// Per-launch sequence — the row identity (`at` collides within a batch, since a nash batch of
    /// up to 200 events is decoded in one pass). Assigned by ``NashEventLog``, ascending in the
    /// order rows enter the ring, including rows restored from a previous run (which are numbered
    /// below every live one). Not durable: the same emission gets a different `seq` in the run
    /// that reloads it.
    public let seq: UInt64
    public let at: Date
    public let sessionID: SessionID
    public let environment: NashEnvironment
    /// The nash process that emitted this, from the batch envelope; empty when unreported. Scopes
    /// the per-process pipeline counter — see ``NashPipelineKey``.
    public let shellID: String
    public let kind: Kind
    /// The one-line statement of what nash observed (`git status` / `1> out.txt`).
    public let summary: String
    /// Optional structural second line (cwd, exit code, pipeline link).
    public let detail: String?
    /// The structured facts the viewer lays out per kind.
    public let details: Details
    /// The decoded, already-redacted preview of the data that crossed a redirect / command
    /// substitution / pipe. Nil for kinds that carry no payload, and for special files
    /// (`/dev/null`) where nash sends an empty preview.
    public let preview: String?
    /// Total bytes that crossed, for the data-flow kinds (nil otherwise).
    public let bytes: Int?
    /// True when `preview` is only the head of a larger payload.
    public let truncated: Bool
    /// nash's FNV-1a hash of the captured payload; nil when it sent none. Two rows with the same
    /// hash carried the same bytes — the cheap way to spot a payload repeating across a pipeline.
    public let hash: String?
    /// This command has not returned yet — the row is nash's `exec-start` announcement, and its
    /// exit code and duration are still unknown (docs/NASH.md §5.1). Only ever true for `.exec`
    /// rows, and only until the matching completion lands, which rewrites this row in place.
    ///
    /// Deliberately **not** part of ``details`` (and never persisted): a row is in flight only for
    /// as long as this run is watching it. A "still running" row restored from a previous launch
    /// would be claiming something about a process that has been dead since the app quit.
    public let isRunning: Bool

    public var id: UInt64 { seq }

    /// The pipeline this row belongs to, for the two kinds that name one: a `pipe` link, and a
    /// command nash saw running as one of that pipeline's stages.
    public var pipelineKey: NashPipelineKey? {
        switch details {
        case .pipe(let pipelineID, _, _, _, _):
            NashPipelineKey(shellID: shellID, pipelineID: pipelineID)
        case .command(_, _, _, _, .some(let slot)):
            NashPipelineKey(shellID: shellID, pipelineID: slot.id)
        default:
            nil
        }
    }

    public init(
        seq: UInt64, at: Date, sessionID: SessionID, environment: NashEnvironment = .agent,
        shellID: String = "", kind: Kind, summary: String,
        detail: String? = nil, details: Details, preview: String? = nil, bytes: Int? = nil,
        truncated: Bool = false, hash: String? = nil, isRunning: Bool = false
    ) {
        self.seq = seq
        self.at = at
        self.sessionID = sessionID
        self.environment = environment
        self.shellID = shellID
        self.kind = kind
        self.summary = summary
        self.detail = detail
        self.details = details
        self.preview = preview
        self.bytes = bytes
        self.truncated = truncated
        self.hash = hash
        self.isRunning = isRunning
    }

    /// The same row, with what the command turned out to do — how the in-flight row becomes the
    /// finished one. Everything identifying the row (``seq``, ``at``, its emitter) is kept, so the
    /// panel updates the row where it already is instead of adding a second one for the same
    /// command; only what was unknown while it ran is taken from `finished`.
    func completed(by finished: NashEvent) -> NashEvent {
        NashEvent(
            seq: seq, at: at, sessionID: sessionID, environment: environment, shellID: shellID,
            kind: finished.kind, summary: finished.summary, detail: finished.detail,
            details: finished.details, preview: finished.preview, bytes: finished.bytes,
            truncated: finished.truncated, hash: finished.hash)
    }
}

/// The log of everything nash emitted, per session — what the Nash Viewer panel renders.
///
/// Kept **per session** rather than as one global ring so a chatty session can never evict another
/// session's history: the panel's promise is "all emissions from nash for *this* session", and a
/// shared ring would quietly break it. Each session gets its own oldest-first buffer trimmed to
/// `capacity`; lifetime per-kind totals survive the trim so the header stays honest about what
/// scrolled away.
///
/// **Durable** (docs/NASH.md §9.4): every recorded row is also written to `nash_event` through
/// ``NashEventPersisting``, and ``hydrate(sessionID:)`` reads a session's previous runs back into
/// the ring when its chat is opened — a relaunch no longer empties the panel. Two layers, because
/// they answer different questions: the in-memory ring is what a live viewer scrolls (bounded by
/// `capacity`, cheap to filter), and the table is the audit history behind it (bounded by
/// ``NashRetention``, an order of magnitude larger). Writes are buffered and flushed in batches —
/// nash delivers up to 200 events per batch, and a transaction per row would put SQLite in the
/// path of every pipe the agent opens. The cost of that buffer is the last sub-second of events on
/// a hard kill; the `CMDTRACE` unified-logging trail `AppStore.observeShellEvent` writes remains
/// the synchronous record.
///
/// `@MainActor` alongside `AppStore`, whose shell-event path already runs there, so recording is a
/// plain synchronous call with no ordering races; only the flush and the hydration read leave the
/// actor.
@MainActor
@Observable
public final class NashEventLog {
    /// Oldest-first events per session, each trimmed to `capacity` (the viewer renders newest last).
    public private(set) var eventsBySession: [SessionID: [NashEvent]] = [:]
    /// Lifetime counts per session per kind, unaffected by trimming or `clear`.
    public private(set) var totalsBySession: [SessionID: [NashEvent.Kind: Int]] = [:]
    /// Emitters seen during this session, unaffected by ring trimming or `clear`, so an otherwise
    /// quiet VM/container remains selectable after a burst of agent events rolls its rows away.
    public private(set) var environmentsBySession: [SessionID: Set<NashEnvironment>] = [:]
    /// Max retained events per session.
    public let capacity: Int
    /// Max pipelines per session kept in the stage index.
    public let pipelineCapacity: Int

    /// Stage commands per pipeline, built from the exec events nash stamped with a pipeline slot.
    ///
    /// Kept outside the event ring, and outside `clear`, on purpose: a pipe row's whole value is
    /// naming the commands either side of the link, and the exec events that name them are trimmed
    /// on the same schedule as the pipe rows themselves. Bounded separately by `pipelineCapacity`,
    /// evicting whole pipelines oldest-first.
    private var pipelines: [SessionID: PipelineIndex] = [:]

    /// One session's pipeline→stage map plus the insertion order that drives eviction.
    private struct PipelineIndex {
        var stages: [NashPipelineKey: [Int: NashPipelineStage]] = [:]
        var order: [NashPipelineKey] = []

        /// First writer per stage wins. A stage that is a compound command (`{ a; b; } | c`) reports
        /// one exec per command inside it — all genuinely part of that stage — and the first is the
        /// one the stage opened with. (Command substitutions inside a stage are *not* among them:
        /// they run during expansion, before the stage's own command, and nash deliberately does
        /// not stamp them — see `invoke_command_in_subshell_and_get_output`.)
        mutating func insert(_ stage: NashPipelineStage, at index: Int, key: NashPipelineKey, capacity: Int) {
            if stages[key] == nil {
                stages[key] = [:]
                order.append(key)
                if order.count > capacity {
                    stages.removeValue(forKey: order.removeFirst())
                }
            }
            if stages[key]?[index] == nil { stages[key]?[index] = stage }
        }
    }

    /// Live rows number up from here; rows restored by ``hydrate(sessionID:)`` number *down* from
    /// just below it. Both stay unique (which is all `id` needs to be), and history always sorts
    /// before anything this run recorded — a seq ordering that agrees with the array order the
    /// panel reads, without knowing how many rows a session will turn out to have.
    private static let liveSeqBase: UInt64 = 1 << 32
    private var nextSeq: UInt64 = liveSeqBase
    private var nextHistorySeq: UInt64 = liveSeqBase - 1

    // MARK: Durable feed

    /// Where the rows go (nil → in-memory only: a store that doesn't persist nash events, or a
    /// test that doesn't care). See ``NashEventPersisting``.
    private let store: (any NashEventPersisting)?
    /// This app run's id, stamped onto every row it writes. Hydration loads rows from *other*
    /// launches only: a session that emitted while its chat was closed has already flushed those
    /// rows, and re-reading them would double every one of them in the panel.
    public let launchID: String
    private let retention: NashRetention
    /// Rows recorded but not yet written. Drained by ``flush()``.
    private var pending: [NashEventRecord] = []
    private var flushTask: Task<Void, Never>?
    /// Sessions ``hydrate(sessionID:)`` has already run for — restoring twice would duplicate a
    /// session's whole history.
    private var hydrated: Set<SessionID> = []
    /// Per-kind counts restored from previous runs, kept apart from `totalsBySession` (which
    /// counts *this* run exactly, past the ring trim). ``total(for:)`` reports the sum.
    private var priorTotalsBySession: [SessionID: [NashEvent.Kind: Int]] = [:]
    /// Preview bytes written since the last global budget sweep. Starts "over" so the first flush
    /// of every launch sweeps — otherwise a quiet run never reclaims what a noisy one left.
    private var bytesSinceSweep = Int.max

    /// Flush when a batch this big has piled up (nash's own batches cap at 200)…
    private static let flushThreshold = 256
    /// …or this long after the first row of a quiet trickle.
    private static let flushDelay = Duration.milliseconds(750)
    /// Payload written between global budget sweeps. The sweep is a whole-table scan, so it is
    /// worth paying only once a meaningful amount of new payload has landed.
    private static let sweepInterval = 8 * 1024 * 1024

    public init(
        capacity: Int = 1000, pipelineCapacity: Int = 256,
        store: (any NashEventPersisting)? = nil, retention: NashRetention = NashRetention(),
        launchID: String = UUID().uuidString
    ) {
        self.capacity = max(1, capacity)
        self.pipelineCapacity = max(1, pipelineCapacity)
        self.store = store
        self.retention = retention
        self.launchID = launchID
    }

    /// This session's retained emissions, oldest first.
    public func events(for sessionID: SessionID) -> [NashEvent] {
        eventsBySession[sessionID] ?? []
    }

    /// Environments that have emitted during this session, in picker order.
    public func environments(for sessionID: SessionID) -> [NashEnvironment] {
        Array(environmentsBySession[sessionID] ?? []).sorted { $0.sortKey < $1.sortKey }
    }

    /// How many emissions of `kind` this session has produced — exact for this run (including
    /// ones the ring trimmed away), plus what previous runs left on disk once
    /// ``hydrate(sessionID:)`` has restored them. Emissions dropped by *disk* retention are gone
    /// and are not counted.
    public func total(for sessionID: SessionID, kind: NashEvent.Kind) -> Int {
        (totalsBySession[sessionID]?[kind] ?? 0) + (priorTotalsBySession[sessionID]?[kind] ?? 0)
    }

    /// Total emissions this session has produced, on the same terms as the per-kind count.
    public func total(for sessionID: SessionID) -> Int {
        (totalsBySession[sessionID]?.values.reduce(0, +) ?? 0)
            + (priorTotalsBySession[sessionID]?.values.reduce(0, +) ?? 0)
    }

    /// The commands nash saw running as each stage of `pipeline`, keyed by stage index.
    ///
    /// Empty until the stages exit — a link is reported by its tee the moment the producer closes,
    /// which can beat the producer's own exec event to the host. The viewer therefore has to render
    /// a pipe row that names no stages yet, and let it fill in.
    public func stages(for sessionID: SessionID, pipeline: NashPipelineKey) -> [Int: NashPipelineStage] {
        pipelines[sessionID]?.stages[pipeline] ?? [:]
    }

    /// Record one emission, projecting the wire `Kind` onto the flat row shape.
    ///
    /// The one emission that does not simply append is a command's completion: when nash already
    /// announced that command as running (docs/NASH.md §5.1), the finished row **replaces** the
    /// in-flight one where it sits rather than landing as a second row for the same command — see
    /// ``append(_:sessionID:call:to:)``.
    @discardableResult
    public func record(
        sessionID: SessionID, call: MCPApprovalServer.ShellReportCall,
        environment: NashEnvironment? = nil, at: Date = Date()
    ) -> NashEvent {
        record(sessionID: sessionID, calls: [call], environment: environment, at: at)[0]
    }

    /// Record one nash **batch** — every event of a `/shell-event` post, in the order nash emitted
    /// them. Returns the rows as they landed, one per call.
    ///
    /// This is the shape the live route delivers (docs/NASH_STREAM_PERF_PLAN.md §P1): a batch of up
    /// to 200 events costs *one* in-place mutation of the session's ring, one trim, and therefore
    /// one `@Observable` invalidation, rather than paying all three per event. The ring is mutated
    /// through ``withRing(_:_:)`` so appending never copies the 1000-row buffer.
    @discardableResult
    public func record(
        sessionID: SessionID, calls: [MCPApprovalServer.ShellReportCall],
        environment: NashEnvironment? = nil, at: Date = Date()
    ) -> [NashEvent] {
        guard !calls.isEmpty else { return [] }
        var landed: [NashEvent] = []
        landed.reserveCapacity(calls.count)
        // Trimming is deferred to the end of the batch so the ring is walked once; the in-flight
        // rows the trim invalidates are pruned after, outside the ring's exclusive access.
        var trimmedFrom: UInt64?
        withRing(sessionID) { bucket in
            for call in calls {
                // A running row is stamped with when its command *started*, not when we heard about
                // it: nash waits out its announce threshold and then the batch waits out a flush, so
                // a row clocked from arrival would show a build as a second younger than it is — and
                // would keep that error for as long as the row counts up. The completion inherits
                // this stamp, so the finished row also sits at the time the command began.
                var at = at
                if case .execStart(_, _, let elapsedMs, _) = call.kind {
                    at -= Double(elapsedMs) / 1000
                }
                let event = Self.project(
                    seq: nextSeq, at: at, sessionID: sessionID,
                    environment: environment ?? call.environment ?? .agent,
                    shellID: call.shellID ?? "", kind: call.kind)
                landed.append(append(event, sessionID: sessionID, call: call, to: &bucket))
            }
            if bucket.count > capacity {
                bucket.removeFirst(bucket.count - capacity)
                trimmedFrom = bucket.first?.seq ?? 0
            }
        }
        if let trimmedFrom { pruneRunningRows(sessionID: sessionID, below: trimmedFrom) }
        return landed
    }

    /// Land one projected row in `bucket` — appending it, or folding it into the in-flight row it
    /// completes — and do the bookkeeping that follows. Returns the row as it landed.
    ///
    /// Takes the ring `inout` rather than reaching for `eventsBySession`, because its caller is
    /// already holding that property under exclusive access (see ``withRing(_:_:)``); everything
    /// else it touches is a different property.
    private func append(
        _ event: NashEvent, sessionID: SessionID, call: MCPApprovalServer.ShellReportCall,
        to bucket: inout [NashEvent]
    ) -> NashEvent {
        if case .exec = call.kind,
            let landed = retireRunningRow(sessionID: sessionID, call: call, event: event, in: &bucket)
        {
            return landed
        }
        nextSeq += 1
        bucket.append(event)
        environmentsBySession[sessionID, default: []].insert(event.environment)
        // An in-flight row is a *promise* of a row, not one of its own: it is not counted (its
        // completion is the emission nash made), not written to the durable feed (a running row
        // means nothing once this run is over), and not credited as a pipeline stage — the stage
        // index takes the first writer per slot, and this row does not know its exit code yet.
        // All three happen when the command returns.
        guard !event.isRunning else {
            if let key = runningKey(sessionID: sessionID, call: call) {
                runningRows[key] = event.seq
            }
            return event
        }
        indexPipelineStage(event, sessionID: sessionID)
        totalsBySession[sessionID, default: [:]][event.kind, default: 0] += 1
        enqueue(event)
        return event
    }

    /// Mutate one session's ring **in place**.
    ///
    /// `&eventsBySession[sessionID, default: []]` threads `_modify` accessors all the way down — the
    /// `@Observable` property's and the dictionary subscript's — so the body writes straight into
    /// the stored buffer: no copy-on-write of the 1000-row array, and exactly one
    /// willSet/didSet pair for the whole body however many rows it appends. Reading or writing
    /// `eventsBySession` from inside `body` would overlap that exclusive access, so the body is
    /// handed the ring and must use it.
    private func withRing<R>(_ sessionID: SessionID, _ body: (inout [NashEvent]) -> R) -> R {
        body(&eventsBySession[sessionID, default: []])
    }

    /// Identity of one command run in flight: nash's per-shell gate token, scoped by the shell that
    /// issued it (the token counter restarts in every nash process, exactly like the pipeline one).
    private struct RunningKey: Hashable {
        let sessionID: SessionID
        let shellID: String
        let token: Int
    }

    /// Rows still showing as running, keyed by the command they are waiting on → the row's `seq`.
    /// Small by nature (one entry per command actually in flight) and never restored from disk.
    private var runningRows: [RunningKey: UInt64] = [:]

    /// Keyed on the session the row was *recorded under* rather than the one on the wire: an
    /// app-scoped shell (Build & Run) has no session and is filed under a synthetic origin, and its
    /// running rows have to be found again under that same id.
    private func runningKey(
        sessionID: SessionID, call: MCPApprovalServer.ShellReportCall
    ) -> RunningKey? {
        guard let token = call.execToken else { return nil }
        return RunningKey(sessionID: sessionID, shellID: call.shellID ?? "", token: token)
    }

    /// Fold a completion into the in-flight row that was waiting for it, if there is one: the row
    /// keeps its place, its clock (the time the command *started*) and its identity, and gains the
    /// exit code and duration it was missing. Returns the row it landed on, or nil when this
    /// command was never announced as running — the ordinary case, which `record` then appends.
    ///
    /// A row the ring has already trimmed away is *not* resurrected: the key is dropped and the
    /// completion appends like any other, because a row nobody can see is not one to update.
    ///
    /// `bucket` is the session's ring, held `inout` by the batch that is recording into it — the
    /// row this folds into may well have been appended by an earlier call of the same batch.
    private func retireRunningRow(
        sessionID: SessionID, call: MCPApprovalServer.ShellReportCall, event: NashEvent,
        in bucket: inout [NashEvent]
    ) -> NashEvent? {
        guard let key = runningKey(sessionID: sessionID, call: call),
            let seq = runningRows.removeValue(forKey: key),
            // Newest-last, and a running row is by definition recent, so this finds it immediately.
            let index = bucket.lastIndex(where: { $0.seq == seq })
        else { return nil }
        let landed = bucket[index].completed(by: event)
        bucket[index] = landed
        indexPipelineStage(landed, sessionID: sessionID)
        totalsBySession[sessionID, default: [:]][landed.kind, default: 0] += 1
        enqueue(landed)
        return landed
    }

    /// Forget in-flight rows the ring has trimmed away. Without this, a shell that died mid-command
    /// would leave its key behind forever — the map is tiny, but it is not self-cleaning.
    private func pruneRunningRows(sessionID: SessionID, below oldest: UInt64) {
        guard !runningRows.isEmpty else { return }
        runningRows = runningRows.filter { $0.key.sessionID != sessionID || $0.value >= oldest }
    }

    /// Join an exec event that ran as a pipeline stage into the stage index the `pipe` rows either
    /// side of it are named from. Shared by live recording and hydration — a restored pipe row has
    /// to name its ends too, and the exec rows that name them come back in the same load.
    private func indexPipelineStage(_ event: NashEvent, sessionID: SessionID) {
        guard case .command(let argv, _, let exitCode, _, .some(let slot)) = event.details,
            let key = event.pipelineKey
        else { return }
        pipelines[sessionID, default: PipelineIndex()].insert(
            NashPipelineStage(
                command: argv.joined(separator: " "), exitCode: exitCode, length: slot.length),
            at: slot.index, key: key, capacity: pipelineCapacity)
    }

    /// Drop this session's retained rows — in memory *and* on disk, since a row the operator
    /// cleared must not reappear at the next launch. Lifetime totals stay; they count what
    /// happened, and clearing the rows doesn't unhappen it.
    public func clear(sessionID: SessionID) {
        eventsBySession[sessionID] = []
        // The rows those commands would have landed on are gone, so their completions append as
        // ordinary finished rows — which is the honest outcome of clearing mid-command.
        runningRows = runningRows.filter { $0.key.sessionID != sessionID }
        // Nothing already recorded for this session may still land after the delete. Dropping the
        // un-flushed rows here and deleting *behind* a flush is what orders the two.
        pending.removeAll { $0.sessionID == sessionID }
        delete(sessionID)
    }

    /// Forget a session entirely (it was deleted). Its stored history goes with it — nothing can
    /// view it again, and `nash_event` has no session foreign key to cascade it away.
    public func forget(sessionID: SessionID) {
        eventsBySession.removeValue(forKey: sessionID)
        totalsBySession.removeValue(forKey: sessionID)
        priorTotalsBySession.removeValue(forKey: sessionID)
        environmentsBySession.removeValue(forKey: sessionID)
        pipelines.removeValue(forKey: sessionID)
        runningRows = runningRows.filter { $0.key.sessionID != sessionID }
        hydrated.remove(sessionID)
        pending.removeAll { $0.sessionID == sessionID }
        delete(sessionID)
    }

    // MARK: - Durable feed (docs/NASH.md §9.4)

    /// Restore what previous runs recorded for this session, oldest-first ahead of anything this
    /// run has already logged for it. Called when a chat is opened; idempotent, and a no-op
    /// without a store.
    ///
    /// Bounded by the ring: at most `capacity` rows are held, so a session with more history than
    /// that gets its newest `capacity` — the same tail the ring would have kept had the app never
    /// quit. The full history stays on disk under ``NashRetention``.
    public func hydrate(sessionID: SessionID) async {
        guard let store, !hydrated.contains(sessionID) else { return }
        hydrated.insert(sessionID)
        // Totals first, and unconditionally: the header counts restored emissions even when the
        // ring is already full of live ones and no row can be loaded.
        priorTotalsBySession[sessionID] =
            (try? await store.nashEventTotals(sessionID: sessionID, excludingLaunch: launchID)) ?? [:]
        let room = capacity - (eventsBySession[sessionID]?.count ?? 0)
        guard room > 0,
            let records = try? await store.loadNashEvents(
                sessionID: sessionID, excludingLaunch: launchID, limit: room),
            !records.isEmpty
        else { return }
        // One contiguous descending block, assigned ascending within itself, so restored rows read
        // in the order they happened and still sort below every live row.
        let base = nextHistorySeq - UInt64(records.count) + 1
        nextHistorySeq = base - 1
        let history = records.enumerated().map { $0.element.event(seq: base + UInt64($0.offset)) }
        for event in history {
            indexPipelineStage(event, sessionID: sessionID)
            environmentsBySession[sessionID, default: []].insert(event.environment)
        }
        var bucket = history
        bucket.append(contentsOf: eventsBySession[sessionID] ?? [])
        if bucket.count > capacity { bucket.removeFirst(bucket.count - capacity) }
        eventsBySession[sessionID] = bucket
    }

    /// Write out everything buffered. Awaited by tests and by the clear/delete paths (which must
    /// not race a flush); the normal path is the debounce in ``enqueue(_:)``.
    public func flush() async {
        flushTask = nil
        guard let store, !pending.isEmpty else { return }
        let batch = pending
        pending = []
        var retention = self.retention
        if bytesSinceSweep >= Self.sweepInterval {
            bytesSinceSweep = 0
        } else {
            retention.previewByteBudget = nil  // not enough new payload to pay for the scan
        }
        // A failed write costs history, never a live row: the panel is already showing this batch.
        try? await store.saveNashEvents(batch, retention: retention)
    }

    private func enqueue(_ event: NashEvent) {
        guard store != nil else { return }
        pending.append(NashEventRecord(event: event, launchID: launchID))
        if bytesSinceSweep != Int.max {
            bytesSinceSweep += event.preview?.utf8.count ?? 0
        }
        guard pending.count < Self.flushThreshold else {
            flushTask = Task { await self.flush() }
            return
        }
        guard flushTask == nil else { return }
        flushTask = Task {
            try? await Task.sleep(for: Self.flushDelay)
            await self.flush()
        }
    }

    private func delete(_ sessionID: SessionID) {
        guard let store else { return }
        Task {
            await self.flush()  // ordered behind any batch already on its way to the table
            try? await store.deleteNashEvents(sessionID: sessionID)
        }
    }

    // MARK: - Projection
    //
    // Pure and `nonisolated` — the projection touches no instance state, so it stays testable
    // (and callable) without hopping to the main actor.

    nonisolated static func project(
        seq: UInt64, at: Date, sessionID: SessionID, environment: NashEnvironment = .agent,
        shellID: String = "", kind: MCPApprovalServer.ShellReportCall.Kind
    ) -> NashEvent {
        func event(
            _ k: NashEvent.Kind, _ summary: String, detail: String? = nil,
            details: NashEvent.Details, preview: String? = nil,
            bytes: Int? = nil, truncated: Bool = false, hash: String = "",
            isRunning: Bool = false
        ) -> NashEvent {
            NashEvent(
                seq: seq, at: at, sessionID: sessionID, environment: environment, shellID: shellID,
                kind: k, summary: summary, detail: detail, details: details, preview: preview,
                bytes: bytes, truncated: truncated, hash: hash.isEmpty ? nil : hash,
                isRunning: isRunning)
        }

        switch kind {
        case .execStart(let argv, let cwd, _, let pipeline):
            // The same row shape as a finished command, minus the two facts nobody knows yet: the
            // exit code reads 0 because nothing has failed (the row shows no exit at all until the
            // command returns), and the duration is nil because it is still being spent.
            return event(
                .exec, argv.joined(separator: " "), detail: "in \(cwd) · running",
                details: .command(
                    argv: argv, cwd: cwd, exitCode: 0, durationMs: nil, pipeline: pipeline),
                isRunning: true)
        case .exec(let argv, let cwd, let exitCode, let durationMs, let pipeline):
            var detail = "in \(cwd)"
            if exitCode != 0 { detail += " · exit \(exitCode)" }
            if let ms = durationMs { detail += " · \(formatDuration(ms))" }
            return event(
                .exec, argv.joined(separator: " "), detail: detail,
                details: .command(
                    argv: argv, cwd: cwd, exitCode: exitCode, durationMs: durationMs,
                    pipeline: pipeline))
        case .cd(let from, let to):
            return event(.cd, "cd \(to)", detail: "from \(from)", details: .directory(from: from, to: to))
        case .export(let op, let names):
            // Names only — nash never reports values (docs/NASH.md §5.4).
            return event(
                .export, "\(op) \(names.joined(separator: " "))",
                details: .environment(op: op, names: names))
        case .redirect(let op, let fd, let target, let bytes, let truncated, let hash, let previewB64):
            // nash already prefixes the fd onto the operator for stderr (`2>`), so composing the two
            // naively printed `22> err.log` — and this string is what search and Copy hand back.
            return event(
                .redirect, "\(fd)\(op.drop(while: \.isNumber)) \(target ?? "(inline)")",
                detail: formatBytes(bytes),
                details: .redirect(op: op, fd: fd, target: target),
                preview: decode(previewB64), bytes: bytes, truncated: truncated, hash: hash)
        case .cmdsub(let bytes, let truncated, let hash, let previewB64):
            return event(
                .cmdsub, "$(…) → \(formatBytes(bytes))", details: .substitution,
                preview: decode(previewB64), bytes: bytes, truncated: truncated, hash: hash)
        case .pipe(
            let pipelineID, let fromIndex, let toIndex, let fromText, let toText, let bytes,
            let truncated, let hash, let previewB64):
            // Name the ends when nash sent them — `curl … | sh` is the line worth searching and
            // pasting. Falling back to 1-based stage numbers, matching the viewer's "stage 2 of 3":
            // nash's own indices are 0-based, but a row and its copied text disagreeing about which
            // stage is which is worse than either convention alone.
            let producer = fromText.isEmpty ? "stage \(fromIndex + 1)" : fromText
            let consumer = toText.isEmpty ? "stage \(toIndex + 1)" : toText
            return event(
                .pipe, "\(producer) | \(consumer) → \(formatBytes(bytes))",
                detail: "pipeline \(pipelineID)",
                details: .pipe(
                    pipelineID: pipelineID, fromIndex: fromIndex, toIndex: toIndex,
                    fromText: fromText, toText: toText),
                preview: decode(previewB64), bytes: bytes, truncated: truncated, hash: hash)
        case .fallback(let reason, let input):
            return event(
                .fallback, "re-execed bash: \(reason)", detail: input.isEmpty ? nil : input,
                details: .fallback(reason: reason, input: input))
        case .dropped(let count):
            return event(
                .dropped, "dropped \(count) event\(count == 1 ? "" : "s") under back-pressure",
                details: .dropped(count: count))
        case .policy(let lever, let value, let action, let note):
            // The summary leads with the verb because that is the fact an operator scanning the
            // feed needs: "ignored NUCLEIC_NASH_DISABLE=1" is a command that tried to switch its
            // own observation off, and it should read that way at a glance.
            let assignment = value.isEmpty ? lever : "\(lever)=\(value)"
            return event(
                .policy, "\(action) \(assignment)", detail: note.isEmpty ? nil : note,
                details: .policy(lever: lever, value: value, action: action, note: note))
        }
    }

    /// nash sends previews base64-encoded so arbitrary bytes survive JSON. Non-UTF-8 payloads
    /// (a binary redirect) decode to nil rather than mojibake — the row still shows the byte count.
    nonisolated private static func decode(_ base64: String) -> String? {
        guard !base64.isEmpty, let data = Data(base64Encoded: base64),
            let text = String(data: data, encoding: .utf8), !text.isEmpty
        else { return nil }
        return text
    }

    /// Shared with the Nash Viewer so a row's inline byte count and its payload inspector's header
    /// never disagree about how the same number reads.
    public nonisolated static func formatBytes(_ bytes: Int) -> String {
        if bytes < 1024 { return "\(bytes) B" }
        if bytes < 1024 * 1024 { return String(format: "%.1f KB", Double(bytes) / 1024) }
        return String(format: "%.1f MB", Double(bytes) / (1024 * 1024))
    }

    public nonisolated static func formatDuration(_ ms: Int) -> String {
        ms < 1000 ? "\(ms)ms" : String(format: "%.1fs", Double(ms) / 1000)
    }
}
