import Foundation
import NucleicProtocol

@testable import NucleicCore

/// In-memory bidirectional frame pipe — a `socketpair` for `FrameChannel`. Each endpoint's
/// `send` delivers a whole frame to the peer's inbound stream (modeling post-framing
/// delivery), so the full Noise + CBOR path is exercised without real TCP.
final class MemoryChannel: FrameChannel, @unchecked Sendable {
    let peerDescription: String
    /// The transport this channel reports — mirrors how the real host-accepted channels self-identify
    /// (`LANChannel` → `.lan`, …). `nil` by default so existing tests exercise the "unknown transport"
    /// path; a test that drives the LAN-drop reconnect nudge pairs with `.lan`.
    let transport: SyncTransportHint?
    private let stream: AsyncStream<Data>
    private let continuation: AsyncStream<Data>.Continuation
    private var peer: MemoryChannel?

    private init(name: String, transport: SyncTransportHint?) {
        peerDescription = name
        self.transport = transport
        var cont: AsyncStream<Data>.Continuation!
        stream = AsyncStream(bufferingPolicy: .unbounded) { cont = $0 }
        continuation = cont
    }

    func frames() -> AsyncStream<Data> { stream }
    func send(_ frame: Data) { peer?.continuation.yield(frame) }
    func close() { continuation.finish(); peer?.continuation.finish() }

    static func pair(transport: SyncTransportHint? = nil) -> (client: MemoryChannel, server: MemoryChannel) {
        let a = MemoryChannel(name: "client", transport: transport)
        let b = MemoryChannel(name: "server", transport: transport)
        a.peer = b
        b.peer = a
        return (a, b)
    }
}

/// Yields a single, already-accepted connection (the in-memory server endpoint).
final class OneShotListener: SyncListener, @unchecked Sendable {
    private let connection: any FrameChannel
    init(_ connection: any FrameChannel) { self.connection = connection }

    func start() throws -> AsyncStream<any FrameChannel> {
        let conn = connection
        return AsyncStream { continuation in
            continuation.yield(conn)
            continuation.finish()
        }
    }
    func stop() {}
}

/// A listener the test pushes connections into on demand (for multi-connection scenarios
/// like reconnect). `start()` runs synchronously inside `SyncHost.start`, so the
/// continuation is set before any `push`.
final class ManualListener: SyncListener, @unchecked Sendable {
    private var continuation: AsyncStream<any FrameChannel>.Continuation?
    func start() throws -> AsyncStream<any FrameChannel> {
        AsyncStream { self.continuation = $0 }
    }
    func push(_ connection: any FrameChannel) { continuation?.yield(connection) }
    func stop() { continuation?.finish() }
}

/// Scriptable host authority for sync tests: fixed session list/snapshot, a broadcast
/// channel, and recorded approval/input/interrupt calls.
actor FakeSyncBridge: SyncHostBridge {
    let hostInfo: HostInfo
    let capabilities: WireCapabilities
    let modelCatalog: WireModelCatalog
    let releaseChannel: ReleaseChannel?
    private let localAddressesValue: PeerAddresses?

    private var sessions: [WireSessionSummary]
    private var snapshots: [SessionID: SessionSnapshot]
    private var peers: [PeerSummary]
    private var broadcastContinuation: AsyncStream<HostBroadcast>.Continuation?

    private(set) var respondedApprovals: [(ApprovalID, Decision, String)] = []
    private(set) var sentInputs: [(SessionID, AgentInput)] = []
    private(set) var interrupts: [SessionID] = []
    var failApprovalWith: WireError?

    // Mesh P4 remainder instrumentation.
    private var approveMacPairingResponse = true
    private(set) var macPairingRequests: [(deviceID: String, label: String)] = []
    private(set) var peerSummariesCalls = 0
    private(set) var meshPeersChangedCount = 0

    /// When set, this bridge serves Carbon replication from a real `CarbonShardStore`
    /// (CARBON_SHARDING §8.1) — pair with a `capabilities` that advertises `canServeCarbon`.
    private let carbonStore: CarbonShardStore?
    /// When set, this bridge arbitrates the Carbon lock verbs (§17.5) as a lock authority —
    /// the same thin mapping `AppStore`'s conformance performs.
    private let lockManager: CarbonLockManager?

    init(
        hostInfo: HostInfo = HostInfo(hostID: "host", hostName: "Andrew's Mac"),
        capabilities: WireCapabilities = WireCapabilities(canModifyToolInput: true, allowAlwaysScopes: [.session, .toolName]),
        modelCatalog: WireModelCatalog = .empty,
        sessions: [WireSessionSummary] = [],
        snapshots: [SessionID: SessionSnapshot] = [:],
        peers: [PeerSummary] = [],
        releaseChannel: ReleaseChannel? = nil,
        localAddresses: PeerAddresses? = nil,
        carbonStore: CarbonShardStore? = nil,
        lockManager: CarbonLockManager? = nil
    ) {
        self.hostInfo = hostInfo
        self.capabilities = capabilities
        self.modelCatalog = modelCatalog
        self.releaseChannel = releaseChannel
        self.localAddressesValue = localAddresses
        self.sessions = sessions
        self.snapshots = snapshots
        self.peers = peers
        self.carbonStore = carbonStore
        self.lockManager = lockManager
    }

    // Carbon replication serving (CARBON_SHARDING §8.1) — real store when provided, else the
    // protocol defaults' empty answers.
    func carbonHeads(projectUUID: UUID?) async -> [CarbonHead] {
        guard let carbonStore, let projectUUID else { return [] }
        return Array(await carbonStore.heads(projectUUID: projectUUID).values)
    }
    func carbonManifestBlob(id: String, projectUUID: UUID) async -> Data? {
        await carbonStore?.manifestBlob(id: id, projectUUID: projectUUID)
    }
    func carbonShardBlob(id: String, projectUUID: UUID) async -> Data? {
        await carbonStore?.shard(id: id, projectUUID: projectUUID)
    }

    // Carbon lock arbitration (§17.5) — the AppStore-conformance mapping, against the
    // injected manager. Absent manager ⇒ the protocol defaults (never-block / cancelled).
    func carbonLockWouldBlock(_ request: CarbonLockRequest, from deviceID: String) async -> Bool {
        guard let lockManager else { return false }
        return await lockManager.wouldBlock(
            SessionID(rawValue: request.sessionID), domain: LockDomain(request.domainRef),
            files: request.paths ?? [])
    }

    func carbonLockAcquire(_ request: CarbonLockRequest, from deviceID: String) async
        -> CarbonLockDecisionMsg.Outcome
    {
        guard let lockManager, let paths = request.paths, !paths.isEmpty else { return .cancelled }
        switch await lockManager.acquire(
            SessionID(rawValue: request.sessionID), domain: LockDomain(request.domainRef),
            files: paths)
        {
        case .granted, .grantedNeedsReground, .updateConflict: return .granted
        case .cancelled: return .cancelled
        }
    }

    func carbonLockRelease(_ request: CarbonLockRequest, from deviceID: String) async {
        guard let lockManager else { return }
        let sid = SessionID(rawValue: request.sessionID)
        let heads = request.requiredHeads ?? []
        if let paths = request.paths {
            await lockManager.release(sid, paths: paths, requiredHeads: heads)
        } else {
            await lockManager.releaseAll(sid, requiredHeads: heads)
        }
    }

    func sessionSummaries() async -> [WireSessionSummary] { sessions }
    func snapshot(_ id: SessionID, sinceSeq: UInt64?) async -> SessionSnapshot? { snapshots[id] }
    func peerSummaries() async -> [PeerSummary] {
        peerSummariesCalls += 1
        return peers
    }

    var localAddresses: PeerAddresses? { localAddressesValue }

    func approveMacPairing(deviceID: String, label: String) async -> Bool {
        macPairingRequests.append((deviceID, label))
        return approveMacPairingResponse
    }
    func setApproveMacPairing(_ approve: Bool) { approveMacPairingResponse = approve }

    func meshPeersChanged() async { meshPeersChangedCount += 1 }

    // Mesh "join" roster-gossip instrumentation.
    private var rosterToServe = MeshRosterPush()
    private(set) var receivedRosters: [(push: MeshRosterPush, from: String)] = []
    func setRosterToServe(_ roster: MeshRosterPush) { rosterToServe = roster }
    func meshRoster() async -> MeshRosterPush { rosterToServe }
    func mergeMeshRoster(_ push: MeshRosterPush, from deviceID: String) async {
        receivedRosters.append((push, deviceID))
    }

    func respondToApproval(_ id: ApprovalID, _ decision: Decision, by responder: String) async throws {
        if let failApprovalWith { throw failApprovalWith }
        respondedApprovals.append((id, decision, responder))
    }
    func sendInput(_ id: SessionID, _ input: AgentInput) async throws { sentInputs.append((id, input)) }
    func interrupt(_ id: SessionID) async throws { interrupts.append(id) }
    private(set) var cancelledQueued: [(SessionID, UUID)] = []
    func cancelQueuedMessage(_ id: SessionID, _ messageID: UUID) async throws {
        cancelledQueued.append((id, messageID))
    }

    // Control-scope stubs that record calls.
    private(set) var startedChats: [StartChatRequest] = []
    private(set) var capturedTodos: [CaptureTodoRequest] = []
    private(set) var dispatchedTodos: [(TodoID, ProjectID)] = []
    private(set) var todoStatuses: [(TodoID, TodoStatus)] = []
    private(set) var deletedTodos: [TodoID] = []
    private(set) var renamedSessions: [(SessionID, String)] = []
    private(set) var favorites: [(SessionID, Bool)] = []
    private(set) var archives: [(SessionID, Bool)] = []
    private(set) var markedDone: [SessionID] = []
    private(set) var deletedSessions: [SessionID] = []
    private(set) var integrations: [(SessionID, IntegrationMode)] = []
    private(set) var discards: [SessionID] = []
    private(set) var sessionModels: [(SessionID, String?)] = []
    private(set) var sessionEfforts: [(SessionID, String?)] = []
    private(set) var sessionAutos: [(SessionID, Bool)] = []
    private(set) var sessionAutoShips: [(SessionID, Bool)] = []
    private(set) var sessionShipBranches: [(SessionID, String?)] = []
    var dashboard = DashboardSnapshot.empty

    func dashboardSnapshot() async -> DashboardSnapshot { dashboard }
    func startChat(_ request: StartChatRequest) async throws { startedChats.append(request) }
    func captureTodo(_ request: CaptureTodoRequest) async throws { capturedTodos.append(request) }
    func dispatchTodo(_ id: TodoID, in projectID: ProjectID) async throws { dispatchedTodos.append((id, projectID)) }
    func remoteSetTodoStatus(_ id: TodoID, _ status: TodoStatus) async throws { todoStatuses.append((id, status)) }
    func remoteDeleteTodo(_ id: TodoID) async throws { deletedTodos.append(id) }
    func renameSession(_ id: SessionID, to title: String) async throws { renamedSessions.append((id, title)) }
    func setFavorite(_ id: SessionID, _ favorite: Bool) async throws { favorites.append((id, favorite)) }
    func setArchived(_ id: SessionID, _ archived: Bool) async throws { archives.append((id, archived)) }
    func markSessionDoneRemote(_ id: SessionID) async throws { markedDone.append(id) }
    func deleteSessionRemote(_ id: SessionID) async throws { deletedSessions.append(id) }
    func integrate(_ id: SessionID, _ mode: IntegrationMode) async throws { integrations.append((id, mode)) }
    func discard(_ id: SessionID) async throws { discards.append(id) }
    func setSessionModel(_ id: SessionID, _ model: String?) async throws { sessionModels.append((id, model)) }
    func setSessionEffort(_ id: SessionID, _ effort: String?) async throws { sessionEfforts.append((id, effort)) }
    func setSessionAuto(_ id: SessionID, _ auto: Bool) async throws { sessionAutos.append((id, auto)) }
    func setSessionAutoShip(_ id: SessionID, _ autoShip: Bool) async throws { sessionAutoShips.append((id, autoShip)) }
    func setSessionShipBranch(_ id: SessionID, _ branch: String?) async throws { sessionShipBranches.append((id, branch)) }

    // Intelligence delegation instrumentation (COVALENCE_RUNNER §5, item 6).
    private(set) var intelligenceWorkers:
        [(deviceID: String, connectionID: String, profile: IntelligenceWorkerProfile)] = []
    private(set) var intelligenceWorkerDisconnects: [(deviceID: String, connectionID: String)] = []
    private(set) var intelligenceResults: [(result: WireIntelligenceResult, from: String)] = []
    private var intelligenceSends: [String: @Sendable (WireIntelligenceRequest) async -> Void] = [:]

    func intelligenceWorkerConnected(
        deviceID: String, connectionID: String, profile: IntelligenceWorkerProfile,
        send: @escaping @Sendable (WireIntelligenceRequest) async -> Void
    ) async {
        intelligenceWorkers.append((deviceID, connectionID, profile))
        intelligenceSends[deviceID] = send
    }
    func intelligenceWorkerDisconnected(deviceID: String, connectionID: String) async {
        intelligenceWorkerDisconnects.append((deviceID, connectionID))
    }
    func receiveIntelligenceResult(_ result: WireIntelligenceResult, from deviceID: String) async {
        intelligenceResults.append((result, deviceID))
    }
    /// Drive a registered worker's send leg the way the mesh queue would.
    func pushIntelligenceRequest(_ request: WireIntelligenceRequest, to deviceID: String) async {
        await intelligenceSends[deviceID]?(request)
    }

    func broadcasts() -> AsyncStream<HostBroadcast> {
        AsyncStream { continuation in broadcastContinuation = continuation }
    }
    func push(_ broadcast: HostBroadcast) { broadcastContinuation?.yield(broadcast) }
    func setFailApproval(_ error: WireError?) { failApprovalWith = error }
    func setDashboard(_ snapshot: DashboardSnapshot) { dashboard = snapshot }

    // Mesh dispatch instrumentation: idempotent per requestID, like the real bridge.
    private var dispatchSessions: [String: SessionID] = [:]
    private(set) var acknowledgedDispatchIDs: [String] = []
    func startChatAcknowledged(
        _ request: StartChatRequest, requestID: String
    ) async -> WireChatStarted {
        acknowledgedDispatchIDs.append(requestID)
        if let existing = dispatchSessions[requestID] {
            return WireChatStarted(requestID: requestID, sessionID: existing)
        }
        let sessionID = SessionID(rawValue: "dispatched-\(dispatchSessions.count + 1)")
        dispatchSessions[requestID] = sessionID
        startedChats.append(request)
        return WireChatStarted(requestID: requestID, sessionID: sessionID)
    }

    // Mesh casting instrumentation.
    private var castTipsToServe: [CastTip] = []
    private var castCatchUpToServe: [String: [CastCatchUp]] = [:]
    private(set) var castCatchUpRequests: [(origin: String, channel: String, afterSeq: UInt64)] = []
    func setCastCatchUp(tips: [CastTip], batches: [String: [CastCatchUp]]) {
        castTipsToServe = tips
        castCatchUpToServe = batches
    }
    func castTips() async -> [CastTip] { castTipsToServe }
    func castCatchUp(origin: String, channel: String, afterSeq: UInt64) async -> [CastCatchUp] {
        castCatchUpRequests.append((origin, channel, afterSeq))
        return castCatchUpToServe["\(origin)/\(channel)"] ?? []
    }
}

/// Routes `PeerClient` dials to in-memory hosts by hostname — the dial-side companion to
/// `ManualListener`. Every dial mints a fresh `MemoryChannel` pair, pushes the server end
/// into the routed listener, and hands the client end back. Records what was dialed and
/// keeps the client-side channels so a test can sever a live connection mid-flight.
final class LoopbackPeerDialer: PeerDialer, @unchecked Sendable {
    private let lock = NSLock()
    private var routes: [String: ManualListener] = [:]
    private var dialedEndpoints: [PeerEndpoint] = []
    private var channels: [MemoryChannel] = []

    func route(host: String, to listener: ManualListener) {
        lock.withLock { routes[host] = listener }
    }

    var dialed: [PeerEndpoint] {
        lock.withLock { dialedEndpoints }
    }

    var lastChannel: MemoryChannel? {
        lock.withLock { channels.last }
    }

    func dial(_ endpoint: PeerEndpoint) async throws -> any FrameChannel {
        let host = switch endpoint {
        case .lan(let h, _), .tailnet(let h, _): h
        case .bonjour(let name, _): name
        case .relay(_, let membershipToken): membershipToken
        }
        let listener = lock.withLock { () -> ManualListener? in
            dialedEndpoints.append(endpoint)
            return routes[host]
        }
        guard let listener else { throw PeerDialError.unreachable("no route to \(host)") }
        let (client, server) = MemoryChannel.pair()
        lock.withLock { channels.append(client) }
        listener.push(server)
        return client
    }
}

/// A canned `PeerLANBrowsing` for mesh Bonjour-rediscovery tests: returns the endpoint mapped to a
/// peer's static-key fingerprint, or nil when it isn't "on the LAN".
struct StubLANBrowser: PeerLANBrowsing {
    let byFingerprint: [String: PeerEndpoint]
    func endpoint(forFingerprint fingerprint: String) async -> PeerEndpoint? {
        byFingerprint[fingerprint]
    }
}

/// Buffers a `SyncClient` event stream (single-iterator) so a test can await successive
/// matches without re-iterating. `waitFor` resolves from the buffer or suspends until a
/// matching event arrives.
actor EventRecorder {
    private var buffer: [SyncClient.Event] = []
    private var pendingPredicate: (@Sendable (SyncClient.Event) -> Bool)?
    private var continuation: CheckedContinuation<SyncClient.Event, Never>?

    func consume(_ stream: AsyncStream<SyncClient.Event>) {
        Task { [weak self] in
            for await event in stream { await self?.record(event) }
        }
    }

    private func record(_ event: SyncClient.Event) {
        if let predicate = pendingPredicate, predicate(event), let cont = continuation {
            pendingPredicate = nil; continuation = nil
            cont.resume(returning: event)
        } else {
            buffer.append(event)
        }
    }

    func waitFor(_ predicate: @escaping @Sendable (SyncClient.Event) -> Bool) async -> SyncClient.Event {
        if let index = buffer.firstIndex(where: predicate) {
            return buffer.remove(at: index)
        }
        return await withCheckedContinuation { cont in
            pendingPredicate = predicate
            continuation = cont
        }
    }

    /// Whether any already-buffered event matches — a non-blocking probe for "this event must
    /// NOT have arrived" assertions (drive a ping/pong round-trip first to flush the pipe).
    func hasBuffered(_ predicate: @escaping @Sendable (SyncClient.Event) -> Bool) -> Bool {
        buffer.contains(where: predicate)
    }
}

/// Poll an async accessor until it returns a non-empty collection, or the timeout elapses.
/// Robust to scheduler load (the full parallel suite) where bare `Task.yield()` spins aren't
/// enough time for a client→host round-trip to settle.
func poll<C: Collection>(
    timeout: Duration = .seconds(3),
    every interval: Duration = .milliseconds(10),
    _ read: @Sendable () async -> C
) async -> C {
    let deadline = ContinuousClock.now.advanced(by: timeout)
    while ContinuousClock.now < deadline {
        let value = await read()
        if !value.isEmpty { return value }
        try? await Task.sleep(for: interval)
    }
    return await read()
}
