import Foundation
import Testing
import NucleicProtocol

@testable import NucleicCore

private func mirrorEvent(_ seq: UInt64) -> AgentEvent {
    AgentEvent(
        sessionID: SessionID(rawValue: "s1"), seq: seq, at: Date(timeIntervalSince1970: 1_750_000_000),
        backend: .claudeCode, nativeType: nil,
        kind: .assistantText(TextChunk(messageID: "m\(seq)", text: "hi \(seq)", isPartial: false)))
}

private func mirrorHeader(epoch: UInt64? = nil) -> SessionHeader {
    SessionHeader(
        sessionID: SessionID(rawValue: "s1"), backend: .claudeCode,
        worktree: "/tmp/wt", createdAt: Date(timeIntervalSince1970: 1_750_000_000),
        revertEpoch: epoch)
}

/// A post-revert regrowth event — same seq range a pre-revert `mirrorEvent` used, different
/// content, so tests can assert which generation a mirror ended up holding.
private func regrownEvent(_ seq: UInt64) -> AgentEvent {
    AgentEvent(
        sessionID: SessionID(rawValue: "s1"), seq: seq, at: Date(timeIntervalSince1970: 1_750_000_100),
        backend: .claudeCode, nativeType: nil,
        kind: .assistantText(TextChunk(messageID: "r\(seq)", text: "regrown \(seq)", isPartial: false)))
}

private func tempRoot() -> URL {
    FileManager.default.temporaryDirectory.appendingPathComponent("nucleic-mirror-\(UUID().uuidString)")
}

@Suite struct TranscriptMirrorStoreTests {
    @Test func writeFullPreservesOriginalSeqs() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")

        let held = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(),
            events: (1...5).map { mirrorEvent(UInt64($0)) })
        #expect(held == 5)
        #expect(await store.heldSeq(sid) == 5)
        #expect(await store.sourceDeviceID(sid) == "devA")

        // The mirror must be a verbatim copy — seqs NOT re-stamped by the writer.
        let url = try #require(await store.url(sid))
        let read = try TranscriptReader(url: url).read()
        #expect(read.events.map(\.seq) == [1, 2, 3, 4, 5])
        #expect(read.header.sessionID == sid)
    }

    @Test func appendExtendsContiguousAndStopsAtGap() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(),
            events: (1...3).map { mirrorEvent(UInt64($0)) })

        // 4,5 are contiguous; 7,8 are past a gap and must be dropped (not committed as complete).
        let held = try await store.append(sid, events: [4, 5, 7, 8].map { mirrorEvent($0) })
        #expect(held == 5)
        #expect(await store.heldSeq(sid) == 5)
        let url = try #require(await store.url(sid))
        #expect(try TranscriptReader(url: url).read().events.map(\.seq) == [1, 2, 3, 4, 5])
    }

    @Test func truncateDropsTailAndResetsHeld() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(),
            events: (1...5).map { mirrorEvent(UInt64($0)) })

        // A peer reverted the chat to a tip of 3 — everything past seq 3 must drop.
        let held = try await store.truncate(sid, throughSeq: 3)
        #expect(held == 3)
        #expect(await store.heldSeq(sid) == 3)
        let url = try #require(await store.url(sid))
        let read = try TranscriptReader(url: url).read()
        #expect(read.events.map(\.seq) == [1, 2, 3])
        #expect(read.header.sessionID == sid)  // header preserved

        // A fresh cold store recovers the truncated cursor from disk (no resurrected tail).
        #expect(await TranscriptMirrorStore(root: root).heldSeq(sid) == 3)
    }

    @Test func truncateToZeroKeepsOnlyHeader() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(),
            events: (1...4).map { mirrorEvent(UInt64($0)) })

        // Reverting the very first message clears every event; line 0 (the header) stays.
        let held = try await store.truncate(sid, throughSeq: 0)
        #expect(held == 0)
        #expect(await store.heldSeq(sid) == 0)
        let read = try TranscriptReader(url: #require(await store.url(sid))).read()
        #expect(read.events.isEmpty)
        #expect(read.header.sessionID == sid)
    }

    @Test func truncateAtOrAboveHeldIsNoOp() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(),
            events: (1...3).map { mirrorEvent(UInt64($0)) })

        // A tip at/above what we hold drops nothing (a stale/duplicate revert can't grow it back).
        #expect(try await store.truncate(sid, throughSeq: 3) == 3)
        #expect(try await store.truncate(sid, throughSeq: 9) == 3)
        #expect(await store.heldSeq(sid) == 3)
        #expect(try TranscriptReader(url: #require(await store.url(sid))).read().events.map(\.seq) == [1, 2, 3])
    }

    @Test func heldSeqRecoveredByScanningFreshStore() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let sid = SessionID(rawValue: "s1")
        _ = try await TranscriptMirrorStore(root: root).writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(),
            events: (1...7).map { mirrorEvent(UInt64($0)) })

        // A brand-new store (cold index) must recover the held cursor from disk.
        let fresh = TranscriptMirrorStore(root: root)
        #expect(await fresh.heldSeq(sid) == 7)
        #expect(await fresh.has(sid))
    }

    @Test func heldEpochRecordedAndRecoveredFromDisk() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(epoch: 2),
            events: (1...3).map { mirrorEvent(UInt64($0)) })
        #expect(await store.heldEpoch(sid) == 2)

        // A brand-new store (cold index) recovers the epoch from the mirror file's own header.
        #expect(await TranscriptMirrorStore(root: root).heldEpoch(sid) == 2)
    }

    @Test func heldEpochNilForPreEpochMirror() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(), events: [mirrorEvent(1)])
        #expect(await store.heldEpoch(sid) == nil)
    }

    @Test func truncateStampsRevertEpochIntoHeader() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(epoch: 1),
            events: (1...5).map { mirrorEvent(UInt64($0)) })

        // A live `.transcriptReverted` carries the owner's post-revert epoch — the truncate
        // stamps it so the next reconcile reads this converged mirror as current.
        let held = try await store.truncate(sid, throughSeq: 3, revertEpoch: 2)
        #expect(held == 3)
        #expect(await store.heldEpoch(sid) == 2)
        let read = try TranscriptReader(url: #require(await store.url(sid))).read()
        #expect(read.header.revertEpoch == 2)
        #expect(read.events.map(\.seq) == [1, 2, 3])

        // The stamp survives a cold rescan.
        #expect(await TranscriptMirrorStore(root: root).heldEpoch(sid) == 2)
    }

    @Test func truncateWithoutEpochLeavesHeaderEpochAlone() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(epoch: 1),
            events: (1...4).map { mirrorEvent(UInt64($0)) })

        // An old host's revert message carries no epoch — the truncate still applies, the
        // recorded epoch stays what the mirror was synced at.
        _ = try await store.truncate(sid, throughSeq: 2)
        #expect(await store.heldSeq(sid) == 2)
        #expect(await store.heldEpoch(sid) == 1)
    }

    @Test func truncateStampsEpochEvenWhenNothingToDrop() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "devA", header: mirrorHeader(epoch: 1),
            events: (1...3).map { mirrorEvent(UInt64($0)) })

        // Already at/below the reverted tip (e.g. we mirrored only part of the dropped range):
        // no events to drop, but the epoch must still advance or the next reconcile would heal
        // a copy that is in fact converged.
        let held = try await store.truncate(sid, throughSeq: 5, revertEpoch: 2)
        #expect(held == 3)
        #expect(await store.heldSeq(sid) == 3)
        #expect(await store.heldEpoch(sid) == 2)
    }

    @Test func removeSourceDropsThatDevicesMirrors() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        _ = try await store.writeFull(
            SessionID(rawValue: "s1"), sourceDeviceID: "devA", header: mirrorHeader(),
            events: [mirrorEvent(1)])
        #expect(await store.has(SessionID(rawValue: "s1")))

        await store.removeSource("devA")
        #expect(await store.has(SessionID(rawValue: "s1")) == false)
        #expect(await store.heldSeq(SessionID(rawValue: "s1")) == 0)
    }
}

@Suite struct TranscriptSyncCoordinatorTests {
    /// Records which (session, afterSeq, holder) fetches the coordinator issued.
    private actor CallLog {
        private(set) var calls: [(sessionID: SessionID, afterSeq: UInt64, holder: String)] = []
        func record(_ s: SessionID, _ a: UInt64, _ h: String) { calls.append((s, a, h)) }
        var count: Int { calls.count }
    }

    private actor FetchGate {
        private var released = false
        private var waiters: [CheckedContinuation<Void, Never>] = []
        func wait() async {
            if released { return }
            await withCheckedContinuation { waiters.append($0) }
        }
        func release() {
            released = true
            for waiter in waiters { waiter.resume() }
            waiters = []
        }
    }

    /// A fetch closure that serves a canned gap-fill from any holder in `goodHolders` and replies
    /// `notHeld` from everyone else, logging every call.
    private func makeFetch(log: CallLog, goodHolders: Set<String>, tip: UInt64)
        -> @Sendable (SessionID, UInt64, String) async -> AsyncStream<PeerClient.TranscriptFetchUpdate> {
        { sessionID, afterSeq, holder in
            await log.record(sessionID, afterSeq, holder)
            return AsyncStream { cont in
                if goodHolders.contains(holder), afterSeq < tip {
                    let events = ((afterSeq + 1)...tip).map { mirrorEvent($0) }
                    cont.yield(.chunk(TranscriptChunk(
                        requestID: "x", sessionID: sessionID, batchIndex: 0, events: events)))
                    cont.yield(.complete(TranscriptFetchComplete(
                        requestID: "x", sessionID: sessionID, throughSeq: tip,
                        headerJSON: afterSeq == 0 ? (try? mirrorHeader().encodedForWire()) : nil)))
                } else if goodHolders.contains(holder) {
                    cont.yield(.complete(TranscriptFetchComplete(
                        requestID: "x", sessionID: sessionID, throughSeq: afterSeq)))
                } else {
                    cont.yield(.unavailable(TranscriptUnavailable(
                        requestID: "x", sessionID: sessionID, reason: .notHeld)))
                }
                cont.finish()
            }
        }
    }

    private func settle(_ coordinator: TranscriptSyncCoordinator, _ sid: SessionID) async {
        var n = 0
        while await coordinator.isFetching(sid), n < 10_000 { await Task.yield(); n += 1 }
    }

    @Test func backfillsGapFromOwner() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store, fetch: makeFetch(log: log, goodHolders: ["owner"], tip: 5))
        let sid = SessionID(rawValue: "s1")

        await coordinator.sync(
            sessions: [.init(sessionID: sid, lastSeq: 5, archived: false, ownerDeviceID: "owner")],
            candidates: ["owner"])
        await settle(coordinator, sid)

        #expect(await store.heldSeq(sid) == 5)
        let url = try #require(await store.url(sid))
        #expect(try TranscriptReader(url: url).read().events.map(\.seq) == [1, 2, 3, 4, 5])
    }

    @Test func interactivePriorityJoinsExistingMirrorFetch() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let log = CallLog()
        let gate = FetchGate()
        let sid = SessionID(rawValue: "single-flight")
        let coordinator = TranscriptSyncCoordinator(mirrorStore: store) {
            sessionID, afterSeq, holder in
            await log.record(sessionID, afterSeq, holder)
            return AsyncStream { continuation in
                Task {
                    await gate.wait()
                    continuation.yield(.chunk(TranscriptChunk(
                        requestID: "joined", sessionID: sessionID, batchIndex: 0,
                        events: (1...3).map { mirrorEvent(UInt64($0)) })))
                    continuation.yield(.complete(TranscriptFetchComplete(
                        requestID: "joined", sessionID: sessionID, throughSeq: 3,
                        headerJSON: try? mirrorHeader().encodedForWire())))
                    continuation.finish()
                }
            }
        }
        let ref = TranscriptSyncCoordinator.SessionRef(
            sessionID: sid, lastSeq: 3, archived: false, ownerDeviceID: "owner")

        await coordinator.sync(sessions: [ref], candidates: ["owner"])
        while await log.count == 0 { await Task.yield() }
        let foreground = Task { await coordinator.prioritize(ref, candidates: ["owner"]) }
        for _ in 0..<20 { await Task.yield() }
        #expect(await log.count == 1) // the open joined; it did not request the history again

        await gate.release()
        await foreground.value
        #expect(await store.heldSeq(sid) == 3)
        #expect(await log.count == 1)
    }

    @Test func fallsThroughNotHeldToNextHolder() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        // The owner is offline (not a candidate); peerB doesn't hold it, peerC does.
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store, fetch: makeFetch(log: log, goodHolders: ["peerC"], tip: 4))
        let sid = SessionID(rawValue: "s1")

        await coordinator.sync(
            sessions: [.init(sessionID: sid, lastSeq: 4, archived: false, ownerDeviceID: "owner")],
            candidates: ["peerB", "peerC"])
        await settle(coordinator, sid)

        #expect(await store.heldSeq(sid) == 4)
        #expect(await log.count == 2)  // probed peerB (notHeld) then peerC (served)
    }

    @Test func noFetchWhenAlreadyCurrent() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(),
            events: (1...5).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store, fetch: makeFetch(log: log, goodHolders: ["owner"], tip: 5))

        await coordinator.sync(
            sessions: [.init(sessionID: sid, lastSeq: 5, archived: false, ownerDeviceID: "owner")],
            candidates: ["owner"])
        await settle(coordinator, sid)

        #expect(await log.count == 0)  // held == tip ⇒ no gap ⇒ no fetch
    }

    @Test func archivedSessionDropsMirror() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(), events: [mirrorEvent(1)])
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store, fetch: makeFetch(log: log, goodHolders: ["owner"], tip: 1))

        await coordinator.sync(
            sessions: [.init(sessionID: sid, lastSeq: 1, archived: true, ownerDeviceID: "owner")],
            candidates: ["owner"])

        #expect(await store.has(sid) == false)  // archived ⇒ mirror dropped
    }

    /// A device offline during an undo never sees the `.transcriptReverted` push, so its mirror
    /// keeps the reverted tail. On reconnect the owner's summary reports a *lower* tip than we hold;
    /// the coordinator must heal by refetching the owner's authoritative history from scratch (a
    /// blind truncate would keep stale content at seqs the post-revert turns reuse).
    @Test func healsMirrorThatOutranOwnerAfterMissedRevert() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        // We mirrored up to seq 5, then went offline; the owner reverted to a tip of 3 meanwhile.
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(),
            events: (1...5).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store, fetch: makeFetch(log: log, goodHolders: ["owner"], tip: 3))

        await coordinator.sync(
            sessions: [.init(sessionID: sid, lastSeq: 3, archived: false, ownerDeviceID: "owner")],
            candidates: ["owner"])
        await settle(coordinator, sid)

        #expect(await store.heldSeq(sid) == 3)  // healed down to the owner's post-revert tip
        let url = try #require(await store.url(sid))
        #expect(try TranscriptReader(url: url).read().events.map(\.seq) == [1, 2, 3])  // 4,5 gone
        #expect(await log.calls.first?.afterSeq == 0)  // a full refetch, not a gap-fill
    }

    /// What one holder serves in the epoch-aware fetch fake: its tip, the revert epoch it stamps
    /// on the served header + completion (nil = an old build that strips/never had it), and
    /// whether its content is the post-revert regrowth (vs the original pre-revert events).
    private struct Holder: Sendable {
        let tip: UInt64
        let epoch: UInt64?
        var regrown = false
    }

    /// Epoch-aware fetch fake: each holder in `holders` serves its canned generation; everyone
    /// else replies `notHeld`. Logs every call like `makeFetch`.
    private func makeEpochFetch(log: CallLog, holders: [String: Holder])
        -> @Sendable (SessionID, UInt64, String) async -> AsyncStream<PeerClient.TranscriptFetchUpdate> {
        { sessionID, afterSeq, holder in
            await log.record(sessionID, afterSeq, holder)
            return AsyncStream { cont in
                guard let h = holders[holder] else {
                    cont.yield(.unavailable(TranscriptUnavailable(
                        requestID: "x", sessionID: sessionID, reason: .notHeld)))
                    cont.finish()
                    return
                }
                if afterSeq < h.tip {
                    let events = ((afterSeq + 1)...h.tip).map {
                        h.regrown ? regrownEvent($0) : mirrorEvent($0)
                    }
                    cont.yield(.chunk(TranscriptChunk(
                        requestID: "x", sessionID: sessionID, batchIndex: 0, events: events)))
                }
                cont.yield(.complete(TranscriptFetchComplete(
                    requestID: "x", sessionID: sessionID, throughSeq: max(afterSeq, h.tip),
                    headerJSON: afterSeq == 0
                        ? (try? mirrorHeader(epoch: h.epoch).encodedForWire()) : nil,
                    revertEpoch: h.epoch)))
                cont.finish()
            }
        }
    }

    /// THE missed-revert case tip arithmetic can't see (the reason revert epochs exist): we
    /// mirrored to seq 5, went offline, the owner reverted to 3 and regrew to 7 — its tip is now
    /// *ahead* of our cursor, so the old `held > lastSeq` heal never fires and a plain top-up
    /// would stitch new events 6..7 onto our stale 4..5. The epoch mismatch must force a full
    /// refetch that replaces the whole mirror with the owner's post-revert generation.
    @Test func healsEpochMismatchWhenRegrowthPassedOurCursor() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(epoch: 1),
            events: (1...5).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store,
            fetch: makeEpochFetch(
                log: log, holders: ["owner": Holder(tip: 7, epoch: 2, regrown: true)]))

        await coordinator.sync(
            sessions: [.init(
                sessionID: sid, lastSeq: 7, revertEpoch: 2, archived: false,
                ownerDeviceID: "owner")],
            candidates: ["owner"])
        await settle(coordinator, sid)

        #expect(await log.calls.first?.afterSeq == 0)  // full refetch, NOT a 5→7 top-up
        #expect(await store.heldSeq(sid) == 7)
        #expect(await store.heldEpoch(sid) == 2)
        let read = try TranscriptReader(url: #require(await store.url(sid))).read()
        // Every seq — including the reused 1..5 range — is the owner's post-revert content.
        #expect(read.events.allSatisfy {
            if case .assistantText(let c) = $0.kind { return c.text.hasPrefix("regrown") }
            return false
        })
    }

    /// A top-up whose serve raced a fresh revert (the holder's epoch moved past the summary we
    /// reconciled against) must be discarded, not appended — the next tick heals against the new
    /// advertised epoch instead.
    @Test func topUpRejectedWhenServeRacesARevert() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(epoch: 1),
            events: (1...3).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store,
            fetch: makeEpochFetch(
                log: log, holders: ["owner": Holder(tip: 5, epoch: 2, regrown: true)]))

        // The summary we reconcile on still says epoch 1 — the owner reverted after reporting it.
        await coordinator.sync(
            sessions: [.init(
                sessionID: sid, lastSeq: 5, revertEpoch: 1, archived: false,
                ownerDeviceID: "owner")],
            candidates: ["owner"])
        await settle(coordinator, sid)

        #expect(await log.calls.first?.afterSeq == 3)  // it was a top-up attempt…
        #expect(await store.heldSeq(sid) == 3)         // …whose mismatched serve was discarded
        #expect(await store.heldEpoch(sid) == 1)
    }

    /// A heal must not accept a copy from a holder still serving the *old* generation (its own
    /// mirror missed the same revert) — it probes on and takes the copy whose epoch matches.
    @Test func healRejectsStaleHolderAndTakesMatchingOne() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(epoch: 1),
            events: (1...5).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        // Owner offline. peerB is stale (epoch 1, pre-revert content); peerC synced post-revert.
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store,
            fetch: makeEpochFetch(log: log, holders: [
                "peerB": Holder(tip: 5, epoch: 1),
                "peerC": Holder(tip: 5, epoch: 2, regrown: true),
            ]))

        await coordinator.sync(
            sessions: [.init(
                sessionID: sid, lastSeq: 5, revertEpoch: 2, archived: false,
                ownerDeviceID: "owner")],
            candidates: ["peerB", "peerC"])
        await settle(coordinator, sid)

        #expect(await log.count == 2)  // probed stale peerB, rejected it, took peerC
        #expect(await store.heldEpoch(sid) == 2)
        let read = try TranscriptReader(url: #require(await store.url(sid))).read()
        #expect(read.events.allSatisfy {
            if case .assistantText(let c) = $0.kind { return c.text.hasPrefix("regrown") }
            return false
        })
    }

    /// While the owner is unreachable and the only holder can't prove its epoch (an old build
    /// that strips the header field), the heal runs once per advertised epoch — not every tick —
    /// and re-runs the moment the owner is back among the candidates.
    @Test func epochHealRunsOncePerEpochWhileOwnerUnreachable() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        // A pre-epoch mirror of a session whose owner has reverted at some point (epoch 1).
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(),
            events: (1...3).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store,
            fetch: makeEpochFetch(log: log, holders: [
                "peerB": Holder(tip: 3, epoch: nil),  // old build: can't vouch for its epoch
                "owner": Holder(tip: 3, epoch: 1, regrown: true),
            ]))
        let ref = { (candidates: [String]) async in
            await coordinator.sync(
                sessions: [.init(
                    sessionID: sid, lastSeq: 3, revertEpoch: 1, archived: false,
                    ownerDeviceID: "owner")],
                candidates: candidates)
            await self.settle(coordinator, sid)
        }

        await ref(["peerB"])
        #expect(await log.count == 1)  // healed once (best-effort copy from peerB, epoch unknown)
        #expect(await store.heldEpoch(sid) == nil)

        await ref(["peerB"])
        #expect(await log.count == 1)  // guard: same epoch, owner still unreachable — no re-drop

        await ref(["peerB", "owner"])
        #expect(await log.count == 2)  // owner reachable again — heal re-runs and confirms
        #expect(await store.heldEpoch(sid) == 1)

        await ref(["peerB", "owner"])
        #expect(await log.count == 2)  // converged — nothing more to do
    }

    /// A connected peer that applied the live `.transcriptReverted` (truncate + epoch stamp) must
    /// NOT be healed on the next reconcile — its copy already converged.
    @Test func stampedTruncatePreventsSpuriousHeal() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(epoch: 1),
            events: (1...5).map { mirrorEvent(UInt64($0)) })
        // The live revert path: truncate to the new tip, stamping the owner's bumped epoch.
        _ = try await store.truncate(sid, throughSeq: 3, revertEpoch: 2)
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store,
            fetch: makeEpochFetch(log: log, holders: ["owner": Holder(tip: 3, epoch: 2)]))

        await coordinator.sync(
            sessions: [.init(
                sessionID: sid, lastSeq: 3, revertEpoch: 2, archived: false,
                ownerDeviceID: "owner")],
            candidates: ["owner"])
        await settle(coordinator, sid)

        #expect(await log.count == 0)  // epochs match, cursor matches — nothing to fetch
    }

    /// A pre-epoch mirror of a session that has NEVER been reverted (owner advertises epoch 0)
    /// is fine as-is — no upgrade-day refetch storm.
    @Test func preEpochMirrorOfNeverRevertedSessionNotHealed() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(),
            events: (1...3).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store,
            fetch: makeEpochFetch(log: log, holders: ["owner": Holder(tip: 3, epoch: 0)]))

        await coordinator.sync(
            sessions: [.init(
                sessionID: sid, lastSeq: 3, revertEpoch: 0, archived: false,
                ownerDeviceID: "owner")],
            candidates: ["owner"])
        await settle(coordinator, sid)

        #expect(await log.count == 0)
    }

    /// If no holder is reachable to heal from, the stale mirror is still dropped — better to hold
    /// nothing (a later on-demand open refetches fresh) than to keep serving the reverted tail.
    @Test func dropsStaleMirrorWhenNoHolderCanHeal() async throws {
        let root = tempRoot()
        defer { try? FileManager.default.removeItem(at: root) }
        let store = TranscriptMirrorStore(root: root)
        let sid = SessionID(rawValue: "s1")
        _ = try await store.writeFull(
            sid, sourceDeviceID: "owner", header: mirrorHeader(),
            events: (1...5).map { mirrorEvent(UInt64($0)) })
        let log = CallLog()
        // Owner offline; the only reachable candidate doesn't hold it.
        let coordinator = TranscriptSyncCoordinator(
            mirrorStore: store, fetch: makeFetch(log: log, goodHolders: [], tip: 3))

        await coordinator.sync(
            sessions: [.init(sessionID: sid, lastSeq: 3, archived: false, ownerDeviceID: "owner")],
            candidates: ["peerB"])
        await settle(coordinator, sid)

        #expect(await store.has(sid) == false)
    }
}
