import Foundation
import CryptoKit
import RastrilloApple
import RastrilloNative
import RastrilloTestSupport
import XCTest
@testable import DocsKit

/// Uses the production WASM replica: a lost response really integrates bytes.
private actor IntegratingServer: NativeTransport {
    nonisolated let origin = URL(string: "http://localhost:8099")!
    var creationTokens: Set<String> = []
    func documentCount() -> Int { creationTokens.count }
    var online = true
    var title = "Document"
    var titleFailures = 0
    var titleDelay = false
    var updateDelay = false
    var updateRequests = 0
    func delayUpdates() { updateDelay = true }
    func updateRequestCount() -> Int { updateRequests }
    var titleRequests = 0
    func delayTitles() { titleDelay = true }
    func titleRequestCount() -> Int { titleRequests }
    func failTitles(_ count: Int) { titleFailures = count }
    func savedTitle() -> String { title }

    var loseNextResponse = false
    var sequence: Int64 = 0
    var rows: [Int64: Data] = [:]
    var listeners: [UUID: AsyncStream<Int64>.Continuation] = [:]
    func setOnline(_ value: Bool) { online = value }
    let replica = CRDTReplica(guid: "offline-property")
    var accepted: [UUID: Data] = [:]
    func open(_ initial: Data) async throws { _ = try await replica.open(update: initial) }
    func accept(_ entries: [OutboxEntry]) async throws {
        for entry in entries { _ = try await replica.apply(entry.payload); accepted[entry.id] = entry.payload }
    }
    func snapshot() async throws -> Data { try await replica.persistentState().state }
    func text() async throws -> String { try await replica.projection().children.map(\.plainText).joined() }
    func responseDetails(_ path: String, method: String, body: Data?, contentType: String, maxBytes: Int) async throws -> NativeResponse {
        guard online else { throw NativeClientError.offline(.notConnectedToInternet) }
        if method == "POST", path == "/documents", let body {
            let token = URLComponents(string: "?" + String(decoding: body, as: UTF8.self))?.queryItems?.first(where: { $0.name == "creation_token" })?.value ?? ""
            creationTokens.insert(token)
            return NativeResponse(data: Data(#"{"sqid":"doc","title":"Document"}"#.utf8), status: 201, headers: [:])
        }
        if method == "POST", path.hasSuffix("/title"), let body {
            titleRequests += 1
            if titleDelay { try await Task.sleep(for: .milliseconds(200)) }
            if titleFailures > 0 { titleFailures -= 1; throw NativeClientError.server(status: 503) }
            title = URLComponents(string: "?" + String(decoding: body, as: UTF8.self))?.queryItems?.first(where: { $0.name == "title" })?.value ?? ""
            return NativeResponse(data: Data(), status: 200, headers: [:])
        }
        if path.hasPrefix("/?") {
            return NativeResponse(data: Data(#"{"documents":[],"can_create":true}"#.utf8), status: 200, headers: [:])
        }
        if method == "POST", path.hasSuffix("/updates"), let body {
            updateRequests += 1
            if updateDelay { try await Task.sleep(for: .milliseconds(300)) }
            _ = try await replica.apply(body); sequence += 1
            rows[sequence] = body
            for listener in listeners.values { listener.yield(sequence) }
            if loseNextResponse { loseNextResponse = false; throw NativeClientError.offline(.networkConnectionLost) }
            return NativeResponse(data: Data(), status: 204, headers: [:])
        }
        if path.contains("/updates") {
            let since = URLComponents(string: path)?.queryItems?.first(where: { $0.name == "since" })?.value.flatMap(Int64.init) ?? 0
            let data: Data
            if since > 0 {
                let updates = rows.keys.sorted().filter { $0 > since }.compactMap { rows[$0] }
                data = updates.isEmpty ? Data() : try await replica.mergeUpdates(updates)
            } else { data = try await snapshot() }
            return NativeResponse(data: data, status: 200, headers: ["Docs-Seq": String(sequence)])
        }
        let detail = DocumentDetail(sqid: "doc", title: title, guid: "offline-property", can_write: true, can_share: false)
        return NativeResponse(data: try JSONEncoder().encode(detail), status: 200, headers: [:])
    }
    func stream(_ path: String, onOpen: @Sendable () async throws -> Void, onLine: @Sendable (String) async throws -> Void) async throws {
        guard online else { throw NativeClientError.offline(.notConnectedToInternet) }
        let id = UUID(), events = AsyncStream<Int64>.makeStream(bufferingPolicy: .bufferingNewest(1))
        listeners[id] = events.continuation
        defer { listeners.removeValue(forKey: id) }
        try await onOpen(); try await onLine("data: \(sequence)")
        for await sequence in events.stream { try Task.checkCancellation(); try await onLine("data: \(sequence)") }
    }
    func download(_ path: String, to destination: URL) async throws { throw NativeClientError.notFound }

    func contains(_ id: UUID) -> Bool { accepted[id] != nil }
}
private final class MetadataLockStore: BlobStore, @unchecked Sendable {
    let base = MemoryBlobStore()
    private let lock = NSLock()
    private var locked = true
    func unlock() { lock.withLock { locked = false } }
    func read(_ path: String) throws -> Data? { try base.read(path) }
    func list(_ path: String) throws -> [String] { try base.list(path) }
    func remove(_ path: String) throws { try base.remove(path) }
    func write(_ bytes: Data, to path: String) throws {
        if path.hasSuffix("/meta"), lock.withLock({ locked }) { throw BlobStoreError.unavailable }
        try base.write(bytes, to: path)
    }
}
private struct OfflineRandom: RandomNumberGenerator {
    var seed: UInt64
    mutating func next() -> UInt64 { seed = seed &* 6364136223846793005 &+ 1442695040888963407; return seed }
}
@MainActor final class OfflineDurabilityTests: XCTestCase {
    private func initial() async throws -> Data {
        let seed = CRDTReplica(guid: "offline-property")
        _ = try await seed.open(update: Data())
        let (update, _) = try await seed.command(["op": .string("insert_element"), "at": .number(0), "type": .string("paragraph")])
        await seed.close(); return update
    }
    private func edit(_ replica: CRDTReplica, _ text: String) async throws -> Data {
        let projection = try await replica.projection()
        let at = projection.children.map(\.plainText).joined().utf16.count
        return try await replica.command(["op": .string("display_edit"), "at": .number(Double(at)), "length": .number(0), "text": .string(text)]).0
    }
    func testMissingOrCorruptMetadataHoldsOnlyItsLane() async throws {
        for damage in 0..<3 {
            let root = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
            defer { try? FileManager.default.removeItem(at: root) }
            let disk = NativeSealedStore(root: root, accountID: "a", keys: ScriptedKeyProvider())
            let connectivity = NativeConnectivity(source: ScriptedPathSource(.satisfied))
            while !connectivity.isPathSatisfied { await Task.yield() }
            let transport = ScriptedTransport([])
            let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: connectivity, transport: { transport })
            if damage > 0 { try disk.write(Data("invalid metadata".utf8), to: "cache/docs/a/meta") }
            if damage == 2 { try Data([0]).write(to: root.appendingPathComponent("cache/docs/a/meta")) }
            try await account.outbox.enqueue(lane: "doc:a", kind: "metadata", payload: Data())
            let detail = DocumentDetail(sqid: "b", title: "B", guid: "b", can_write: true, can_share: false)
            try account.saveMetadata(DocsOfflineMetadata(detail: detail), key: "b")
            try await account.outbox.enqueue(lane: "doc:b", kind: "metadata", payload: Data())
            await account.drainer.drain()
            let broken = await account.outbox.entries(lane: "doc:a")
            let good = await account.outbox.entries(lane: "doc:b")
            XCTAssertEqual(broken.first?.hold, Hold(.rejected("metadata")))
            XCTAssertTrue(good.isEmpty, "Missing metadata must not block another document")
        }
    }

    func testHeldMetadataMarkerDoesNotMultiplyOnFlush() async throws {
        let account = DocsOfflineAccount(store: MemoryBlobStore(), accountID: "a", connectivity: NativeConnectivity(source: ScriptedPathSource(.unsatisfied)), transport: { ScriptedTransport([]) })
        let detail = DocumentDetail(sqid: "doc", title: "Title", guid: "doc", can_write: true, can_share: false)
        try account.saveMetadata(DocsOfflineMetadata(detail: detail, pendingTitle: "Rename"), key: "doc")
        try await account.ensureMetadataEntry("doc")
        let original = await account.outbox.entries(lane: "doc:doc")
        try await account.outbox.hold(original.map(\.id), Hold(.accessLost))
        for _ in 0..<4 { try await account.ensureMetadataEntry("doc") }
        let entries = await account.outbox.entries(lane: "doc:doc")
        XCTAssertEqual(entries.map(\.id), original.map(\.id))
    }

    func testDurableOfflineLaneUsesSyncStatusWithoutAnErrorMessage() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let path = ScriptedPathSource(.satisfied)
        let connectivity = NativeConnectivity(source: path)
        let account = DocsOfflineAccount(store: MemoryBlobStore(), accountID: "a", connectivity: connectivity, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: connectivity, offlineAccount: account)
        await model.open("doc"); _ = await model.saveAll()
        path.send(.unsatisfied)
        while connectivity.isPathSatisfied { await Task.yield() }
        model.didRender(model.projectionGeneration)
        XCTAssertTrue(model.enqueue(["op": .string("display_edit"), "at": .number(0), "length": .number(0), "text": .string("Offline")]))
        while model.hasPendingInput { await Task.yield() }
        await model.flush()
        XCTAssertNil(model.errorKey, "A durable pending lane does not require keeping this window open")
        XCTAssertEqual(model.syncStatus, .offline(pending: 1))
        await model.stop(); account.drainer.stop()
    }

    func testWriterBetweenCommandAndEnqueueCannotPersistTheCommand() async throws {
        let disk = MemoryBlobStore(), initial = try await initial()
        let replica = CRDTReplica(guid: "offline-property")
        let lane = DocsPersistence(store: disk, outbox: Outbox(store: disk, account: "a"), key: "doc", replica: replica)
        try await lane.reopen(fallback: initial, sequence: 17)
        let wroteInitial = try await lane.persist()
        XCTAssertTrue(wroteInitial)
        let before = disk.contents[lane.path]
        let update = try await edit(replica, "A")
        let premature = try await lane.persist()
        XCTAssertFalse(premature, "I1a: command returned, but enqueue has not")
        XCTAssertEqual(disk.contents[lane.path], before)
        try await lane.enqueue(update)
        let wrote = try await lane.persist()
        XCTAssertTrue(wrote)
        let saved = try JSONDecoder().decode(PersistedReplica.self, from: XCTUnwrap(disk.contents[lane.path]))
        XCTAssertEqual(saved.sequence, 17)
        XCTAssertFalse(saved.state.isEmpty)
    }
    func testReopenAppliesTheWholeLaneBeforePublishingAndDoesNotDuplicate() async throws {
        let disk = MemoryBlobStore(), initial = try await initial()
        let outbox = Outbox(store: disk, account: "a"), replica = CRDTReplica(guid: "offline-property")
        let lane = DocsPersistence(store: disk, outbox: outbox, key: "draft-token", replica: replica)
        try await lane.reopen(fallback: initial, sequence: 0)
        try await lane.enqueue(edit(replica, "A"))
        _ = try await lane.persist()
        try await lane.enqueue(edit(replica, "B"))
        for _ in 0..<2 {
            let reopened = CRDTReplica(guid: "offline-property")
            let next = DocsPersistence(store: disk, outbox: Outbox(store: disk, account: "a"), key: "draft-token", replica: reopened)
            try await next.reopen(fallback: Data(), sequence: 0)
            let text = try await reopened.projection().children.map(\.plainText).joined()
            XCTAssertEqual(text, "AB", "I1b: replica contains A; lane contains A and B")
        }
        try await lane.discard()
        XCTAssertNil(disk.contents[lane.path], "Discard cannot leave stranded local text in the replica")
        let entries = await outbox.entries(lane: lane.lane)
        XCTAssertTrue(entries.isEmpty)
    }
    /// Deterministic generated histories, plus an exhaustive crash sweep at
    /// replacement's write and every delete. Every assertion reads committed
    /// bytes through a fresh outbox/replica, not the live writer's memory.
    func testGeneratedCrashHistoriesPreserveI1AndConverge() async throws {
        let initial = try await initial()
        for seed in 1...12 {
            var random = OfflineRandom(seed: UInt64(seed))
            var disk = FaultingBlobStore()
            var outbox = Outbox(store: disk, account: "a")
            var replica = CRDTReplica(guid: "offline-property")
            var lane = DocsPersistence(store: disk, outbox: outbox, key: "doc", replica: replica)
            try await lane.reopen(fallback: initial, sequence: 0)
            let server = IntegratingServer(); try await server.open(initial)
            var durableText = ""
            var flying: [OutboxEntry] = []
            var symbol = 0
            for step in 0..<60 {
                if step % 4 == 0 { disk.crash(after: Int.random(in: 0...10, using: &random)) }
                do {
                    switch Int.random(in: 0..<8, using: &random) {
                    case 0, 1:
                        symbol += 1
                        let text = "[\(symbol)]"
                        let update = try await edit(replica, text)
                        let premature = try await lane.persist()
                        XCTAssertFalse(premature, "Writer tick between command and enqueue")
                        try await lane.enqueue(update); durableText += text
                    case 2: _ = try await lane.persist()
                    case 3, 4:
                        if flying.isEmpty { flying = try await outbox.beginSend(lane: lane.lane, maxBytes: 80) }
                        try await server.accept(flying)
                        if step % 2 == 0 {
                            for entry in flying { let has = await server.contains(entry.id); XCTAssertTrue(has) }
                            try await outbox.ack(flying.map(\.id))
                        } // otherwise the response is lost: server has it; disk keeps it
                        await outbox.endSend(flying.map(\.id)); flying = []
                    case 5:
                        let entries = await outbox.entries(lane: lane.lane)
                        if entries.count >= 2 {
                            if step % 2 == 0 {
                                flying = try await outbox.beginSend(lane: lane.lane, maxBytes: 1)
                                do { try await outbox.replace(entries.map(\.id), with: replica.mergeUpdates(entries.map(\.payload))); XCTFail("In-flight replacement must throw") }
                                catch OutboxError.inFlight {}
                                await outbox.endSend(flying.map(\.id)); flying = []
                            } else { try await lane.compact() }
                        }
                    case 6:
                        // A crash before any subsequent store op, including a
                        // writer tick or replacement's first/second step.
                        disk.crash(after: Int.random(in: 0...2, using: &random))
                        let entries = await outbox.entries(lane: lane.lane)
                        if entries.count >= 2 { try await outbox.replace(entries.map(\.id), with: replica.mergeUpdates(entries.map(\.payload))) }
                        else { _ = try await lane.persist() }
                    default:
                        for entry in try await outbox.beginSend(lane: lane.lane, maxBytes: nil) {
                            try await server.accept([entry]); try await outbox.ack([entry.id])
                        }
                    }
                } catch is SimulatedCrash { }
                // Reopen after *every* event. This simultaneously checks the
                // missing-lane and duplicate-text cases at each crash boundary.
                let frozen = disk.committed
                let checked = CRDTReplica(guid: "offline-property")
                let checkLane = DocsPersistence(store: frozen, outbox: Outbox(store: frozen, account: "a"), key: "doc", replica: checked)
                try await checkLane.reopen(fallback: server.snapshot(), sequence: 0)
                // Catch up with already acknowledged edits if the replica writer
                // lagged behind ack; real reopen performs the same server catch-up.
                _ = try await checked.apply(server.snapshot())
                let recovered = try await checked.projection().children.map(\.plainText).joined()
                XCTAssertEqual(recovered, durableText, "seed \(seed), event \(step): I1b / no duplicates")
                if let data = frozen.contents[lane.path] {
                    let saved = try JSONDecoder().decode(PersistedReplica.self, from: data)
                    let proof = CRDTReplica(guid: "offline-property")
                    _ = try await proof.open(update: server.snapshot())
                    for entry in await Outbox(store: frozen, account: "a").entries(lane: lane.lane) { _ = try await proof.apply(entry.payload) }
                    let prior = try await proof.projection().children.map(\.plainText)
                    _ = try await proof.apply(saved.state)
                    let after = try await proof.projection().children.map(\.plainText)
                    XCTAssertEqual(prior, after, "I1a: replica contains no stranded local edit")
                    await proof.close()
                }
                await checked.close()
                if disk.hasCrashed {
                    await replica.close()
                    disk = FaultingBlobStore(frozen); outbox = Outbox(store: disk, account: "a")
                    replica = CRDTReplica(guid: "offline-property")
                    lane = DocsPersistence(store: disk, outbox: outbox, key: "doc", replica: replica)
                    try await lane.reopen(fallback: server.snapshot(), sequence: 0)
                    _ = try await replica.apply(server.snapshot())
                    flying = []
                }
            }
            // Reconnect and fully drain, with the real integrating replica.
            let recovered = Outbox(store: disk.committed, account: "a")
            let entries = await recovered.entries(lane: lane.lane)
            try await server.accept(entries); try await recovered.ack(entries.map(\.id))
            let text = try await server.text()
            XCTAssertEqual(text, durableText, "seed \(seed): convergence after drain")
            await replica.close()
        }
    }
    func testEveryReplaceCrashBoundaryReopensWithoutLossOrDuplicateText() async throws {
        let initial = try await initial()
        let base = MemoryBlobStore(), source = CRDTReplica(guid: "offline-property")
        let setup = DocsPersistence(store: base, outbox: Outbox(store: base, account: "a"), key: "doc", replica: source)
        try await setup.reopen(fallback: initial, sequence: 0)
        _ = try await setup.persist()
        for text in ["A", "B", "C"] { try await setup.enqueue(edit(source, text)) }
        for crashAfter in 0...3 {
            let disk = FaultingBlobStore(MemoryBlobStore(base.contents))
            let outbox = Outbox(store: disk, account: "a")
            try await outbox.load()
            let entries = await outbox.entries(lane: "doc:doc")
            let merged = try await source.mergeUpdates(entries.map(\.payload))
            disk.crash(after: crashAfter)
            do { try await outbox.replace(entries.map(\.id), with: merged) }
            catch is SimulatedCrash { }
            let rebuilt = CRDTReplica(guid: "offline-property")
            let frozen = disk.committed
            let recovered = DocsPersistence(store: frozen, outbox: Outbox(store: frozen, account: "a"), key: "doc", replica: rebuilt)
            try await recovered.reopen(fallback: Data(), sequence: 0)
            let text = try await rebuilt.projection().children.map(\.plainText).joined()
            XCTAssertEqual(text, "ABC", "Crash at replace operation \(crashAfter)")
            await rebuilt.close()
        }
        await source.close()
    }

    func testModelReopensOfflineAndDrainsToIntegratingTransport() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let disk = FaultingBlobStore()
        let connectivity = NativeConnectivity(source: ScriptedPathSource(.satisfied))
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: connectivity, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: connectivity, offlineAccount: account)
        await model.open("doc")
        XCTAssertNotNil(model.projection)
        await server.setOnline(false)
        model.didRender(model.projectionGeneration)
        XCTAssertTrue(model.enqueue(["op": .string("display_edit"), "at": .number(0), "length": .number(0), "text": .string("Offline edit")]))
        while model.hasPendingInput { try await Task.sleep(for: .milliseconds(5)) }
        XCTAssertTrue(model.hasUnsentChanges)
        // Kill before the two-second writer. Freeze this process's store so
        // its tasks cannot mutate the new process's committed view.
        account.drainer.stop(); disk.crash(after: 0)
        let frozen = disk.committed
        let nextAccount = DocsOfflineAccount(store: frozen, accountID: "a", connectivity: connectivity, transport: { server })
        let reopened = DocsModel(transport: { server }, connectivity: connectivity, offlineAccount: nextAccount)
        await reopened.open("doc")
        XCTAssertEqual(reopened.projection?.children.map(\.plainText).joined(), "Offline edit")
        await server.setOnline(true)
        await reopened.retry()
        let deadline = ContinuousClock.now.advanced(by: .seconds(3))
        while reopened.hasUnsavedChanges && .now < deadline { await reopened.flush(); try await Task.sleep(for: .milliseconds(10)) }
        XCTAssertFalse(reopened.hasUnsavedChanges)
        let text = try await server.text()
        XCTAssertEqual(text, "Offline edit")
        await reopened.stop(); nextAccount.drainer.stop()
    }

    func testUnlockRetriesFailedUpdateAndResumesAlreadyQueuedCommands() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let disk = FaultingBlobStore()
        let connectivity = NativeConnectivity(source: ScriptedPathSource(.satisfied))
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: connectivity, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: connectivity, offlineAccount: account)
        await model.open("doc")
        await server.setOnline(false)
        disk.unavailable = true
        for text in ["A", "B"] {
            XCTAssertTrue(model.enqueue(["op": .string("display_edit"), "at": .number(0), "length": .number(0), "text": .string(text)]))
        }
        let deadline = ContinuousClock.now.advanced(by: .seconds(2))
        while !model.inputPaused && .now < deadline { try await Task.sleep(for: .milliseconds(5)) }
        XCTAssertTrue(model.inputPaused)
        disk.unavailable = false
        await model.retry()
        let resumed = ContinuousClock.now.advanced(by: .seconds(2))
        while model.hasPendingInput && .now < resumed { try await Task.sleep(for: .milliseconds(5)) }
        XCTAssertFalse(model.hasPendingInput, "Retry must resume the queued second command")
        XCTAssertEqual(model.projection?.children.map(\.plainText).joined(), "BA")
        await model.persistReplica()
        XCTAssertFalse(model.hasUnpersistedInput)
        await model.discardAll(); account.drainer.stop()
    }

    func testDurableTitleSurvivesReopenBeforeFirstReplicaWrite() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let disk = MemoryBlobStore()
        let path = ScriptedPathSource(.unsatisfied)
        let network = NativeConnectivity(source: path)
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { server })
        let detail = DocumentDetail(sqid: "doc", title: "Local title", guid: "offline-property", can_write: true, can_share: false)
        try account.saveMetadata(DocsOfflineMetadata(detail: detail, pendingTitle: "Local title"), key: "doc")
        let reopened = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        await reopened.open("doc")
        XCTAssertEqual(reopened.detail?.title, "Local title")
        XCTAssertEqual(try account.metadata("doc")?.pendingTitle, "Local title")
        await reopened.discardAll(); account.drainer.stop()
    }

    func testTitleOnlyTransientFailureRetriesWithoutAnotherEdit() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let network = NativeConnectivity(source: ScriptedPathSource(.satisfied))
        let account = DocsOfflineAccount(store: MemoryBlobStore(), accountID: "a", connectivity: network, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        await model.open("doc")
        try await Task.sleep(for: .milliseconds(100))
        await server.failTitles(1)
        XCTAssertTrue(model.rename("Retry title"))
        let deadline = ContinuousClock.now.advanced(by: .seconds(3))
        while await server.savedTitle() != "Retry title", .now < deadline { try await Task.sleep(for: .milliseconds(10)) }
        let title = await server.savedTitle()
        XCTAssertEqual(title, "Retry title")
        await model.discardAll(); account.drainer.stop()
    }

    func testHomeOnlyReplayPreservesAcknowledgedTextForOfflineReopen() async throws {
        let seed = try await initial(), disk = MemoryBlobStore()
        let server = IntegratingServer(); try await server.open(seed)
        let network = NativeConnectivity(source: ScriptedPathSource(.satisfied))
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { server })
        let detail = DocumentDetail(sqid: "doc", title: "Document", guid: "offline-property", can_write: true, can_share: false)
        try account.saveMetadata(DocsOfflineMetadata(detail: detail), key: "doc")
        let replica = CRDTReplica(guid: detail.guid)
        let lane = DocsPersistence(store: disk, outbox: account.outbox, key: "doc", replica: replica)
        try await lane.reopen(fallback: seed, sequence: 0)
        _ = try await lane.persist()
        try await lane.enqueue(edit(replica, "Home replay"))
        await replica.close()
        // Home loads the list; it never opens the document's editor.
        let home = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        await home.load()
        let deadline = ContinuousClock.now.advanced(by: .seconds(3))
        while !(await account.outbox.summary).isEmpty, .now < deadline { try await Task.sleep(for: .milliseconds(10)) }
        let serverText = try await server.text()
        XCTAssertEqual(serverText, "Home replay")
        let summary = await account.outbox.summary
        XCTAssertTrue(summary.isEmpty)
        account.drainer.stop(); await server.setOnline(false)
        let next = DocsOfflineAccount(store: disk, accountID: "a", connectivity: NativeConnectivity(source: ScriptedPathSource(.unsatisfied)), transport: { server })
        let reopened = DocsModel(transport: { server }, offlineAccount: next)
        await reopened.open("doc")
        XCTAssertEqual(reopened.projection?.children.map(\.plainText).joined(), "Home replay")
        await reopened.discardAll(); next.drainer.stop()
    }

    func testRenameDuringTitleRequestSendsTheLatestTitle() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let network = NativeConnectivity(source: ScriptedPathSource(.satisfied))
        let account = DocsOfflineAccount(store: MemoryBlobStore(), accountID: "a", connectivity: network, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        await model.open("doc")
        try await Task.sleep(for: .milliseconds(100))
        await server.delayTitles()
        XCTAssertTrue(model.rename("First"))
        let started = ContinuousClock.now.advanced(by: .seconds(2))
        while await server.titleRequestCount() == 0, .now < started { try await Task.sleep(for: .milliseconds(5)) }
        XCTAssertTrue(model.rename("Latest"))
        let deadline = ContinuousClock.now.advanced(by: .seconds(3))
        while await server.savedTitle() != "Latest", .now < deadline { try await Task.sleep(for: .milliseconds(10)) }
        let title = await server.savedTitle()
        XCTAssertEqual(title, "Latest")
        XCTAssertNil(try account.metadata("doc")?.pendingTitle)
        await model.discardAll(); account.drainer.stop()
    }

    func testAttachedDraftMetadataRepairsMissingIndexAtEveryWriteBoundary() async throws {
        let base = MemoryBlobStore()
        let network = NativeConnectivity(source: ScriptedPathSource(.unsatisfied))
        let original = DocsOfflineAccount(store: base, accountID: "a", connectivity: network, transport: { nil })
        let draft = DocumentDetail(sqid: "draft-token", title: "Draft", guid: "g", can_write: true, can_share: false)
        try original.saveMetadata(DocsOfflineMetadata(detail: draft, pendingTitle: "Renamed", draftToken: "token"), key: "draft-token")
        let attached = DocumentDetail(sqid: "real", title: "Renamed", guid: "g", can_write: true, can_share: false)
        for crashAfter in 0...8 {
            let disk = FaultingBlobStore(MemoryBlobStore(base.contents))
            let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { nil })
            disk.crash(after: crashAfter)
            do { try account.saveMetadata(DocsOfflineMetadata(detail: attached, pendingTitle: "Renamed"), key: "draft-token") }
            catch is SimulatedCrash { }
            let next = DocsOfflineAccount(store: disk.committed, accountID: "a", connectivity: network, transport: { nil })
            let metadata = try XCTUnwrap(next.metadata("draft-token"))
            if metadata.draftToken == nil {
                XCTAssertEqual(try next.key(for: "real"), "draft-token", "Crash boundary \(crashAfter): discover attached bytes at their original sealed path")
            }
            XCTAssertEqual(metadata.pendingTitle, "Renamed")
        }
    }

    func testConcurrentReplicaCommitsMergeAndDiscardInvalidatesOldCaptures() async throws {
        let disk = MemoryBlobStore(), cache = DocsReplicaCache(store: disk), seed = try await initial()
        let a = CRDTReplica(guid: "offline-property"), b = CRDTReplica(guid: "offline-property")
        _ = try await a.open(update: seed); _ = try await b.open(update: seed)
        let outbox = Outbox(store: disk, account: "a")
        try await outbox.enqueue(lane: "doc:doc", kind: "update", payload: edit(a, "A"))
        try await outbox.enqueue(lane: "doc:doc", kind: "update", payload: edit(b, "B"))
        let first = PersistedReplica(format: 1, sequence: 2, guid: "offline-property", savedAt: Date(), state: try await a.persistentState().state)
        let second = PersistedReplica(format: 1, sequence: 3, guid: "offline-property", savedAt: Date(), state: try await b.persistentState().state)
        async let writeA = cache.commit(first, key: "doc", generation: 0)
        async let writeB = cache.commit(second, key: "doc", generation: 0)
        _ = try await (writeA, writeB)
        // An older editor capture must not erase the headless sender's state.
        try await cache.commit(first, key: "doc", generation: 0)
        let saved = try JSONDecoder().decode(PersistedReplica.self, from: XCTUnwrap(disk.contents["cache/docs/doc/replica"]))
        let checked = CRDTReplica(guid: "offline-property")
        let text = try await checked.open(update: saved.state).children.map(\.plainText).joined()
        XCTAssertEqual(text.count, 2); XCTAssertTrue(text.contains("A")); XCTAssertTrue(text.contains("B"))
        XCTAssertEqual(saved.sequence, 3)
        cache.invalidate("doc"); try disk.remove("cache/docs/doc/replica")
        do { try await cache.commit(first, key: "doc", generation: 0); XCTFail("A discarded capture must not write") }
        catch is CancellationError { }
        XCTAssertNil(disk.contents["cache/docs/doc/replica"])
        await a.close(); await b.close(); await checked.close()
    }

    func testEveryDiscardCrashBoundaryLeavesNoStrandedReplica() async throws {
        let base = MemoryBlobStore(), replica = CRDTReplica(guid: "offline-property")
        let setup = DocsPersistence(store: base, outbox: Outbox(store: base, account: "a"), key: "doc", replica: replica)
        try await setup.reopen(fallback: initial(), sequence: 0)
        for text in ["A", "B"] { try await setup.enqueue(edit(replica, text)) }
        _ = try await setup.persist()
        for crashAfter in 0...6 {
            let disk = FaultingBlobStore(MemoryBlobStore(base.contents))
            let queue = Outbox(store: disk, account: "a")
            try await queue.load()
            let lane = DocsPersistence(store: disk, outbox: queue, key: "doc", replica: replica)
            disk.crash(after: crashAfter)
            do { try await lane.discard() } catch is SimulatedCrash { }
            let frozen = disk.committed
            let entries = await Outbox(store: frozen, account: "a").entries(lane: "doc:doc")
            if frozen.contents[lane.path] != nil {
                XCTAssertEqual(entries.count, 2, "Crash \(crashAfter): no unsent text may survive without its complete lane")
            }
        }
        await replica.close()
    }

    func testLockedRenameIsRetriedBeforeLocalFailureClears() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let disk = FaultingBlobStore(), network = NativeConnectivity(source: ScriptedPathSource(.unsatisfied))
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        await model.open("doc")
        disk.unavailable = true
        model.renameInput = "Keep this title"
        XCTAssertFalse(model.commitRename())
        XCTAssertFalse(model.commitRename(), "Equality with the failed in-memory title is not a durable commit")
        disk.unavailable = false
        await model.retry()
        XCTAssertEqual(try account.metadata("doc")?.pendingTitle, "Keep this title")
        XCTAssertTrue(model.commitRename())
        await model.persistReplica()
        XCTAssertEqual(try account.metadata("doc")?.pendingTitle, "Keep this title")
        await model.discardAll(); account.drainer.stop()
    }

    func testRenameAfterTitlePOSTButBeforeBatchAckKeepsSuccessorTrigger() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let path = ScriptedPathSource(.unsatisfied), disk = MemoryBlobStore()
        let network = NativeConnectivity(source: path)
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        while network.path?.status != .unsatisfied { try await Task.sleep(for: .milliseconds(5)) }
        await model.open("doc")
        model.didRender(model.projectionGeneration)
        XCTAssertTrue(model.enqueue(["op": .string("display_edit"), "at": .number(0), "length": .number(0), "text": .string("A")]))
        while model.hasPendingInput { try await Task.sleep(for: .milliseconds(5)) }
        var metadata = try XCTUnwrap(account.metadata("doc"))
        metadata.pendingTitle = "First"
        try account.saveMetadata(metadata, key: "doc")
        try await account.ensureMetadataEntry("doc")
        let queuedEntries = await account.outbox.entries(lane: "doc:doc")
        XCTAssertEqual(queuedEntries.map(\.kind), ["update", "metadata"])
        await server.delayUpdates(); path.send(.satisfied)
        let started = ContinuousClock.now.advanced(by: .seconds(2))
        while await server.updateRequestCount() == 0, .now < started { try await Task.sleep(for: .milliseconds(5)) }
        let flying = await account.outbox.summary.inFlight
        XCTAssertEqual(flying, 2)
        metadata = try XCTUnwrap(account.metadata("doc"))
        metadata.pendingTitle = "Successor"
        try account.saveMetadata(metadata, key: "doc")
        try await account.ensureMetadataEntry("doc")
        let deadline = ContinuousClock.now.advanced(by: .seconds(3))
        while await server.savedTitle() != "Successor", .now < deadline { try await Task.sleep(for: .milliseconds(10)) }
        let title = await server.savedTitle(); XCTAssertEqual(title, "Successor")
        await model.discardAll(); account.drainer.stop()
    }

    func testLostChangesNoticePersistsUntilDismissal() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let disk = MemoryBlobStore(["outbox/0000000000000001.entry": Data("corrupt".utf8)])
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: NativeConnectivity(source: ScriptedPathSource(.unsatisfied)), transport: { server })
        let home = DocsModel(transport: { server }, offlineAccount: account)
        await home.load()
        try await Task.sleep(for: .milliseconds(50))
        XCTAssertEqual(home.lostChangesNotice, NativeSyncStrings.kit.unsentLost)
        await home.load()
        XCTAssertNotNil(home.lostChangesNotice, "A list refresh must not dismiss the notice")
        await home.dismissLostChangesNotice()
        try await Task.sleep(for: .milliseconds(50))
        XCTAssertNil(home.lostChangesNotice)
        account.drainer.stop()
    }

    func testLaneCapCannotBeBypassedAndClearsAfterDrain() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let network = NativeConnectivity(source: ScriptedPathSource(.unsatisfied))
        let account = DocsOfflineAccount(store: MemoryBlobStore(), accountID: "a", connectivity: network, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        await model.open("doc")
        try await account.outbox.enqueue(lane: "doc:doc", kind: "metadata", payload: Data(repeating: 0, count: 32 * 1024 * 1024 + 1))
        try await Task.sleep(for: .milliseconds(50))
        XCTAssertTrue(model.inputPaused)
        XCTAssertEqual(model.syncStatus, .failed(pending: 1, reason: .tooManyUnsent))
        await model.retry(); XCTAssertTrue(model.inputPaused)
        let reopened = DocsModel(transport: { server }, connectivity: network, offlineAccount: account)
        await reopened.open("doc")
        XCTAssertTrue(reopened.inputPaused, "Reopening cannot bypass the byte ceiling")
        try await account.outbox.discard(lane: "doc:doc")
        try await Task.sleep(for: .milliseconds(50))
        XCTAssertFalse(model.inputPaused)
        XCTAssertEqual(model.syncStatus, .saved)
        await reopened.discardAll(); await model.discardAll(); account.drainer.stop()
    }

    func testIntegratingServerStreamsSubsequentUpdates() async throws {
        let server = IntegratingServer(), seed = try await initial()
        try await server.open(seed)
        let events = expectation(description: "Initial and subsequent sequence")
        events.expectedFulfillmentCount = 2
        let listener = Task { try await server.stream("/d/doc/stream", onOpen: {}) { _ in events.fulfill() } }
        defer { listener.cancel() }
        try await Task.sleep(for: .milliseconds(30))
        let local = CRDTReplica(guid: "offline-property"); _ = try await local.open(update: seed)
        let update = try await edit(local, "Remote")
        _ = try await server.responseDetails("/d/doc/updates", method: "POST", body: update, contentType: "application/octet-stream", maxBytes: 1024)
        await fulfillment(of: [events], timeout: 1)
        await local.close()
    }

    func testDraftAttachmentReopensAfterEveryStoreOperation() async throws {
        let seed = try await initial(), base = MemoryBlobStore()
        let network = NativeConnectivity(source: ScriptedPathSource(.satisfied))
        let original = DocsOfflineAccount(store: base, accountID: "a", connectivity: network, transport: { nil })
        let detail = DocumentDetail(sqid: "draft-token", title: "Draft title", guid: "offline-property", can_write: true, can_share: false)
        try original.saveMetadata(DocsOfflineMetadata(detail: detail, pendingTitle: "Draft title", draftToken: "token"), key: "draft-token")
        let source = CRDTReplica(guid: detail.guid); _ = try await source.open(update: seed)
        try await original.outbox.enqueue(lane: "doc:draft-token", kind: "update", payload: seed)
        try await original.outbox.enqueue(lane: "doc:draft-token", kind: "update", payload: edit(source, "Draft content"))
        await source.close()
        // Discover the actual operation count, then crash before every one.
        var limit = 0
        for crashAfter in [-1] + Array(0...80) {
            if crashAfter > limit && crashAfter >= 0 { break }
            let server = IntegratingServer(); try await server.open(Data())
            let disk = FaultingBlobStore(MemoryBlobStore(base.contents))
            let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { server })
            if crashAfter >= 0 { disk.crash(after: crashAfter) }
            do { try await account.activate(); await account.drainer.drain() } catch is SimulatedCrash { }
            account.drainer.stop()
            if crashAfter == -1 { limit = disk.operationCount }
            let frozen = disk.committed
            let recovered = DocsOfflineAccount(store: frozen, accountID: "a", connectivity: network, transport: { server })
            try await recovered.activate(); await recovered.drainer.drain()
            XCTAssertEqual(try recovered.key(for: "doc"), "draft-token", "Boundary \(crashAfter)")
            XCTAssertNil(try recovered.metadata("draft-token")?.draftToken)
            let text = try await server.text(), count = await server.documentCount(), summary = await recovered.outbox.summary
            XCTAssertEqual(text, "Draft content", "Boundary \(crashAfter)")
            XCTAssertEqual(count, 1, "Creation-token retry cannot duplicate the document")
            XCTAssertTrue(summary.isEmpty)
            let reopened = CRDTReplica(guid: detail.guid)
            let lane = DocsPersistence(store: frozen, outbox: recovered.outbox, key: "draft-token", replica: reopened)
            try await lane.reopen(fallback: Data(), sequence: 0)
            let offline = try await reopened.projection().children.map(\.plainText).joined()
            XCTAssertEqual(offline, "Draft content")
            await reopened.close(); recovered.drainer.stop()
        }
        XCTAssertLessThanOrEqual(limit, 80, "Expand the exhaustive sweep if attachment grows")
    }

    func testDirectDocumentOpenReportsLostKeyBeforeReadingItsIndex() async throws {
        let server = IntegratingServer(); try await server.open(initial())
        let root = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
        defer { try? FileManager.default.removeItem(at: root) }
        let keys = ScriptedKeyProvider(key: SymmetricKey(size: .bits256))
        let disk = NativeSealedStore(root: root, accountID: "a", keys: keys)
        let network = NativeConnectivity(source: ScriptedPathSource(.unsatisfied))
        let old = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { server })
        let detail = DocumentDetail(sqid: "doc", title: "Document", guid: "offline-property", can_write: true, can_share: false)
        try old.saveMetadata(DocsOfflineMetadata(detail: detail), key: "doc")
        try await old.outbox.enqueue(lane: "doc:doc", kind: "update", payload: Data([0]))
        _ = keys.deleteKey()
        // A process restart also drops NativeSealedStore's cached key/state.
        // Reusing the old store would deliberately keep the still-usable key.
        let reopened = NativeSealedStore(root: root, accountID: "a", keys: keys)
        let next = DocsOfflineAccount(store: reopened, accountID: "a", connectivity: network, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: network, offlineAccount: next)
        await model.open("doc")
        try await Task.sleep(for: .milliseconds(50))
        XCTAssertNotNil(model.projection)
        XCTAssertEqual(model.lostChangesNotice, NativeSyncStrings.kit.unsentLost)
        await model.discardAll(); next.drainer.stop()
    }

    func testDraftSeedSurvivesMetadataWriteFailureAndConvergesAfterUnlock() async throws {
        let server = IntegratingServer(); try await server.open(Data())
        let disk = MetadataLockStore(), network = NativeConnectivity(source: ScriptedPathSource(.satisfied))
        let account = DocsOfflineAccount(store: disk, accountID: "a", connectivity: network, transport: { server })
        let model = DocsModel(transport: { server }, connectivity: network, draft: true, offlineAccount: account)
        await model.newDraft()
        XCTAssertTrue(model.inputPaused)
        disk.unlock(); await model.retry()
        XCTAssertNotNil(model.projection)
        model.didRender(model.projectionGeneration)
        XCTAssertTrue(model.enqueue(["op": .string("display_edit"), "at": .number(0), "length": .number(0), "text": .string("Recovered draft")]))
        while model.hasPendingInput { try await Task.sleep(for: .milliseconds(5)) }
        let saved = await model.saveAll(); XCTAssertTrue(saved)
        let text = try await server.text()
        XCTAssertEqual(text, "Recovered draft", "The paragraph dependency must reach the server too")
        await model.discardAll(); account.drainer.stop()
    }

}