import Foundation
import Testing

@testable import NucleicCore

/// The safety net required by docs/TRANSCRIPT_INCREMENTAL_PROJECTION.md, mirrored from the iOS
/// suite (`Tests/NucleicRemoteProjectionTests`): replay an event stream and assert
/// `incremental(prefix) == TranscriptProjection.items(prefix, worktreeRoot:)` — items *and*
/// `lockLines` — for **every** prefix. This mechanically proves the seam never reorders items,
/// orphans a lock note, splits a tool run or a thinking-tokens meter, or bakes a stale message —
/// the entire correctness argument of `IncrementalTranscriptProjection`.
@Suite struct IncrementalTranscriptProjectionTests {

    static let root = "/repo/wt"

    // MARK: - Stream builder

    /// Grows a synthetic `AgentEvent` stream with the same shapes the decoders emit — streaming
    /// deltas, whole-message finals, tool lifecycles with input-delta bursts and trailing file
    /// changes, subagent scopes, lock-lifecycle notes, thinking-tokens pings — so the equivalence
    /// property is exercised on realistic seams.
    final class Stream {
        private(set) var events: [AgentEvent] = []
        private var seq: UInt64

        init(startSeq: UInt64 = 0) { self.seq = startSeq }

        func add(_ kind: AgentEvent.Kind, nativeType: String? = nil) {
            seq += 1
            events.append(AgentEvent(
                sessionID: SessionID(rawValue: "s"), seq: seq,
                at: Date(timeIntervalSince1970: 1_750_000_000),
                backend: .claudeCode, nativeType: nativeType, kind: kind))
        }

        func started(cwd: String = IncrementalTranscriptProjectionTests.root) {
            add(.sessionStarted(SessionStarted(
                backendSessionID: "native", model: "claude-fable-5", cwd: cwd, toolNames: [])))
        }
        func user(_ id: String, _ text: String) {
            add(.userText(TextChunk(messageID: id, text: text, isPartial: false)))
        }
        func textDelta(_ id: String, _ text: String, parent: String? = nil) {
            add(.assistantText(TextChunk(messageID: id, text: text, isPartial: true, parentToolCallID: parent)))
        }
        func textFinal(_ id: String, _ text: String, parent: String? = nil) {
            add(.assistantText(TextChunk(messageID: id, text: text, isPartial: false, parentToolCallID: parent)))
        }
        func thinkingDelta(_ id: String, _ text: String, parent: String? = nil) {
            add(.thinking(TextChunk(messageID: id, text: text, isPartial: true, parentToolCallID: parent)))
        }
        func thinkingFinal(_ id: String, _ text: String, parent: String? = nil) {
            add(.thinking(TextChunk(messageID: id, text: text, isPartial: false, parentToolCallID: parent)))
        }
        func toolStart(_ id: String, _ name: String, _ input: JSONValue = .object([:]), parent: String? = nil) {
            add(.toolCallStarted(ToolCall(toolCallID: id, name: name, input: input, parentToolCallID: parent)))
        }
        func inputDelta(_ id: String, _ partial: String) {
            add(.toolCallInputDelta(ToolInputDelta(toolCallID: id, partialJSON: partial)))
        }
        func toolComplete(_ id: String, _ name: String, _ input: JSONValue = .object([:]), parent: String? = nil) {
            add(.toolCallCompleted(ToolCall(toolCallID: id, name: name, input: input, parentToolCallID: parent)))
        }
        func result(_ id: String, _ content: JSONValue = .string("ok"), isError: Bool = false) {
            add(.toolResult(ToolResult(toolCallID: id, content: content, isError: isError)))
        }
        func fileChange(_ id: String?, _ path: String, _ kind: FileChange.ChangeKind = .update) {
            add(.fileChange(FileChange(path: path, kind: kind, toolCallID: id)))
        }
        func note(_ text: String, icon: String? = nil, lockEvent: Bool = false, lock: NoteLock? = nil,
                  toolCallID: String? = nil, narratesTool: Bool = false) {
            add(.note(NoteEvent(text: text, icon: icon, lockEvent: lockEvent, lock: lock,
                                toolCallID: toolCallID, narratesTool: narratesTool)))
        }
        func usage() { add(.usage(Usage(inputTokens: 100, outputTokens: 20, contextInputTokens: 1000))) }
        func turn() { add(.turnCompleted(TurnCompleted(stopReason: "end_turn", usage: nil))) }
        func finished(_ outcome: RunFinished.Outcome = .completed) {
            add(.runFinished(RunFinished(outcome: outcome)))
        }
        func error(_ message: String) {
            add(.error(AgentError(recoverable: false, message: message)))
        }
        func approval(_ toolName: String) {
            add(.approvalRequested(ApprovalRequest(
                id: ApprovalID(rawValue: "appr-\(seq)"), sessionID: SessionID(rawValue: "s"),
                toolCallID: nil, toolName: toolName, input: .object([:]), title: toolName,
                risk: .execute, createdAt: Date(timeIntervalSince1970: 1_750_000_000))))
        }
        func raw(_ body: String) {
            add(.raw(Raw(native: .string(body))), nativeType: "mystery")
        }
        /// A Claude `system/thinking_tokens` progress ping — coalesced by the projection into one
        /// live `.thinkingTokens` meter per run.
        func thinkingTokens(_ total: Int, delta: Int = 10) {
            add(.raw(Raw(native: .object([
                "estimated_tokens": .number(Double(total)),
                "estimated_tokens_delta": .number(Double(delta)),
            ]))), nativeType: "system/thinking_tokens")
        }

        /// A full edit-class tool lifecycle: Edit start/complete/result + inferred fileChange,
        /// the exact adjacency ClaudeStreamDecoder emits.
        func edit(_ id: String, path: String, parent: String? = nil) {
            let input = JSONValue.object(["file_path": .string(path)])
            toolStart(id, "Edit", input, parent: parent)
            toolComplete(id, "Edit", input, parent: parent)
            result(id)
            fileChange(id, path)
        }
    }

    // MARK: - The property

    /// Elementwise equality for summarizable units (tuples, so no synthesized `==`).
    static func unitsEqual(
        _ a: [TranscriptProjection.SummarizableUnit], _ b: [TranscriptProjection.SummarizableUnit]
    ) -> Bool {
        guard a.count == b.count else { return false }
        for (x, y) in zip(a, b) {
            guard x.signature == y.signature, x.calls == y.calls,
                  x.allFinished == y.allFinished else { return false }
        }
        return true
    }

    /// Replays every prefix of `events` through one live projector instance and diffs it against
    /// the one-shot fold; optionally asserts the watermark actually advanced (so a projector that
    /// degenerates to "never seal" — trivially equivalent — can't pass silently). The units
    /// consumer is asserted alongside on the same instance (the mixed display+summary case, the
    /// two consumers sharing one seam) *and* on a dedicated units-only instance (the store's
    /// summary-pass case, where the display fold is never requested).
    @discardableResult
    func assertEquivalent(
        _ events: [AgentEvent], worktreeRoot: String? = IncrementalTranscriptProjectionTests.root,
        expectSealing: Bool = true, sourceLocation: SourceLocation = #_sourceLocation
    ) -> IncrementalTranscriptProjection {
        let incremental = IncrementalTranscriptProjection()
        let unitsOnly = IncrementalTranscriptProjection()
        for p in 0...events.count {
            let prefix = Array(events.prefix(p))
            let full = TranscriptProjection.items(prefix, worktreeRoot: worktreeRoot)
            let live = incremental.update(events: prefix, worktreeRoot: worktreeRoot)
            if live.items != full.items {
                Issue.record(
                    "items diverged at prefix \(p)/\(events.count): incremental \(live.items.count) items, full \(full.items.count)",
                    sourceLocation: sourceLocation)
                return incremental
            }
            if live.lockLines != full.lockLines {
                Issue.record(
                    "lockLines diverged at prefix \(p)/\(events.count): incremental \(live.lockLines), full \(full.lockLines)",
                    sourceLocation: sourceLocation)
                return incremental
            }
            let fullUnits = TranscriptProjection.summarizableUnits(prefix)
            if !Self.unitsEqual(incremental.summarizableUnits(events: prefix), fullUnits) {
                Issue.record(
                    "units (shared instance) diverged at prefix \(p)/\(events.count)",
                    sourceLocation: sourceLocation)
                return incremental
            }
            if !Self.unitsEqual(unitsOnly.summarizableUnits(events: prefix), fullUnits) {
                Issue.record(
                    "units (units-only instance) diverged at prefix \(p)/\(events.count)",
                    sourceLocation: sourceLocation)
                return incremental
            }
        }
        if expectSealing {
            #expect(incremental.sealedEventCountForTesting > 0,
                    "watermark never advanced — no incremental win", sourceLocation: sourceLocation)
            #expect(unitsOnly.sealedEventCountForTesting > 0,
                    "units-only instance never sealed", sourceLocation: sourceLocation)
        }
        return incremental
    }

    // MARK: - Scenarios

    @Test func multiTurnStreaming() {
        let s = Stream()
        s.started()
        for t in 0..<4 {
            s.user("u\(t)", "do thing \(t)")
            s.thinkingDelta("m\(t)", "hmm ")
            s.thinkingDelta("m\(t)", "okay")
            s.thinkingFinal("m\(t)", "hmm okay")
            s.textDelta("m\(t)", "work")
            s.textDelta("m\(t)", "ing…")
            s.toolStart("t\(t)a", "Bash", .object(["command": .string("ls")]))
            s.toolComplete("t\(t)a", "Bash", .object(["command": .string("ls")]))
            s.result("t\(t)a", .string("files"))
            s.textFinal("m\(t)", "working…")
            s.usage()
            s.turn()
        }
        s.finished()
        let projector = assertEquivalent(s.events)
        // The frozen turns must actually be sealed — the live tail is the current turn, not
        // the whole session.
        #expect(projector.sealedEventCountForTesting > s.events.count / 2)
    }

    @Test func toolRunsAndEmptyThinking() {
        let s = Stream()
        s.started()
        s.user("u0", "go")
        // A run of tools with a redacted (empty) thinking block in the middle: held rows must
        // not split the run, and the seam must not land inside it.
        s.textFinal("m0", "let me look")
        s.toolStart("t1", "Read"); s.toolComplete("t1", "Read"); s.result("t1")
        s.thinkingFinal("empty", "")
        s.toolStart("t2", "Grep"); s.toolComplete("t2", "Grep"); s.result("t2")
        s.toolStart("t3", "Read"); s.toolComplete("t3", "Read"); s.result("t3")
        s.textFinal("m1", "found it")
        s.turn()
        s.user("u1", "next")
        s.textFinal("m2", "on it")
        s.turn()
        assertEquivalent(s.events)
    }

    @Test func inputDeltaBursts() {
        let s = Stream()
        s.started()
        s.user("u0", "write a lot")
        s.textFinal("m0", "streaming a big input")
        s.toolStart("t1", "Write")
        for i in 0..<25 { s.inputDelta("t1", "chunk \(i) ") }
        s.toolComplete("t1", "Write", .object(["file_path": .string("\(Self.root)/big.txt"),
                                               "content": .string("all chunks")]))
        s.result("t1")
        s.fileChange("t1", "\(Self.root)/big.txt")
        s.textFinal("m1", "written")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "ok")
        s.turn()
        assertEquivalent(s.events)
    }

    @Test func subagentNestingWithGrandchild() {
        let s = Stream()
        s.started()
        s.user("u0", "fan out")
        s.textFinal("m0", "spawning")
        s.toolStart("task1", "Task", .object(["prompt": .string("explore")]))
        s.toolComplete("task1", "Task", .object(["prompt": .string("explore")]))
        // Child activity interleaves with the parent's own stream.
        s.thinkingFinal("c1", "child thinks", parent: "task1")
        s.toolStart("ct1", "Read", .object(["file_path": .string("\(Self.root)/a.txt")]), parent: "task1")
        s.toolComplete("ct1", "Read", .object(["file_path": .string("\(Self.root)/a.txt")]), parent: "task1")
        s.result("ct1")
        // Grandchild: the child spawns its own Task.
        s.toolStart("task2", "Task", parent: "task1")
        s.toolComplete("task2", "Task", parent: "task1")
        s.textFinal("g1", "grandchild prose", parent: "task2")
        s.result("task2", .string("inner done"))
        s.textFinal("c2", "child wraps up", parent: "task1")
        s.result("task1", .string("done"))
        s.textFinal("m1", "all merged")
        s.turn()
        s.user("u1", "again")
        s.textFinal("m2", "ok")
        s.turn()
        assertEquivalent(s.events)
    }

    @Test func fanOutBlockOfTasks() {
        let s = Stream()
        s.started()
        s.user("u0", "parallel")
        s.textFinal("m0", "spawning two")
        s.toolStart("A", "Task"); s.toolComplete("A", "Task")
        s.toolStart("B", "Task"); s.toolComplete("B", "Task")
        s.textFinal("a1", "agent A working", parent: "A")
        s.textFinal("b1", "agent B working", parent: "B")
        s.result("A", .string("a done"))
        s.result("B", .string("b done"))
        s.textFinal("m1", "both returned")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "next")
        s.turn()
        assertEquivalent(s.events)
    }

    /// The subtle run-split case: two Task spawns that are *adjacent at the top level* but
    /// separated in the raw stream by the first Task's child activity and result. Child prose
    /// must not count as a top-level separator, or the seam could split the fan-out block.
    @Test func adjacentSpawnsSeparatedByChildActivity() {
        let s = Stream()
        s.started()
        s.user("u0", "two waves")
        s.textFinal("m0", "first spawn")
        s.toolStart("task1", "Task"); s.toolComplete("task1", "Task")
        s.textFinal("c1", "child prose only — no trailing child tool", parent: "task1")
        s.result("task1", .string("done"))
        // Top level now ends with the task1 card; the next spawn must coalesce with it.
        s.toolStart("task2", "Task"); s.toolComplete("task2", "Task")
        s.textFinal("c2", "second child", parent: "task2")
        s.result("task2", .string("done"))
        s.textFinal("m1", "both waves done")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "ok")
        s.turn()
        assertEquivalent(s.events)
    }

    @Test func lockNotesFoldWithinTurn() {
        let s = Stream()
        s.started()
        s.user("u0", "edit please")
        s.textFinal("m0", "editing")
        s.note("Waiting for a lock on src/a.swift…", icon: "lock", lockEvent: true,
               lock: NoteLock(state: .waiting, paths: ["src/a.swift"]))
        s.edit("e1", path: "\(Self.root)/src/a.swift")
        s.note("Lock acquired on src/a.swift.", icon: "lock.fill", lockEvent: true,
               lock: NoteLock(state: .acquired, paths: ["src/a.swift"]))
        s.note("Lock released on src/a.swift.", icon: "lock.open", lockEvent: true,
               lock: NoteLock(state: .released, paths: ["src/a.swift"]))
        s.textFinal("m1", "done")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "next")
        s.turn()
        assertEquivalent(s.events)
    }

    /// A lock/nvrsion note *tagged* with a subagent's edit nests in that subagent's card instead
    /// of leaking into the main transcript; an untagged note stays at the top level. The
    /// incremental projector must agree across every prefix — a routed note stays on its
    /// subagent's side of every seal seam.
    @Test func taggedNotesNestInSubagentCard() {
        let s = Stream()
        s.started()
        s.user("u0", "delegate an edit")
        s.textFinal("m0", "spawning a subagent")
        s.toolStart("A", "Task"); s.toolComplete("A", "Task")
        s.textFinal("a0", "on it", parent: "A")
        s.edit("e1", path: "\(Self.root)/src/a.swift", parent: "A")
        s.note("Lock acquired on src/a.swift.", icon: "lock.fill", lockEvent: true,
               lock: NoteLock(state: .acquired, paths: ["src/a.swift"]), toolCallID: "e1")
        s.note("nvrsion: landed src/a.swift · abc1234", icon: "checkmark.circle", toolCallID: "e1")
        s.result("A", .string("edited"))
        s.textFinal("m1", "subagent done")
        s.turn()
        // The turn-end release is untagged: it lands after A returned, so it stays top level.
        s.note("Lock released on src/a.swift.", icon: "lock.open", lockEvent: true,
               lock: NoteLock(state: .released, paths: ["src/a.swift"]))
        s.textFinal("m2", "landed")
        s.turn()
        assertEquivalent(s.events)
    }

    /// The hard case the `PriorEdit` registry exists for: a lock's `released` note lands when the
    /// file lands in the parent — several turns after the edit card it must fold onto is sealed.
    @Test func lockReleaseManyTurnsAfterEdit() {
        let s = Stream()
        s.started()
        s.user("u0", "edit")
        s.textFinal("m0", "editing")
        s.edit("e1", path: "\(Self.root)/src/deep/file.swift")
        s.note("Lock acquired on src/deep/file.swift.", lockEvent: true,
               lock: NoteLock(state: .acquired, paths: ["src/deep/file.swift"]))
        s.textFinal("m1", "edited")
        s.turn()
        // Two full unrelated turns so the edit is deeply sealed.
        for t in 1...2 {
            s.user("u\(t)", "other work")
            s.textFinal("mx\(t)", "sure")
            s.toolStart("t\(t)", "Bash"); s.toolComplete("t\(t)", "Bash"); s.result("t\(t)")
            s.textFinal("my\(t)", "done")
            s.turn()
        }
        // The landing releases the lock — the note must fold under the sealed edit's id.
        s.note("Lock released on src/deep/file.swift.", lockEvent: true,
               lock: NoteLock(state: .released, paths: ["src/deep/file.swift"]))
        s.textFinal("m9", "shipped")
        s.turn()
        s.user("u9", "tail")
        s.textFinal("m10", "yep")
        s.turn()
        let projector = assertEquivalent(s.events)
        // Make the hard case explicit: the release actually folded onto the sealed edit.
        let (_, lockLines) = projector.update(events: s.events, worktreeRoot: Self.root)
        #expect(lockLines["e1"]?.map(\.state) == [.acquired, .released])
    }

    @Test func multiPathAndUnmatchedLockNotes() {
        let s = Stream()
        s.started()
        s.user("u0", "edit two files")
        s.textFinal("m0", "editing both")
        s.edit("e1", path: "\(Self.root)/src/a.swift")
        s.edit("e2", path: "\(Self.root)/src/b.swift")
        // Multi-path note routes each path to its own card.
        s.note("Lock acquired on src/a.swift, src/b.swift.", lockEvent: true,
               lock: NoteLock(state: .acquired, paths: ["src/a.swift", "src/b.swift"]))
        // Unmatched path → the whole note stays a standalone row.
        s.note("Waiting for a lock on src/other.swift…", lockEvent: true,
               lock: NoteLock(state: .waiting, paths: ["src/other.swift"]))
        // Lock-flagged note with no structured paths (re-ground) → standalone, never folded.
        s.note("Re-grounding on src/a.swift: the parent advanced.", lockEvent: true)
        s.textFinal("m1", "done")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "next")
        s.turn()
        // Cross-turn: a later multi-path release folding onto two *sealed* cards.
        s.note("Lock released on src/a.swift, src/b.swift.", lockEvent: true,
               lock: NoteLock(state: .released, paths: ["src/a.swift", "src/b.swift"]))
        s.textFinal("m3", "landed")
        s.turn()
        assertEquivalent(s.events)
    }

    /// Claude's stream shape: partial deltas for a message, then tool items, then the
    /// whole-message decode re-emits the same messageID's authoritative non-partial text — a
    /// message must not seal while its final chunks can still arrive.
    @Test func messageFinalArrivesAfterToolItems() {
        let s = Stream()
        s.started()
        s.user("u0", "go")
        s.textDelta("m0", "let me ")
        s.textDelta("m0", "check")
        s.toolStart("t1", "Read")
        s.textFinal("m0", "let me check — rephrased")  // authoritative replacement
        s.toolComplete("t1", "Read")
        s.result("t1")
        s.textFinal("m1", "found")
        s.turn()
        s.user("u1", "more")
        s.textFinal("m2", "ok")
        s.turn()
        assertEquivalent(s.events)
    }

    @Test func interruptedRunLeavesToolUnfinished() {
        let s = Stream()
        s.started()
        s.user("u0", "run forever")
        s.textFinal("m0", "starting")
        s.toolStart("t1", "Bash", .object(["command": .string("sleep 1000")]))
        s.toolComplete("t1", "Bash", .object(["command": .string("sleep 1000")]))
        // No result — the user interrupts.
        s.finished(.interrupted)
        s.turn()
        s.user("u1", "ok stop")
        s.textFinal("m1", "stopped")
        s.turn()
        assertEquivalent(s.events)
    }

    @Test func rawApprovalErrorAndRateLimitRows() {
        let s = Stream()
        s.started()
        s.user("u0", "go")
        s.raw("unknown native payload")
        s.approval("Bash")
        s.add(.approvalResolved(ApprovalResolved(
            id: ApprovalID(rawValue: "appr-1"), decision: .allow(),
            decidedBy: "user", decidedAt: Date(timeIntervalSince1970: 1_750_000_000))))
        s.error("transient API error")
        s.add(.rateLimit(RateLimit(status: "allowed_warning")))
        s.textFinal("m0", "done anyway")
        s.usage()
        s.turn()
        s.user("u1", "…")
        s.textFinal("m1", "ok")
        s.turn()
        assertEquivalent(s.events)
    }

    /// Claude's `system/thinking_tokens` pings coalesce into one live meter that stays open
    /// across interleaved thinking deltas — the seam must never split a ping run (a split would
    /// mint a second meter row with a different id).
    @Test func thinkingTokensMeterRuns() {
        let s = Stream()
        s.started()
        s.user("u0", "think hard")
        s.thinkingTokens(100)
        s.thinkingDelta("m0", "pondering ")
        s.thinkingTokens(250)
        s.thinkingDelta("m0", "still pondering")
        s.thinkingTokens(400)
        s.thinkingFinal("m0", "pondering still pondering")
        s.textFinal("m1", "answer")
        s.turn()
        // A second run in a later turn — with a total *below* the first run's (a fresh session),
        // and a tool call closing it mid-turn.
        s.user("u1", "again")
        s.thinkingTokens(50)
        s.thinkingTokens(120)
        s.toolStart("t1", "Bash"); s.toolComplete("t1", "Bash"); s.result("t1")
        s.thinkingTokens(60)
        s.textFinal("m2", "done")
        s.turn()
        s.user("u2", "…")
        s.textFinal("m3", "ok")
        s.turn()
        assertEquivalent(s.events)
    }

    /// A VM-control call the backend also narrates as a pretty note: the generic tool card is
    /// dropped in favor of the friendly line, and the seam must never leave the card sealed on
    /// one side with its narration note on the other deciding differently.
    @Test func narratedToolCardDrops() {
        let s = Stream()
        s.started()
        s.user("u0", "shut the vm down")
        s.textFinal("m0", "shutting down")
        s.toolStart("vm1", MCPApprovalServer.qualifiedMacVMControlToolName,
                    .object(["op": .string("shutdown")]))
        s.toolComplete("vm1", MCPApprovalServer.qualifiedMacVMControlToolName,
                       .object(["op": .string("shutdown")]))
        s.result("vm1")
        s.note("Shutting down the macOS VM…", icon: "power", narratesTool: true)
        s.textFinal("m1", "done")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "ok")
        s.turn()
        assertEquivalent(s.events)
    }

    /// One long autonomous turn: no turn boundary for hundreds of events. Intra-turn sealing
    /// must still advance the watermark (messages close as the assistant moves on).
    @Test func intraTurnSealingInOneLongTurn() {
        let s = Stream()
        s.started()
        s.user("u0", "autonomous run")
        for i in 0..<30 {
            s.thinkingFinal("th\(i)", "step \(i)")
            s.toolStart("t\(i)", "Bash", .object(["command": .string("step \(i)")]))
            s.toolComplete("t\(i)", "Bash", .object(["command": .string("step \(i)")]))
            s.result("t\(i)")
            s.textFinal("m\(i)", "step \(i) done")
        }
        // Still mid-turn — no turnCompleted yet.
        let projector = assertEquivalent(s.events)
        #expect(projector.sealedEventCountForTesting > s.events.count / 2,
                "intra-turn sealing should keep the live tail small inside one long turn")
    }

    @Test func streamSwitchResets() {
        let a = Stream()
        a.started()
        a.user("u0", "stream A")
        a.textFinal("m0", "hello from A")
        a.turn()
        a.user("u1", "…")
        a.textFinal("m1", "more A")
        a.turn()

        let b = Stream(startSeq: 5000)  // different firstSeq → new identity
        b.started(cwd: "/other/wt")
        b.user("v0", "stream B")
        b.textFinal("n0", "hello from B")
        b.turn()

        let incremental = IncrementalTranscriptProjection()
        for p in 0...a.events.count {
            let prefix = Array(a.events.prefix(p))
            let full = TranscriptProjection.items(prefix, worktreeRoot: Self.root)
            let live = incremental.update(events: prefix, worktreeRoot: Self.root)
            #expect(live.items == full.items && live.lockLines == full.lockLines)
        }
        // Switch sessions on the same instance (the open reset: cleared, then refilled).
        #expect(incremental.update(events: [], worktreeRoot: "/other/wt").items.isEmpty)
        for p in 0...b.events.count {
            let prefix = Array(b.events.prefix(p))
            let full = TranscriptProjection.items(prefix, worktreeRoot: "/other/wt")
            let live = incremental.update(events: prefix, worktreeRoot: "/other/wt")
            #expect(live.items == full.items && live.lockLines == full.lockLines)
        }
    }

    /// A revert (or transcript clear) shrinks history under the watermark: the projector must
    /// notice and refold rather than serve stale sealed items.
    @Test func historyShrinkResets() {
        let s = Stream()
        s.started()
        for t in 0..<3 {
            s.user("u\(t)", "turn \(t)")
            s.textFinal("m\(t)", "reply \(t)")
            s.turn()
        }
        let incremental = IncrementalTranscriptProjection()
        _ = incremental.update(events: s.events, worktreeRoot: Self.root)
        #expect(incremental.sealedEventCountForTesting > 0)

        // Revert to the first turn: fewer events than are sealed.
        let reverted = Array(s.events.prefix(4))
        let live = incremental.update(events: reverted, worktreeRoot: Self.root)
        let full = TranscriptProjection.items(reverted, worktreeRoot: Self.root)
        #expect(live.items == full.items && live.lockLines == full.lockLines)
    }

    /// A backfill merge can rewrite history *under* the watermark (events slot in by seq). The
    /// projector must notice the shifted prefix and reset rather than serve stale sealed items.
    @Test func backfillUnderWatermarkResets() {
        let s = Stream()
        s.started()
        s.user("u2", "late half")   // seqs 1, 2, …
        s.textFinal("m2", "tail history")
        s.turn()
        s.user("u3", "current")
        s.textFinal("m3", "working")
        s.turn()

        let incremental = IncrementalTranscriptProjection()
        _ = incremental.update(events: s.events, worktreeRoot: Self.root)

        // A full-history backfill arrives: older events (lower seq) merge in above the tail.
        let older = Stream(startSeq: 2000)
        older.user("u1", "ancient history")
        older.textFinal("m1", "long ago")
        older.turn()
        let merged = (older.events + s.events.map { $0.withSeq($0.seq + 3000) })
            .sorted { $0.seq < $1.seq }
        let live = incremental.update(events: merged, worktreeRoot: Self.root)
        let full = TranscriptProjection.items(merged, worktreeRoot: Self.root)
        #expect(live.items == full.items && live.lockLines == full.lockLines)
    }

    /// The worktree root is a fold input (lock-path normalization): changing it must refold the
    /// sealed prefix, not keep folds normalized against the old root.
    @Test func worktreeRootChangeRefolds() {
        let s = Stream()
        s.started()
        s.user("u0", "edit")
        s.textFinal("m0", "editing")
        s.edit("e1", path: "\(Self.root)/src/a.swift")
        s.note("Lock acquired on src/a.swift.", lockEvent: true,
               lock: NoteLock(state: .acquired, paths: ["src/a.swift"]))
        s.textFinal("m1", "done")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "next")
        s.turn()

        let incremental = IncrementalTranscriptProjection()
        _ = incremental.update(events: s.events, worktreeRoot: Self.root)
        // Under a different root the edit path no longer strips, so the note must not fold.
        for root in [nil, "/elsewhere"] as [String?] {
            let live = incremental.update(events: s.events, worktreeRoot: root)
            let full = TranscriptProjection.items(s.events, worktreeRoot: root)
            #expect(live.items == full.items && live.lockLines == full.lockLines)
        }
    }

    /// The O(scope²) case the intra-scope sealing exists for: one `Task` runs for hundreds of
    /// events with no result. The outer watermark must stay pinned before the spawn (the card's
    /// children still grow) while (a) the scope's own projector seals most of the closed child
    /// activity, and (b) the pinned fast path skips the full watermark rescan for almost every
    /// child-only prefix.
    @Test func longOpenSubagentRunSealsInsideScope() {
        let s = Stream()
        s.started()
        s.user("u0", "long fan-out")
        s.textFinal("m0", "spawning")
        s.textFinal("m1", "kicking off the agent")  // closes m0; m1 stays the open pin
        s.toolStart("task1", "Task", .object(["prompt": .string("dig in")]))
        s.toolComplete("task1", "Task", .object(["prompt": .string("dig in")]))
        let spawnCount = s.events.count
        for i in 0..<60 {
            s.thinkingFinal("cth\(i)", "child step \(i)", parent: "task1")
            s.toolStart("ct\(i)", "Bash", .object(["command": .string("step \(i)")]), parent: "task1")
            s.toolComplete("ct\(i)", "Bash", .object(["command": .string("step \(i)")]), parent: "task1")
            s.result("ct\(i)")
            s.textFinal("cm\(i)", "child note \(i)", parent: "task1")
        }
        // Still no result for task1 — the run is live.
        let projector = assertEquivalent(s.events, expectSealing: true)
        #expect(projector.sealedEventCountForTesting < spawnCount,
                "the outer seam must not pass an open Task's spawn")
        let scopeSeals = projector.scopeSealedCountsForTesting()
        let scopeEventCount = s.events.count - spawnCount
        #expect((scopeSeals["task1"] ?? 0) > scopeEventCount / 2,
                "the open scope's closed children should seal: \(scopeSeals)")
        // ~5 events per prefix step × 2 reads (items + units) would be ~600 full scans without
        // the pinned fast path; with it, full scans stop shortly after the spawn pins the seam.
        #expect(projector.fullWatermarkScansForTesting < 60,
                "pinned fast path should skip full rescans, ran \(projector.fullWatermarkScansForTesting)")
    }

    /// A fan-out where two Tasks return and a third keeps streaming: the closed siblings sit in
    /// the unsealed tail (the open straggler pins the seam before the whole run of spawn cards),
    /// so their scopes must fold through their own projectors rather than from scratch per read.
    @Test func fanOutStragglerKeepsClosedSiblingsCheap() {
        let s = Stream()
        s.started()
        s.user("u0", "three ways")
        s.textFinal("m0", "spawning three")
        s.textFinal("m1", "here we go")
        for t in ["A", "B", "C"] {
            s.toolStart(t, "Task"); s.toolComplete(t, "Task")
        }
        for t in ["A", "B"] {
            s.textFinal("\(t)m0", "agent \(t) working", parent: t)
            s.toolStart("\(t)t0", "Read", parent: t)
            s.toolComplete("\(t)t0", "Read", parent: t)
            s.result("\(t)t0")
            s.textFinal("\(t)m1", "agent \(t) done", parent: t)
            s.result(t, .string("\(t) done"))
        }
        for i in 0..<25 {
            s.toolStart("Ct\(i)", "Grep", parent: "C")
            s.toolComplete("Ct\(i)", "Grep", parent: "C")
            s.result("Ct\(i)")
            s.textFinal("Cm\(i)", "C step \(i)", parent: "C")
        }
        assertEquivalent(s.events)
    }

    /// A nested scope that keeps streaming: the child Task spawns a grandchild that runs long.
    /// Sealing must advance *inside* the child scope up to the open grandchild's spawn, and the
    /// grandchild's own projector must seal its closed activity — recursion of the same rule.
    @Test func openGrandchildScopeSealsRecursively() {
        let s = Stream()
        s.started()
        s.user("u0", "nest deep")
        s.textFinal("m0", "spawning")
        s.textFinal("m1", "go")
        s.toolStart("task1", "Task"); s.toolComplete("task1", "Task")
        s.textFinal("c0", "child kicking off a helper", parent: "task1")
        s.textFinal("c1", "spawning helper now", parent: "task1")
        s.toolStart("task2", "Task", parent: "task1")
        s.toolComplete("task2", "Task", parent: "task1")
        for i in 0..<30 {
            s.toolStart("gt\(i)", "Read", parent: "task2")
            s.toolComplete("gt\(i)", "Read", parent: "task2")
            s.result("gt\(i)")
            s.textFinal("gm\(i)", "grandchild step \(i)", parent: "task2")
        }
        let projector = assertEquivalent(s.events)
        let scopeSeals = projector.scopeSealedCountsForTesting()
        #expect((scopeSeals["task2"] ?? 0) > 30, "the open grandchild scope should seal internally: \(scopeSeals)")
    }

    /// A closure-premise violation *inside* a scope: a grandchild Task returns, its card seals
    /// into the child scope's prefix (children baked, cursors snapshotted) — and then a late
    /// event arrives for the supposedly-finished grandchild. The cursor check must catch it and
    /// the scope must refold, keeping every prefix byte-identical to the full fold.
    @Test func lateEventIntoBakedScopeSelfHeals() {
        let s = Stream()
        s.started()
        s.user("u0", "nested")
        s.textFinal("m0", "spawning")
        s.textFinal("m1", "go")
        s.toolStart("task1", "Task"); s.toolComplete("task1", "Task")
        s.toolStart("task2", "Task", parent: "task1")
        s.toolComplete("task2", "Task", parent: "task1")
        s.textFinal("g0", "grandchild prose", parent: "task2")
        s.result("task2", .string("inner done"))
        // Plenty of closed child activity so task2's card seals inside task1's scope.
        for i in 0..<20 {
            s.toolStart("ct\(i)", "Bash", parent: "task1")
            s.toolComplete("ct\(i)", "Bash", parent: "task1")
            s.result("ct\(i)")
            s.textFinal("cm\(i)", "child step \(i)", parent: "task1")
        }
        // The violation: task2 was result-fenced, yet another grandchild event lands.
        s.textFinal("gLate", "late grandchild straggler", parent: "task2")
        for i in 20..<28 {
            s.toolStart("ct\(i)", "Bash", parent: "task1")
            s.toolComplete("ct\(i)", "Bash", parent: "task1")
            s.result("ct\(i)")
            s.textFinal("cm\(i)", "child step \(i)", parent: "task1")
        }
        assertEquivalent(s.events)
    }

    /// The store's pattern: sealing advances under display reads alone, then the summary pass
    /// asks for units for the first time mid-session — the units consumer must prime itself
    /// from the already-sealed prefix (fold the gap once) and stay equivalent thereafter.
    @Test func unitsConsumerPrimesLate() {
        let s = Stream()
        s.started()
        for t in 0..<4 {
            s.user("u\(t)", "turn \(t)")
            s.textFinal("m\(t)a", "working")
            s.toolStart("t\(t)", "Bash", .object(["command": .string("cmd \(t)")]))
            s.toolComplete("t\(t)", "Bash", .object(["command": .string("cmd \(t)")]))
            s.result("t\(t)")
            s.textFinal("m\(t)b", "done")
            s.turn()
        }
        let incremental = IncrementalTranscriptProjection()
        let half = s.events.count / 2
        for p in 0...half {
            _ = incremental.update(events: Array(s.events.prefix(p)), worktreeRoot: Self.root)
        }
        #expect(incremental.sealedEventCountForTesting > 0)
        for p in half...s.events.count {
            let prefix = Array(s.events.prefix(p))
            _ = incremental.update(events: prefix, worktreeRoot: Self.root)
            #expect(Self.unitsEqual(incremental.summarizableUnits(events: prefix),
                                    TranscriptProjection.summarizableUnits(prefix)))
        }
    }

    /// A root change refolds the display consumer but must not disturb the units consumer (the
    /// units fold has no root input) — both stay equivalent through the flip, on one instance.
    @Test func rootChangeLeavesUnitsIntact() {
        let s = Stream()
        s.started()
        s.user("u0", "edit")
        s.textFinal("m0", "editing")
        s.edit("e1", path: "\(Self.root)/src/a.swift")
        s.note("Lock acquired on src/a.swift.", lockEvent: true,
               lock: NoteLock(state: .acquired, paths: ["src/a.swift"]))
        s.textFinal("m1", "done")
        s.turn()
        s.user("u1", "…")
        s.textFinal("m2", "next")
        s.turn()

        let incremental = IncrementalTranscriptProjection()
        _ = incremental.update(events: s.events, worktreeRoot: Self.root)
        #expect(Self.unitsEqual(incremental.summarizableUnits(events: s.events),
                                TranscriptProjection.summarizableUnits(s.events)))
        for root in [nil, "/elsewhere", Self.root] as [String?] {
            let live = incremental.update(events: s.events, worktreeRoot: root)
            let full = TranscriptProjection.items(s.events, worktreeRoot: root)
            #expect(live.items == full.items && live.lockLines == full.lockLines)
            #expect(Self.unitsEqual(incremental.summarizableUnits(events: s.events),
                                    TranscriptProjection.summarizableUnits(s.events)))
        }
    }

    // MARK: - Fuzz

    /// Deterministic fuzz: compose random pattern blocks (turns, tool runs, subagents, locks,
    /// interrupts, empty thinking, thinking-tokens pings, input-delta bursts) and assert the
    /// property on every prefix.
    @Test(arguments: [UInt64(1), 7, 42, 1234, 99999])
    func fuzzedStreams(seed: UInt64) {
        var rng = ProjectionFuzzRNG(seed: seed)
        let s = Stream()
        s.started()
        var msg = 0, tool = 0, editPaths: [String] = []

        for turnIndex in 0..<Int(rng.next(in: 3...6)) {
            s.user("u\(turnIndex)", "turn \(turnIndex)")
            for _ in 0..<Int(rng.next(in: 1...6)) {
                switch rng.next(in: 0...10) {
                case 0...2:  // streamed message
                    let id = "m\(msg)"; msg += 1
                    for _ in 0..<Int(rng.next(in: 1...3)) { s.textDelta(id, "word ") }
                    if rng.next(in: 0...1) == 0 { s.textFinal(id, "final text \(id)") }
                case 3:      // thinking (sometimes redacted-empty), sometimes with token pings
                    let id = "th\(msg)"; msg += 1
                    if rng.next(in: 0...3) == 0 { s.thinkingFinal(id, "") }
                    else {
                        if rng.next(in: 0...1) == 0 { s.thinkingTokens(Int(rng.next(in: 10...500))) }
                        s.thinkingDelta(id, "hmm")
                        if rng.next(in: 0...1) == 0 { s.thinkingTokens(Int(rng.next(in: 10...900))) }
                        s.thinkingFinal(id, "hmm indeed")
                    }
                case 4...5:  // tool run, occasionally with an input-delta burst
                    for _ in 0..<Int(rng.next(in: 1...3)) {
                        let id = "t\(tool)"; tool += 1
                        s.toolStart(id, "Bash", .object(["command": .string("cmd \(id)")]))
                        for _ in 0..<Int(rng.next(in: 0...3)) { s.inputDelta(id, "{\"partial\":") }
                        s.toolComplete(id, "Bash", .object(["command": .string("cmd \(id)")]))
                        if rng.next(in: 0...5) > 0 { s.result(id) }  // occasionally unfinished
                    }
                case 6:      // edit + lock chatter
                    let path = "src/f\(tool).swift"
                    let id = "e\(tool)"; tool += 1
                    if rng.next(in: 0...1) == 0 {
                        s.note("Waiting for a lock on \(path)…", lockEvent: true,
                               lock: NoteLock(state: .waiting, paths: [path]))
                    }
                    s.edit(id, path: "\(Self.root)/\(path)")
                    s.note("Lock acquired on \(path).", lockEvent: true,
                           lock: NoteLock(state: .acquired, paths: [path]))
                    editPaths.append(path)
                case 7:      // subagent — occasionally a longer run left open (interrupted, or
                             // still streaming when the turn is cut), so seams and scope sealers
                             // see open scopes too
                    let id = "task\(tool)"; tool += 1
                    s.toolStart(id, "Task"); s.toolComplete(id, "Task")
                    s.textFinal("c\(msg)", "child prose", parent: id); msg += 1
                    for _ in 0..<Int(rng.next(in: 1...5)) {
                        let child = "ct\(tool)"; tool += 1
                        s.toolStart(child, "Read", parent: id)
                        s.toolComplete(child, "Read", parent: id)
                        if rng.next(in: 0...5) > 0 { s.result(child) }
                        s.textFinal("c\(msg)", "child update", parent: id); msg += 1
                    }
                    if rng.next(in: 0...3) > 0 { s.result(id, .string("subagent done")) }
                case 8:      // a long-held lock releases against some sealed edit
                    if let path = editPaths.randomElement(using: &rng.generator) {
                        s.note("Lock released on \(path).", lockEvent: true,
                               lock: NoteLock(state: .released, paths: [path]))
                    }
                case 9:      // approval / error rows
                    if rng.next(in: 0...1) == 0 { s.approval("Bash") } else { s.error("api hiccup") }
                default:     // ambient rows
                    if rng.next(in: 0...1) == 0 { s.usage() } else { s.note("nvrsion: landed edit.") }
                }
            }
            if rng.next(in: 0...7) == 0 { s.finished(.interrupted) }
            s.turn()
        }
        assertEquivalent(s.events, expectSealing: true)
    }
}

/// Small deterministic RNG so fuzz failures reproduce exactly (mirrors the iOS suite's
/// SplitMix64, renamed to avoid a collision if the two suites ever share a target).
struct ProjectionFuzzRNG {
    struct Generator: RandomNumberGenerator {
        var state: UInt64
        mutating func next() -> UInt64 {
            state &+= 0x9E37_79B9_7F4A_7C15
            var z = state
            z = (z ^ (z >> 30)) &* 0xBF58_476D_1CE4_E5B9
            z = (z ^ (z >> 27)) &* 0x94D0_49BB_1331_11EB
            return z ^ (z >> 31)
        }
    }
    var generator: Generator
    init(seed: UInt64) { generator = Generator(state: seed) }
    mutating func next(in range: ClosedRange<UInt64>) -> UInt64 {
        UInt64.random(in: range, using: &generator)
    }
}
