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
    private let stream: AsyncStream<Data>
    private let continuation: AsyncStream<Data>.Continuation
    private var peer: MemoryChannel?

    private init(name: String) {
        peerDescription = name
        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() -> (client: MemoryChannel, server: MemoryChannel) {
        let a = MemoryChannel(name: "client")
        let b = MemoryChannel(name: "server")
        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

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

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

    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] = [:]
    ) {
        self.hostInfo = hostInfo
        self.capabilities = capabilities
        self.modelCatalog = modelCatalog
        self.sessions = sessions
        self.snapshots = snapshots
    }

    func sessionSummaries() async -> [WireSessionSummary] { sessions }
    func snapshot(_ id: SessionID, sinceSeq: UInt64?) async -> SessionSnapshot? { snapshots[id] }

    func respondToApproval(_ id: ApprovalID, _ decision: Decision) async throws {
        if let failApprovalWith { throw failApprovalWith }
        respondedApprovals.append((id, decision))
    }
    func sendInput(_ id: SessionID, _ input: AgentInput) async throws { sentInputs.append((id, input)) }
    func interrupt(_ id: SessionID) async throws { interrupts.append(id) }

    // 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 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 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)) }

    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 }
}

/// 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
        }
    }
}

/// 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()
}
