import Foundation
import Testing
import NucleicProtocol

@testable import NucleicCore

/// Mesh P5 — full session-transfer round-trip: a source `SessionTransferCoordinator` drives a
/// destination `SessionTransferImporter` over an in-memory loopback channel, moving a real git
/// branch (bundle) + transcript between two temp repos, with real GRDB stores on each side.
@Suite struct SessionTransferTests {
    let now = Date(timeIntervalSince1970: 1_700_000_000)

    /// An in-memory channel wiring a coordinator's `ClientMsg` sends straight into an importer and
    /// feeding the importer's `HostMsg` replies back as `TransferReply`s — the loopback the two
    /// real sync endpoints would otherwise bridge.
    final class LoopbackChannel: TransferChannel, @unchecked Sendable {
        private let importer: SessionTransferImporter
        private let sourceDeviceID: String
        private let stream: AsyncStream<TransferReply>
        private let continuation: AsyncStream<TransferReply>.Continuation
        /// Set true to drop the channel after N sends (simulate the peer going away).
        var dropAfterSends: Int?
        /// Drop the connection exactly when the source sends `transferCommit` — modelling a crash
        /// after the source tombstones but before it hears the destination's committed ack.
        var dropOnCommit = false
        /// Drop `transferCommit` *before* the destination processes it — modelling the DESTINATION
        /// crashing right as commit arrives, so it stays stranded at `.ready` (staged, not activated).
        var swallowCommit = false
        /// How many transcript chunks the source actually streamed — 0 when the destination
        /// adopted its mesh mirror (`alreadyHaveItems`) and the coordinator skipped the item.
        private(set) var transcriptChunkSends = 0
        private var sendCount = 0

        init(importer: SessionTransferImporter, sourceDeviceID: String) {
            self.importer = importer
            self.sourceDeviceID = sourceDeviceID
            (stream, continuation) = AsyncStream.makeStream()
        }

        func send(_ msg: ClientMsg) async {
            sendCount += 1
            if case .transferChunk(let chunk) = msg, chunk.kind == .transcript {
                transcriptChunkSends += 1
            }
            if swallowCommit, case .transferCommit = msg {
                // The destination never processes the commit (it crashed) — stranded at `.ready`.
                continuation.yield(.disconnected("peer offline"))
                return
            }
            if dropOnCommit, case .transferCommit = msg {
                // The importer still commits (it received the verb) but the source never hears back.
                _ = await importer.receiveCommit({ if case .transferCommit(let id) = msg { return id }; return "" }())
                continuation.yield(.disconnected("peer offline"))
                return
            }
            if let drop = dropAfterSends, sendCount > drop {
                continuation.yield(.disconnected("peer offline"))
                return
            }
            let replies: [HostMsg]
            switch msg {
            case .transferOffer(let offer):
                replies = await importer.receiveOffer(offer, from: sourceDeviceID)
            case .transferChunk(let chunk):
                replies = await importer.receiveChunk(chunk)
            case .transferCommit(let id):
                replies = await importer.receiveCommit(id)
            case .transferCancel(let id):
                await importer.receiveCancel(id); replies = []
            default:
                replies = []
            }
            for reply in replies { continuation.yield(map(reply)) }
        }

        func replies() -> AsyncStream<TransferReply> { stream }

        private func map(_ msg: HostMsg) -> TransferReply {
            switch msg {
            case .transferAccept(let a): .accept(a)
            case .transferReject(let r): .reject(r)
            case .transferReady(let id): .ready(id)
            case .transferCommitted(let id): .committed(id)
            case .transferChunkAck(let ack): .chunkAck(ack)
            default: .disconnected("unexpected")
            }
        }
    }

    /// A destination environment backed by a single resolvable project + a temp transcripts dir.
    struct FakeImportEnvironment: TransferImportEnvironment {
        let project: Project?
        let transcriptsDir: URL
        let nativeRoot: URL?
        let onActivate: @Sendable (Session) async -> Void
        /// When set, the destination "holds a current mesh mirror" of any offered transcript.
        var mirrorURL: URL? = nil

        func resolveProject(_ descriptor: ProjectDescriptor) async -> Project? { project }
        func nativeTranscriptDestination(
            session: Session, project: Project, worktreePath: String, backendSessionID: String
        ) -> URL? {
            nativeRoot?.appendingPathComponent("\(backendSessionID).jsonl")
        }
        func didActivateTransferredSession(_ session: Session) async { await onActivate(session) }
        func mirroredTranscript(
            sessionID: SessionID, from deviceID: String, minSeq: UInt64
        ) async -> URL? { mirrorURL }
    }

    /// Build a source repo with a session worktree that has committed + uncommitted work, plus a
    /// transcript file, returning everything the coordinator needs.
    private func makeSource() async throws -> (
        repo: GitTestRepo, wm: GitWorktreeManager, project: Project, worktree: Worktree,
        session: Session, transcript: URL, tip: String, store: GRDBMetadataStore, stagingRoot: URL
    ) {
        let repo = try await GitTestRepo()
        let wm = GitWorktreeManager()
        let project = repo.project()
        let wt = try await wm.create(for: SessionID(rawValue: "s1"), in: project, base: "main", slug: "feature")
        try repo.write("code.txt", "committed work\n", in: wt.path)
        try await repo.run(["add", "-A"], in: wt.path)
        try await repo.run(["commit", "-m", "work"], in: wt.path)
        try await wm.finalize(wt, commit: .auto(message: "nucleic: transfer snapshot"), onCommit: nil)
        let tip = try await repo.revParse("HEAD", in: wt.path)

        // A transcript file with a header + a couple events.
        let transcript = URL(fileURLWithPath: repo.container)
            .appendingPathComponent("transcript.jsonl")
        let header = SessionHeader(
            sessionID: SessionID(rawValue: "s1"), backend: .claudeCode,
            backendSessionID: "native-1", worktree: wt.path, model: "claude-opus-4-8",
            createdAt: now)
        let writer = try TranscriptWriter(url: transcript, header: header)
        try await writer.append(AgentEvent(
            sessionID: SessionID(rawValue: "s1"), seq: 0, at: now, backend: .claudeCode,
            nativeType: nil, kind: .assistantText(TextChunk(messageID: "m", text: "hello", isPartial: false))))
        try await writer.close()

        let session = Session(
            id: SessionID(rawValue: "s1"), projectID: project.id, backend: .claudeCode,
            backendSessionID: "native-1", title: "feature work", status: .awaitingInput,
            worktreePath: wt.path, branch: wt.branch, baseSHA: wt.baseSHA,
            model: "claude-opus-4-8", effort: "high", lastSeq: 1,
            transcriptPath: transcript.path, createdAt: now, updatedAt: now)

        let store = try GRDBMetadataStore(path: nil)
        try await store.saveProject(project)
        try await store.saveSession(session)

        let stagingRoot = URL(fileURLWithPath: repo.container).appendingPathComponent("xfer-src")
        return (repo, wm, project, wt, session, transcript, tip, store, stagingRoot)
    }

    /// Build a destination: a clone of the source repo (taken before the branch) + its own store.
    private func makeDest(cloning sourceRoot: String) async throws -> (
        project: Project, wm: GitWorktreeManager, store: GRDBMetadataStore,
        transcriptsDir: URL, stagingRoot: URL, container: String
    ) {
        let container = (NSTemporaryDirectory() as NSString)
            .appendingPathComponent("nucleic-dest-\(UUID().uuidString)")
        let destRoot = (container as NSString).appendingPathComponent("repo")
        try FileManager.default.createDirectory(atPath: container, withIntermediateDirectories: true)
        try await GitRunner().checked(["clone", sourceRoot, destRoot], in: container)
        let project = Project(id: .generate(), name: "dest", rootPath: destRoot, defaultBranch: "main", createdAt: now)
        let store = try GRDBMetadataStore(path: nil)
        try await store.saveProject(project)
        return (
            project, GitWorktreeManager(), store,
            URL(fileURLWithPath: container).appendingPathComponent("sessions"),
            URL(fileURLWithPath: container).appendingPathComponent("xfer-dest"),
            container)
    }

    private func descriptor(for project: Project, rootCommit: String) -> ProjectDescriptor {
        ProjectDescriptor(projectID: project.id, name: project.name,
                          rootCommitSHA: rootCommit, normalizedRemote: nil,
                          defaultBranch: project.defaultBranch.value)
    }

    @Test func happyPathMovesTheSessionAndTombstonesTheSource() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")  // base = root commit for a fresh repo

        // NOTE: the destination clone must be taken BEFORE the feature branch — makeSource already
        // committed the branch, so re-clone from a pristine mirror. Instead, model the realistic
        // case with a prerequisite-free bundle by cloning now and letting haveSHAs negotiate.
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }

        let activated = ActivatedBox()
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil,
            onActivate: { await activated.set($0) })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env,
            stagingRoot: dest.stagingRoot, now: { self.now })

        let package = try SessionCarbonCopy.prepare(
            transferID: "xfer-1", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)

        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)

        try await coordinator.run(now: now)

        // Source: tombstoned (archived + moved), worktree discarded, transfer lock committed.
        let sourceSession = try await src.store.loadSession(id: src.session.id)
        #expect(sourceSession?.archived == true)
        #expect(sourceSession?.movedToDeviceID == "host-B")
        #expect(sourceSession?.movedAt == now)
        #expect(!FileManager.default.fileExists(atPath: src.worktree.path))  // discarded
        #expect(try await src.store.activeTransfer(sessionID: src.session.id) == nil)

        // Destination: the session is live, at the destination worktree, with the branch checked out.
        let destSession = try await dest.store.loadSession(id: src.session.id)
        #expect(destSession != nil)
        #expect(destSession?.archived != true)
        #expect(destSession?.projectID == dest.project.id)
        #expect(destSession?.branch == "nucleic/feature")
        // No native transcript was sent, so memory-carry doesn't claim it: fresh context.
        #expect(destSession?.backendSessionID == nil)
        // Provenance (mesh P5): the imported session records where it arrived from (real GRDB v24
        // round-trip), so the destination can show "Arrived from <Mac>".
        #expect(destSession?.arrivedFromDeviceID == "host-A")
        #expect(destSession?.arrivedAt == now)
        let activatedSession = await activated.value
        #expect(activatedSession?.id == src.session.id)
        // The imported worktree exists with the committed file.
        if let wtPath = destSession?.worktreePath {
            #expect(FileManager.default.fileExists(atPath: (wtPath as NSString).appendingPathComponent("code.txt")))
        } else { Issue.record("no dest worktree path") }
        // The transcript was placed + its header repointed at the destination worktree.
        if let tPath = destSession?.transcriptPath {
            let (header, events) = try TranscriptReader(url: URL(fileURLWithPath: tPath)).read()
            #expect(header.worktree == destSession?.worktreePath)
            #expect(events.count == 1)
        } else { Issue.record("no dest transcript") }
    }

    /// Covalence: the management marker (`covalenceOriginDeviceID`) rides the transfer record and lands
    /// on the destination session, so the new owner keeps rebalancing the chat between turns.
    @Test func covalenceManagementRidesTheTransfer() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        var session = src.session
        session.covalenceOriginDeviceID = "origin-phone"
        try await src.store.saveSession(session)

        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let activated = ActivatedBox()
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil,
            onActivate: { await activated.set($0) })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env,
            stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "xfer-covalence", session: session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        try await coordinator.run(now: now)

        let destSession = try await dest.store.loadSession(id: session.id)
        #expect(destSession?.covalenceOriginDeviceID == "origin-phone")
    }

    /// Mesh-mirror fast path: a destination holding a current transcript mirror adopts it —
    /// the accept claims the item, the source streams ZERO transcript chunks, and the imported
    /// session still lands with its transcript placed + header repointed.
    @Test func mirrorAdoptionSkipsTranscriptStreaming() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")

        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        // The destination's "mesh mirror" of the session — same content the source would stream.
        let mirror = URL(fileURLWithPath: dest.container).appendingPathComponent("mirror.jsonl")
        try FileManager.default.copyItem(at: src.transcript, to: mirror)

        let activated = ActivatedBox()
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil,
            onActivate: { await activated.set($0) }, mirrorURL: mirror)
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env,
            stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "xfer-mirror", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        try await coordinator.run(now: now)

        // The transcript never crossed the channel…
        #expect(channel.transcriptChunkSends == 0)
        // …yet the destination session landed complete, transcript placed + header repointed.
        let destSession = try await dest.store.loadSession(id: src.session.id)
        #expect(destSession != nil)
        if let tPath = destSession?.transcriptPath {
            let (header, events) = try TranscriptReader(url: URL(fileURLWithPath: tPath)).read()
            #expect(header.worktree == destSession?.worktreePath)
            #expect(events.count == 1)
        } else { Issue.record("no dest transcript") }
    }

    @Test func projectNotFoundRejectsAndLeavesSourceIntact() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }

        // Destination resolves NO project.
        let env = FakeImportEnvironment(
            project: nil, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "x", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)

        await #expect(throws: SessionTransferError.rejected(.projectNotFound, message: TransferRejectReason.projectNotFound.message)) {
            try await coordinator.run(now: now)
        }
        // Source untouched: not archived, worktree present, lock cleared for a retry.
        let sourceSession = try await src.store.loadSession(id: src.session.id)
        #expect(sourceSession?.archived != true)
        #expect(sourceSession?.movedToDeviceID == nil)
        #expect(FileManager.default.fileExists(atPath: src.worktree.path))
        #expect(try await src.store.activeTransfer(sessionID: src.session.id) == nil)
    }

    @Test func versionMismatchRejects() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        // A transcript header version beyond what the destination supports.
        let package = try SessionCarbonCopy.prepare(
            transferID: "x", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.maxSupported + 1,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        await #expect(throws: SessionTransferError.rejected(.versionMismatch, message: TransferRejectReason.versionMismatch.message)) {
            try await coordinator.run(now: now)
        }
        #expect(try await dest.store.loadSession(id: src.session.id) == nil)  // nothing staged/inserted
    }

    @Test func commitIsIdempotent() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "x", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        try await coordinator.run(now: now)

        // A re-delivered commit (idempotency key = transferID) just re-acks; no duplicate row.
        let secondCommit = await importer.receiveCommit("x")
        #expect(secondCommit == [.transferCommitted("x")])
        #expect(try await dest.store.loadSessions(projectID: dest.project.id).count == 1)
    }

    @Test func doubleTransferOfSameSessionIsRejectedByTheLock() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        // Claim an active transfer for the session first.
        try await src.store.beginTransfer(SessionTransferLock(
            transferID: "already", sessionID: src.session.id, direction: .outbound,
            peerDeviceID: "host-B", state: .streaming, createdAt: now, updatedAt: now))

        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let rootCommit = try await src.repo.revParse("HEAD")
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "xfer-2", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        await #expect(throws: SessionTransferError.alreadyTransferring(src.session.id)) {
            try await coordinator.run(now: now)
        }
    }

    @Test func committedButUnconfirmedKeepsSourceMovedAndLockHeld() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "x", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        channel.dropOnCommit = true  // the source tombstones + sends commit, then the link drops
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)

        await #expect(throws: SessionTransferError.committedButUnconfirmed) {
            try await coordinator.run(now: now)
        }
        // Source is durably moved (never revived) and the lock stays at .tombstoned for recovery —
        // NOT cleared. The worktree is NOT discarded (recovery may need it).
        let sourceSession = try await src.store.loadSession(id: src.session.id)
        #expect(sourceSession?.archived == true)
        #expect(sourceSession?.movedToDeviceID == "host-B")
        #expect(try await src.store.transfer(transferID: "x")?.state == .tombstoned)
        #expect(FileManager.default.fileExists(atPath: src.worktree.path))
        // The destination did activate on the commit it received.
        #expect(try await dest.store.loadSession(id: src.session.id) != nil)
    }

    /// Relaunch recovery (mesh P5): a source left at `.tombstoned` by a lost commit ack finishes
    /// the 2-phase commit when the destination is reachable again — the re-sent commit is idempotent
    /// (the destination already activated), so the source discards its worktree and clears the lock.
    @Test func recoverTombstonedCompletesTheCommitOnReconnect() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "x", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        channel.dropOnCommit = true  // tombstone, commit sent + applied, but the ack never lands
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        await #expect(throws: SessionTransferError.committedButUnconfirmed) {
            try await coordinator.run(now: now)
        }
        // Precondition for recovery: lock held at .tombstoned, worktree preserved, dest activated.
        #expect(try await src.store.transfer(transferID: "x")?.state == .tombstoned)
        #expect(FileManager.default.fileExists(atPath: src.worktree.path))

        // Reconnect: a fresh channel to the same (already-activated) importer re-sends the commit.
        let reconnect = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let lock = try #require(await src.store.transfer(transferID: "x"))
        let session = try await src.store.loadSession(id: src.session.id)
        let done = await SessionTransferCoordinator.recoverTombstoned(
            lock: lock, session: session, project: src.project, channel: reconnect,
            store: src.store, worktrees: src.wm, now: now)
        #expect(done)
        // The transfer is complete: lock cleared, source worktree discarded.
        #expect(try await src.store.transfer(transferID: "x") == nil)
        #expect(!FileManager.default.fileExists(atPath: src.worktree.path))
    }

    /// If the destination *also* relaunched (losing its in-memory staging) before the source's
    /// recovery commit, the re-send is rejected — recovery reports "not done" and leaves the lock at
    /// `.tombstoned` (and the worktree intact) for a later attempt, never reviving the source.
    @Test func recoverTombstonedLeavesLockWhenDestinationLostStaging() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "x", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        channel.dropOnCommit = true
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        await #expect(throws: SessionTransferError.committedButUnconfirmed) {
            try await coordinator.run(now: now)
        }

        // Model a destination that relaunched: a brand-new importer over a fresh store has no staging
        // and no activated lock, so it rejects the recovery commit ("nothing staged to activate").
        let dest2 = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest2.container) }
        let env2 = FakeImportEnvironment(
            project: dest2.project, transcriptsDir: dest2.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer2 = SessionTransferImporter(
            store: dest2.store, worktrees: dest2.wm, environment: env2, stagingRoot: dest2.stagingRoot, now: { self.now })
        let reconnect = LoopbackChannel(importer: importer2, sourceDeviceID: "host-A")
        let lock = try #require(await src.store.transfer(transferID: "x"))
        let session = try await src.store.loadSession(id: src.session.id)
        let done = await SessionTransferCoordinator.recoverTombstoned(
            lock: lock, session: session, project: src.project, channel: reconnect,
            store: src.store, worktrees: src.wm, now: now)
        #expect(!done)
        // Nothing lost: lock still tombstoned, worktree preserved, source still inert (never revived).
        #expect(try await src.store.transfer(transferID: "x")?.state == .tombstoned)
        #expect(FileManager.default.fileExists(atPath: src.worktree.path))
        #expect(try await src.store.loadSession(id: src.session.id)?.movedToDeviceID == "host-B")
    }

    /// Drive a transfer to the destination's `.ready` state (staged + verified on disk) but never
    /// let it commit — modelling the destination crashing right as commit arrives. The source, past
    /// its tombstone, ends `committedButUnconfirmed`. Returns the destination context + transferID so
    /// a "relaunched" importer can recover the stranded staging.
    private func strandInboundAtReady(transferID: String) async throws -> (
        src: (repo: GitTestRepo, wm: GitWorktreeManager, project: Project, worktree: Worktree,
              session: Session, transcript: URL, tip: String, store: GRDBMetadataStore, stagingRoot: URL),
        dest: (project: Project, wm: GitWorktreeManager, store: GRDBMetadataStore,
               transcriptsDir: URL, stagingRoot: URL, container: String)
    ) {
        let src = try await makeSource()
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: transferID, session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nil,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        channel.swallowCommit = true  // the destination "crashes" as commit arrives
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        await #expect(throws: SessionTransferError.committedButUnconfirmed) {
            try await coordinator.run(now: now)
        }
        // The destination staged + verified but never activated: no session row, lock at `.ready`.
        #expect(try await dest.store.loadSession(id: src.session.id) == nil)
        #expect(try await dest.store.transfer(transferID: transferID)?.state == .ready)
        return (src, dest)
    }

    @Test func recoverAndActivateStrandedInboundAfterDestRelaunch() async throws {
        let (src, dest) = try await strandInboundAtReady(transferID: "x")
        defer { src.repo.cleanup(); try? FileManager.default.removeItem(atPath: dest.container) }

        // Relaunch: a fresh importer over the same store + stagingRoot (the in-memory staging is gone,
        // but the manifest + imported worktree survive on disk).
        let activated = ActivatedBox()
        let env2 = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil,
            onActivate: { await activated.set($0) })
        let importer2 = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env2, stagingRoot: dest.stagingRoot, now: { self.now })

        // The stranded transfer is recoverable, rebuilt from the manifest.
        let recoverable = await importer2.recoverableInboundTransfers()
        #expect(recoverable.count == 1)
        #expect(recoverable.first?.session.id == src.session.id)
        #expect(recoverable.first?.sourceDeviceID == "host-A")
        #expect(recoverable.first?.session.arrivedFromDeviceID == "host-A")

        // Activate anyway: the session goes live, provenance intact, lock terminal, staging gone.
        let activatedSession = try await importer2.activateRecoveredTransfer("x")
        #expect(activatedSession?.id == src.session.id)
        let destSession = try await dest.store.loadSession(id: src.session.id)
        #expect(destSession != nil)
        #expect(destSession?.arrivedFromDeviceID == "host-A")
        #expect(destSession?.branch == "nucleic/feature")
        #expect(await activated.value?.id == src.session.id)
        #expect(try await dest.store.transfer(transferID: "x")?.state == .activated)
        #expect(await importer2.recoverableInboundTransfers().isEmpty)
    }

    @Test func discardStrandedInboundClearsStagingAndLock() async throws {
        let (src, dest) = try await strandInboundAtReady(transferID: "y")
        defer { src.repo.cleanup(); try? FileManager.default.removeItem(atPath: dest.container) }

        let env2 = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer2 = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env2, stagingRoot: dest.stagingRoot, now: { self.now })
        #expect(await importer2.recoverableInboundTransfers().count == 1)

        // Discard clears the importer's staging dir + the lock; no session is ever inserted.
        await importer2.clearInboundStaging("y")
        #expect(try await dest.store.transfer(transferID: "y") == nil)
        #expect(try await dest.store.loadSession(id: src.session.id) == nil)
        #expect(await importer2.recoverableInboundTransfers().isEmpty)
        #expect(!FileManager.default.fileExists(
            atPath: dest.stagingRoot.appendingPathComponent("y").path))
    }

    @Test func importerRejectsPathTraversalIdentifiers() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }
        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nil, onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })

        func offer(sessionID: String, branch: String, transferID: String = "x", backendSessionID: String? = nil) -> TransferOffer {
            TransferOffer(
                transferID: transferID,
                record: SessionTransferRecord(
                    sessionID: SessionID(rawValue: sessionID), backend: .claudeCode, backendSessionID: backendSessionID,
                    title: "t", branch: branch, baseSHA: rootCommit, model: nil, effort: nil,
                    lastSeq: 0, createdAt: now),
                project: descriptor(for: src.project, rootCommit: rootCommit),
                transcriptHeaderVersion: SessionHeader.currentVersion,
                availableBaseSHAs: [rootCommit], knownItems: [])
        }
        // A traversal sessionID, a non-nucleic branch, a traversal transferID, and a traversal
        // backendSessionID are all rejected BEFORE any filesystem work or lock claim.
        for bad in [
            offer(sessionID: "../../etc/evil", branch: "nucleic/ok"),
            offer(sessionID: "s1", branch: "../../evil"),
            offer(sessionID: "s1", branch: "refs/heads/main"),
            offer(sessionID: "s1", branch: "nucleic/ok", transferID: "../../x"),
            offer(sessionID: "s1", branch: "nucleic/ok", backendSessionID: "../../../../evil"),
        ] {
            let replies = await importer.receiveOffer(bad, from: "host-A")
            guard case .transferReject(let reject) = replies.first else {
                Issue.record("expected reject for \(bad.record.sessionID)/\(bad.record.branch)"); continue
            }
            #expect(reject.reason == .verificationFailed)
        }
        // No lock was claimed and no staging dir created for the rejected offers.
        #expect(try await dest.store.allActiveTransfers().isEmpty)
        let stagedEntries = (try? FileManager.default.contentsOfDirectory(atPath: dest.stagingRoot.path)) ?? []
        #expect(stagedEntries.isEmpty)
    }

    @Test func safePathComponentRules() {
        #expect(SessionTransferImporter.isSafePathComponent("s-abc123_DEF"))
        #expect(!SessionTransferImporter.isSafePathComponent(""))
        #expect(!SessionTransferImporter.isSafePathComponent(".."))
        #expect(!SessionTransferImporter.isSafePathComponent("../x"))
        #expect(!SessionTransferImporter.isSafePathComponent("a/b"))
        #expect(!SessionTransferImporter.isSafePathComponent(".hidden"))
        #expect(!SessionTransferImporter.isSafePathComponent("-flag"))
        #expect(SessionTransferImporter.isSafeBranch("nucleic/feature-1"))
        #expect(!SessionTransferImporter.isSafeBranch("feature"))
        #expect(!SessionTransferImporter.isSafeBranch("nucleic/../evil"))
    }

    @Test func memoryCarryPlacesNativeTranscriptWhenSupported() async throws {
        let src = try await makeSource()
        defer { src.repo.cleanup() }
        let rootCommit = try await src.repo.revParse("HEAD")
        let dest = try await makeDest(cloning: src.repo.root)
        defer { try? FileManager.default.removeItem(atPath: dest.container) }

        // Provide a source native transcript file to carry.
        let nativeSrc = URL(fileURLWithPath: src.repo.container).appendingPathComponent("native.jsonl")
        try #"{"type":"session","id":"native-1"}"#.write(to: nativeSrc, atomically: true, encoding: .utf8)
        let nativeRoot = URL(fileURLWithPath: dest.container).appendingPathComponent("claude-home-native")

        let env = FakeImportEnvironment(
            project: dest.project, transcriptsDir: dest.transcriptsDir, nativeRoot: nativeRoot,
            onActivate: { _ in })
        let importer = SessionTransferImporter(
            store: dest.store, worktrees: dest.wm, environment: env, stagingRoot: dest.stagingRoot, now: { self.now })
        let package = try SessionCarbonCopy.prepare(
            transferID: "x", session: src.session, project: src.project,
            projectDescriptor: descriptor(for: src.project, rootCommit: rootCommit),
            worktree: src.worktree, transcriptURL: src.transcript, nativeTranscriptURL: nativeSrc,
            branchTipSHA: src.tip, transcriptHeaderVersion: SessionHeader.currentVersion,
            stagingRoot: src.stagingRoot, worktrees: src.wm)
        let channel = LoopbackChannel(importer: importer, sourceDeviceID: "host-A")
        let coordinator = SessionTransferCoordinator(
            session: src.session, project: src.project, worktree: src.worktree,
            destinationDeviceID: "host-B", package: package, channel: channel,
            store: src.store, worktrees: src.wm)
        try await coordinator.run(now: now)

        let destSession = try await dest.store.loadSession(id: src.session.id)
        // Memory carried: backendSessionID kept and the native file placed at the dest path.
        #expect(destSession?.backendSessionID == "native-1")
        let expectedNative = nativeRoot.appendingPathComponent("native-1.jsonl")
        #expect(FileManager.default.fileExists(atPath: expectedNative.path))
    }
}

/// Records the activated session across an actor boundary.
actor ActivatedBox {
    private(set) var value: Session?
    func set(_ s: Session) { value = s }
}
