import Foundation
import NucleicProtocol

/// Drives one client connection: routes + runs the Noise handshake, authenticates the
/// device, exchanges hello/welcome, then services `ClientMsg`s and pushes filtered
/// `HostBroadcast`s. One actor per connection — all sealing/sending is actor-isolated, so
/// the Noise nonce order matches the wire order.
actor ConnectionHandler {
    let id = UUID()
    private let connection: any FrameChannel
    /// Which transport this connection arrived on, when the channel self-identifies (LAN/tailnet/
    /// relay). `nonisolated` + immutable so `SyncHost` can read it on unregister without hopping the
    /// actor — it drives the "a LAN link just dropped, nudge the device onto a non-local path" push.
    nonisolated let transport: SyncTransportHint?
    private let bridge: any SyncHostBridge
    private let identity: DeviceIdentity
    private let store: any PairedDeviceStore
    private weak var host: SyncHost?

    private var session: SecureSession?
    private var deviceID = ""
    private var grantedScope: DeviceScope = .approve
    /// What kind of device the authenticated peer is (mesh P4/P5). Recorded at hello so the
    /// session-transfer verbs can be gated to mac peers only — a phone must never drive a transfer.
    private var peerKind: PeerKind = .iphone
    /// Whether the authenticated peer advertised `WireClientCapabilities.canSyncRoster` (mesh
    /// "join"). Gates pushing `HostMsg.meshRoster` to it — an older client that never advertised
    /// it stays on the display-only `listPeers` path and would `.unknown`-drop a roster push.
    private var clientSyncsRoster = false
    /// Whether the authenticated peer advertised `WireClientCapabilities.canCast` (mesh
    /// casting). An old client would `.unknown`-drop a cast anyway; the gate saves the metered
    /// relay bytes.
    private var clientCasts = false
    /// Whether this client mirrors Carbon shards (`WireClientCapabilities.canMirrorCarbon`) —
    /// gates the `carbonHeadAdvanced` push (CARBON_SHARDING §8.2; phones never set it, D13).
    private var clientMirrorsCarbon = false
    /// Whether both ends of this connection can run the direct-path upgrade (Covalence direct,
    /// SYNC §3.3): the peer advertised `canDirectConnect`, we advertise it, and the connection
    /// arrived on the relay wrapped in a `MultipathFrameChannel`. Set (and the mux armed) at
    /// hello; gates the `directOffer`/`directSelect` verbs.
    private var clientDirectConnects = false
    /// Set once the peer sent `castSubscribe` — live casts flow only after the subscribe, so
    /// the catch-up pass and the live tail share one starting point.
    private var castSubscribed = false
    /// The channel filter from the subscribe; nil = every channel at the peer's scope.
    private var castChannels: Set<String>?
    /// The intelligence-worker profile the peer advertised (`canProvideIntelligence` +
    /// `intelligenceProfile`, COVALENCE_RUNNER §5) — non-nil registers it with the host's
    /// mesh AFM queue. A capability bit without a profile (an older or minimal client) is
    /// synthesized from the peer kind: a Mac is desktop-class, anything else mobile.
    private var clientIntelligenceProfile: IntelligenceWorkerProfile?
    /// The in-flight enrollment of this connection as a mesh AFM worker, nil when the peer
    /// never offered execution. `close()` awaits it before withdrawing, so the enroll/withdraw
    /// pair can never invert — a fast connect→drop must not leave a ghost worker the queue
    /// keeps placing jobs on.
    private var intelligenceWorkerRegistration: Task<Void, Never>?
    private var subscriptions: [SessionID: Verbosity] = [:]
    private var closed = false

    // Inbound liveness (the host-side mirror of `SyncClient`'s keepalive). No transport reliably
    // signals a *silently* dead peer: the host's accepted LAN socket has no app-visible EOF when a
    // phone is suspended mid-flight, and the tailnet fd is a raw socketpair a half-open WireGuard
    // tunnel never EOFs — so `messageLoop` would block forever and the handler would linger,
    // reporting a gone device as still connected (suppressing its approval wakes + the LAN-drop
    // reconnect nudge). A live peer always sends *something* well within the deadline — every
    // `SyncClient` pings every 20s — so a lapse means the peer is gone: tear the zombie down. Any
    // inbound frame (a ping, a real message, a handshake step) refreshes `lastInboundAt`.
    private let livenessDeadline: Duration
    private var lastInboundAt: ContinuousClock.Instant?
    private var livenessTask: Task<Void, Never>?

    // A minimal single-consumer async FIFO. A detached pump task owns the (non-Sendable)
    // stream iterator and feeds frames in; `nextFrame()` is the actor-isolated consumer.
    // This keeps iteration out of the actor's isolation region (which Swift 6 forbids for a
    // stored mutating iterator) while serializing consumption.
    private var pending: [Data] = []
    private var waiter: CheckedContinuation<Data?, Never>?
    private var inboundFinished = false

    private func feed(_ frame: Data) {
        lastInboundAt = ContinuousClock.now  // any inbound frame proves the peer is still alive
        if let w = waiter { waiter = nil; w.resume(returning: frame) }
        else { pending.append(frame) }
    }
    private func finishInbound() {
        inboundFinished = true
        if let w = waiter { waiter = nil; w.resume(returning: nil) }
    }
    private func nextFrame() async -> Data? {
        if !pending.isEmpty { return pending.removeFirst() }
        if inboundFinished { return nil }
        return await withCheckedContinuation { continuation in waiter = continuation }
    }
    private func startPump() {
        let frames = connection.frames()
        Task { [weak self] in
            for await frame in frames { await self?.feed(frame) }
            await self?.finishInbound()
        }
    }

    init(
        connection: any FrameChannel, bridge: any SyncHostBridge,
        identity: DeviceIdentity, store: any PairedDeviceStore, host: SyncHost,
        livenessDeadline: Duration = .seconds(60)
    ) {
        self.connection = connection
        self.transport = connection.transport
        self.bridge = bridge
        self.identity = identity
        self.store = store
        self.host = host
        self.livenessDeadline = livenessDeadline
    }

    var peerDescription: String { connection.peerDescription }

    /// The multipath mux this connection rides, when `SyncHost.accept` wrapped it (relay
    /// connections with the direct path enabled). `nonisolated`: `connection` is immutable,
    /// and `SyncHost`'s upgrade coordinator reads this without hopping the actor.
    nonisolated var muxChannel: MultipathFrameChannel? { connection as? MultipathFrameChannel }

    /// The authenticated device id (set during the handshake preamble) — the host uses it
    /// to skip pushing to phones that already hold a live socket.
    func connectedDeviceID() -> String { deviceID }

    /// The authenticated peer's `ResponderLabel` — its paired name (falling back to its device
    /// kind) plus the short fingerprint of its identity key — stamped as `decidedBy` when it
    /// answers an approval. The `deviceID` is the full host id (SHA-256 hex of the static key),
    /// whose 16-char prefix is exactly that key's fingerprint.
    private func responderLabel() async -> String {
        let paired = await store.device(deviceID)?.label
        let name = paired.flatMap { $0.isEmpty ? nil : $0 } ?? peerKind.displayName
        return ResponderLabel.compose(name: name, fingerprint: String(deviceID.prefix(16)))
    }

    // MARK: - Lifecycle

    func run() async {
        startPump()
        // Watch liveness across the whole lifetime — a peer that opens a socket and then goes silent
        // (a stalled handshake, or a suspended phone mid-session) is reaped rather than lingering.
        startLivenessWatch()
        do {
            let secure = try await handshake()
            self.session = secure
            try await helloWelcome()
            await host?.register(self, deviceID: deviceID)
            registerIntelligenceWorkerIfOffered()
            try await messageLoop()
        } catch let error as WireError {
            sendIfPossible(.error(error))
        } catch {
            // Transport/handshake failure: nothing more we can say securely.
        }
        await close()
    }

    func close() async {
        guard !closed else { return }
        closed = true
        livenessTask?.cancel(); livenessTask = nil
        withdrawIntelligenceWorker()
        await host?.unregister(self, deviceID: deviceID)
        connection.close()
    }

    // MARK: - Inbound liveness watchdog

    /// Begin watching for a silently-dead peer: if no inbound frame arrives within `livenessDeadline`
    /// the connection is torn down (→ `unregister`), so the host stops treating a gone device as
    /// connected. Ticks on a fraction of the deadline so detection lands within ~1.5× it.
    private func startLivenessWatch() {
        lastInboundAt = ContinuousClock.now
        let interval = min(livenessDeadline, .seconds(20))
        livenessTask?.cancel()
        livenessTask = Task { [weak self] in
            while !Task.isCancelled {
                try? await Task.sleep(for: interval)
                guard !Task.isCancelled else { break }
                let dead = await self?.reapIfStale() ?? true
                if dead { break }
            }
        }
    }

    /// One liveness tick: tear the connection down if nothing has arrived within the deadline.
    /// Returns whether it was reaped (ends the watch loop).
    private func reapIfStale() async -> Bool {
        guard !closed else { return true }
        if let last = lastInboundAt, last.duration(to: ContinuousClock.now) > livenessDeadline {
            await close()
            return true
        }
        return false
    }

    // MARK: - Handshake routing (SYNC §4.2/§4.3)

    private func handshake() async throws -> SecureSession {
        guard let preFrame = await nextFrame() else { throw SyncTransportError.connectionClosed }
        let preamble = try CBORDecoder().decode(HandshakePreamble.self, from: preFrame)
        guard preamble.protocolVersion == SyncProtocol.version else {
            throw WireError(code: .protocolVersion, message: "handshake protocol mismatch")
        }
        deviceID = preamble.deviceID

        let driver: HandshakeDriver
        var expectedStatic: Data?
        switch preamble.mode {
        case .pair:
            guard let psk = await host?.activePairingSecret() else {
                throw WireError(code: .unauthorized, message: "no pairing in progress")
            }
            driver = NoiseSession.makeResponder(identity: identity, psk: psk)
        case .reconnect:
            guard let device = await store.device(preamble.deviceID) else {
                throw WireError(code: .unauthorized, message: "unknown device")
            }
            expectedStatic = device.staticPublicKey
            grantedScope = device.scope
            driver = NoiseSession.makeResponder(identity: identity, psk: nil)
        }

        while !driver.isComplete {
            if driver.isMyTurnToWrite {
                connection.send(try driver.write())
            } else {
                guard let frame = await nextFrame() else { throw SyncTransportError.connectionClosed }
                _ = try driver.read(frame)
            }
        }
        let secure = try driver.session()

        // Reconnect: the recovered peer static must match the pin (revocation gate).
        if preamble.mode == .reconnect, secure.remoteStaticKey != expectedStatic {
            throw WireError(code: .unauthorized, message: "device key mismatch")
        }
        return secure
    }

    // MARK: - hello / welcome

    private func helloWelcome() async throws {
        guard let frame = await nextFrame() else { throw SyncTransportError.connectionClosed }
        let msg = try open(frame)
        guard case .hello(let hello) = msg else {
            throw WireError(code: .malformed, message: "expected hello")
        }
        guard hello.protocolVersion == SyncProtocol.version else {
            throw WireError(code: .protocolVersion, message: "protocol mismatch")
        }
        // Channel gate: the remote must be built from a channel compatible with this host's
        // (`ReleaseChannel.isCompatible`). Skipped when either side hasn't declared one (a
        // host without a channel, or a remote that predates the field).
        let hostChannel = await bridge.releaseChannel
        if let hostChannel, let deviceChannel = hello.channel, !hostChannel.isCompatible(with: deviceChannel) {
            throw WireError(
                code: .channelMismatch,
                message: "remote is a \(deviceChannel.rawValue) build; this host is \(hostChannel.rawValue)")
        }
        let now = Date()
        // What the peer says it is (mesh P4). A pre-mesh iPhone omits it → `.iphone`.
        let peerKind = hello.kind
        self.peerKind = peerKind
        self.clientSyncsRoster = hello.clientCaps?.canSyncRoster ?? false
        self.clientCasts = hello.clientCaps?.canCast ?? false
        self.clientMirrorsCarbon = hello.clientCaps?.canMirrorCarbon ?? false
        // Offered AFM execution (COVALENCE_RUNNER §5): remember the worker profile so
        // `SyncHost.register` can enroll this device in the mesh intelligence queue.
        if hello.clientCaps?.canProvideIntelligence == true {
            self.clientIntelligenceProfile = hello.clientCaps?.intelligenceProfile
                ?? IntelligenceWorkerProfile(
                    deviceClass: peerKind.canOwnSessions ? .desktop : .mobile)
        }
        let peerCaps = PeerCapabilities(
            canHost: hello.clientCaps?.canHost ?? false,
            canRunAgents: hello.clientCaps?.canRunAgents ?? false)
        if let existing = await store.device(deviceID) {
            // The deviceID is the client's plaintext preamble claim; only the pinned static key
            // is authenticated. A reconnect (IK) already proved key possession above, but a
            // *pairing* (XXpsk0) party may put any deviceID in the preamble — so before touching
            // an existing record, require the just-authenticated key to match its pin. Otherwise
            // a QR holder could claim a paired device's ID and hijack its record + registry slot.
            guard session!.remoteStaticKey == existing.staticPublicKey else {
                throw WireError(code: .unauthorized, message: "device key mismatch")
            }
            // A peer may not silently promote itself to a session-owning kind (mac/cloud) on a
            // later hello — that grant needs the same accepting-side confirm as a fresh mac
            // pairing (locked decision #4). A same-or-lower kind change is harmless.
            if peerKind.canOwnSessions, !existing.kind.canOwnSessions {
                guard await bridge.approveMacPairing(deviceID: deviceID, label: hello.deviceLabel) else {
                    throw WireError(code: .unauthorized, message: MeshPairing.declinedMessage)
                }
            }
            // Reconnect: refresh metadata, keep the pin + granted scope.
            var updated = existing
            updated.label = hello.deviceLabel
            updated.pushToken = hello.pushToken ?? existing.pushToken
            updated.lastSeenAt = now
            updated.kind = peerKind
            updated.capabilities = peerCaps
            // Merge per-field rather than replace: a reconnect hello sent before the peer's tailnet
            // node is up carries a LAN-only snapshot, and overwriting would drop the tailnet IP we
            // need to redial it off-LAN. See `PeerAddresses.merged(with:)`.
            if let incoming = hello.addresses {
                updated.addresses = existing.addresses?.merged(with: incoming) ?? incoming
            } else {
                updated.addresses = existing.addresses
            }
            await store.upsert(updated)
            grantedScope = existing.scope
        } else {
            // Pairing a peer that can own sessions (a Mac) hands it control of this one —
            // QR-secret possession alone isn't enough; the accepting Mac's user confirms
            // (mesh P4 locked decision). Phones keep the existing pure-TOFU flow.
            if peerKind.canOwnSessions {
                guard await bridge.approveMacPairing(deviceID: deviceID, label: hello.deviceLabel) else {
                    throw WireError(code: .unauthorized, message: MeshPairing.declinedMessage)
                }
            }
            // First pairing (TOFU): pin the peer static we just authenticated. The phone is
            // a full control-scope client (parity with the Mac), so grant up to control.
            grantedScope = min(hello.scopeClaim, .control)
            await store.upsert(PairedDevice(
                deviceID: deviceID, label: hello.deviceLabel,
                staticPublicKey: session!.remoteStaticKey, scope: grantedScope,
                pushToken: hello.pushToken, pairedAt: now, lastSeenAt: now,
                kind: peerKind, capabilities: peerCaps, addresses: hello.addresses))
            await host?.didPair()  // pairing consumed; host can drop the QR
        }
        await bridge.meshPeersChanged()

        // Record this phone's APNs topic (its bundle id) before registering its token, so the
        // relay addresses wake + Live Activity pushes to the right per-channel topic. Stored
        // independently of the device token — a phone may run a Live Activity without having
        // granted notification permission (so no device token), and its glance still needs the
        // topic. No-op when the relay isn't configured.
        await host?.setPushTopic(deviceID: deviceID, topic: hello.pushTopic)
        // Mirror a fresh APNS token to the relay's out-of-band wake path (no-op when the
        // relay isn't configured).
        if let token = hello.pushToken, !token.isEmpty {
            await host?.didReceivePushToken(deviceID: deviceID, token: token)
        }

        let capabilities = await bridge.capabilities
        let welcome = Welcome(
            grantedScope: grantedScope,
            host: await bridge.hostInfo,
            capabilities: capabilities,
            modelCatalog: await bridge.modelCatalog,
            channel: hostChannel,
            addresses: await bridge.localAddresses,
            hostKind: (await bridge.hostKind).rawValue,
            // Stamp the account-level synced settings so a device starts from the host's truth
            // without waiting for a `HostMsg.settings` broadcast. Only when we actually sync them.
            settings: capabilities.canSyncSettings ? await bridge.syncedSettings : nil)
        send(.welcome(welcome))

        // Mesh "join": push our roster right after the welcome so a device that scanned a
        // *different* member (or reconnected) converges to the whole group. Gated both ways —
        // only when we serve rosters and the client advertised it can merge them.
        if capabilities.canSyncRoster, clientSyncsRoster {
            send(.meshRoster(await bridge.meshRoster()))
        }

        // Covalence direct (SYNC §3.3): arm the mux's control layer when both ends advertise
        // the upgrade and this connection actually rides one. The binding key is derived from
        // the Noise handshake hash, so the relay can't forge a switch/resync. Explicit
        // both-ways gating (not send-and-see): old hosts *error* on unknown ClientMsg tags.
        if capabilities.canDirectConnect, hello.clientCaps?.canDirectConnect == true,
           let mux = muxChannel, let session
        {
            mux.arm(bindingKey: MultipathFrameChannel.bindingKey(handshakeHash: session.handshakeHash))
            clientDirectConnects = true
        }
    }

    // MARK: - Message loop

    private func messageLoop() async throws {
        while let frame = await nextFrame() {
            let msg: ClientMsg
            do { msg = try open(frame) }
            catch where SecureSession.isAuthenticationFailure(error) {
                // The sender doesn't share this session's keys. On the relay a reconnecting
                // device keeps its routing tag, so its *fresh handshake* lands on this stale
                // handler — and every retry refreshes the liveness watchdog, so without this
                // the zombie eats the device's reconnects forever. Tear down: the close frees
                // the demux slot and the next inbound frame opens a fresh channel + handshake.
                throw SyncTransportError.connectionClosed
            }
            catch { send(.error(WireError(code: .malformed, message: "undecodable frame"))); continue }
            await dispatch(msg)
        }
    }

    private func dispatch(_ msg: ClientMsg) async {
        switch msg {
        case .hello:
            break  // already handled; ignore duplicates
        case .ping:
            send(.pong)
        case .listSessions:
            send(.sessionList(await bridge.sessionSummaries()))
        case .subscribe(let sub):
            subscriptions[sub.sessionID] = sub.verbosity
            if let snapshot = await bridge.snapshot(sub.sessionID, sinceSeq: sub.sinceSeq) {
                send(.snapshot(snapshot))
            } else {
                send(.error(WireError(code: .unknownSession, message: "no such session", sessionID: sub.sessionID)))
            }
        case .unsubscribe(let id):
            subscriptions[id] = nil
        case .approvalRespond(let id, let decision):
            guard grantedScope >= .approve else {
                send(.error(WireError(code: .unauthorized, message: "scope too low"))); return
            }
            let responder = await responderLabel()
            await guardedCall { try await self.bridge.respondToApproval(id, decision, by: responder) }
        case .sendInput(let id, let input):
            guard grantedScope >= .approve else {
                send(.error(WireError(code: .unauthorized, message: "scope too low", sessionID: id))); return
            }
            await guardedCall(sessionID: id) { try await self.bridge.sendInput(id, input) }
        case .interrupt(let id):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.interrupt(id) }
        case .resolveProcessStall(let id, let stallID, let kill):
            // A remote surface resolved a "host command looks hung" alert — Kill (tear down the
            // command's process tree) or Keep-waiting (dismiss). Control scope, matching the Mac's
            // own in-transcript buttons. Gated client-side on `canResolveProcessStall`.
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) {
                try await self.bridge.resolveProcessStall(id, stallID: stallID, kill: kill)
            }
        case .listDashboard:
            send(.dashboard(await bridge.dashboardSnapshot()))
        case .listPeers:
            // Read-only mesh presence (scope ≥ view). Gated client-side on
            // `WireCapabilities.canListPeers`; a client that predates the flag never sends it.
            send(.peerList(await bridge.peerSummaries()))
        case .addressUpdate(let addresses):
            // The authenticated device refreshing where *it* can be dialed (mesh P4). Only
            // its own record changes, so no extra scope beyond authentication is needed.
            // Gated client-side on `WireCapabilities.canUpdateAddresses`.
            if var device = await store.device(deviceID) {
                // Merge per-field: an `addressUpdate` that only refreshes the LAN port (the common
                // case — it rebinds constantly) must not erase the peer's tailnet IP / relay room.
                device.addresses = device.addresses?.merged(with: addresses) ?? addresses
                device.lastSeenAt = Date()
                await store.upsert(device)
                await bridge.meshPeersChanged()
            }
        case .meshRoster(let push):
            // The authenticated peer pushed its view of the group (mesh "join"). Merge it —
            // auto-pinning mac members it introduced and applying revocations. The peer only
            // sends this when we advertised `canSyncRoster`; a forged member can't survive the
            // merge's hostID↔key integrity check.
            await bridge.mergeMeshRoster(push, from: deviceID)
        // Session transfer (mesh P5). The source drives the transfer as our client. Gated on
        // `.control` scope + a mac peer (a phone must never move a session); the source only
        // sends these when we advertised `canReceiveSessionTransfer`.
        case .transferOffer(let offer):
            guard requireTransferPeer() else { return }
            for reply in await bridge.receiveTransferOffer(offer, from: deviceID) { send(reply) }
        case .transferChunk(let chunk):
            guard requireTransferPeer() else { return }
            for reply in await bridge.receiveTransferChunk(chunk) { send(reply) }
        case .transferCommit(let id):
            guard requireTransferPeer() else { return }
            for reply in await bridge.receiveTransferCommit(id) { send(reply) }
        case .transferCancel(let id):
            guard requireTransferPeer() else { return }
            await bridge.receiveTransferCancel(id)
        case .requestSessionTransfer(let id, let destinationDeviceID):
            // A peer asking us to move a session *we* own to a third device (possibly itself).
            // Same gate as receiving a transfer — control scope + a session-owning peer — because
            // it moves a session just as decisively; the transfer itself still runs from here.
            guard requireTransferPeer(), requireControl(id) else { return }
            // Off the message loop: the move packages and streams a whole session, which would
            // otherwise stall every other verb (and our pings) on this connection for its duration.
            // The reply still rides the pump, so the requester hears about a failure either way.
            Task { [weak self] in
                guard let self else { return }
                await self.guardedCall(sessionID: id) {
                    try await self.bridge.transferSession(id, to: destinationDeviceID)
                }
            }
        case .fetchDiff(let id):
            // Read-only (every scope may read diffs — DeviceScope.view's charter).
            if let diff = await bridge.sessionDiff(id) {
                send(.sessionDiff(diff))
            } else {
                send(.error(WireError(code: .unknownSession, message: "no diff for session", sessionID: id)))
            }
        case .fetchTranscript(let req):
            // Read-only full-transcript sync (scope ≥ view, like `fetchDiff`; a phone fetches
            // from its host, a mirror-holding peer serves at view — so NOT the `.control`+mac
            // transfer gate). Gated client-side on `WireCapabilities.canSyncTranscripts`. Served
            // from a live controller OR any on-disk transcript we hold (owned or mirrored), so a
            // non-host peer can answer too. Streamed as ordered batches then a terminal marker.
            await serveTranscriptFetch(req)
        case .fetchCarbonHeads(let req):
            // Carbon replication (CARBON_SHARDING §8.1-8.2). Read-only, scope ≥ view like
            // `fetchTranscript`: heads are IDs/seqs, blobs below are ciphertext — nothing a
            // viewer-scope peer couldn't already learn. Gated client-side on `canServeCarbon`.
            guard req.formatVersion <= CarbonFormat.version else {
                send(.carbonManifestUnavailable(CarbonBlobUnavailable(
                    requestID: req.requestID, id: nil, reason: .unsupportedFormat)))
                return
            }
            send(.carbonHeads(CarbonHeadsReply(
                requestID: req.requestID,
                heads: await bridge.carbonHeads(projectUUID: req.projectUUID))))
        case .fetchCarbonManifests(let req):
            await serveCarbonManifestFetch(req)
        case .fetchCarbonShards(let req):
            await serveCarbonShardFetch(req)
        case .carbonShardAck(let ack):
            // §8.3 window: acked frames free window slots; pump the request's queue.
            if var state = carbonShardQueues[ack.requestID] {
                state.unacked = max(0, state.unacked - ack.ids.count)
                carbonShardQueues[ack.requestID] = state
                await pumpCarbonShards(ack.requestID)
            }
        case .carbonLockAcquire(let req):
            // Lock arbitration (CARBON_SHARDING §17.5). Mutating — control scope, like the
            // other verbs that drive edits. The acquire legitimately parks for as long as
            // contention lasts (LOCKING §4.2), so it runs off the message loop; an interim
            // `queued` decision tells the requester its waiter is parked.
            guard requireControl() else { return }
            guard req.formatVersion <= CarbonFormat.version else {
                send(.carbonLockDecision(CarbonLockDecisionMsg(
                    requestID: req.requestID, projectUUID: req.projectUUID,
                    domainRef: req.domainRef, sessionID: req.sessionID, outcome: .cancelled)))
                return
            }
            let requesterID = deviceID
            if await bridge.carbonLockWouldBlock(req, from: requesterID) {
                send(.carbonLockDecision(CarbonLockDecisionMsg(
                    requestID: req.requestID, projectUUID: req.projectUUID,
                    domainRef: req.domainRef, sessionID: req.sessionID, outcome: .queued)))
            }
            Task { [weak self, bridge] in
                let outcome = await bridge.carbonLockAcquire(req, from: requesterID)
                await self?.send(.carbonLockDecision(CarbonLockDecisionMsg(
                    requestID: req.requestID, projectUUID: req.projectUUID,
                    domainRef: req.domainRef, sessionID: req.sessionID, outcome: outcome)))
            }
        case .carbonLockRelease(let req):
            guard requireControl() else { return }
            // Same format gate as `carbonLockAcquire`: a frame from a newer peer may carry
            // semantics this build doesn't implement, and acting on half-understood release
            // instructions frees files. Drop it — the lock stays held (fail closed).
            guard req.formatVersion <= CarbonFormat.version else { return }
            await bridge.carbonLockRelease(req, from: deviceID)
        case .startChat(let request):
            guard requireControl() else { return }
            if let requestID = request.requestID {
                // Mesh dispatch: the requester needs a definitive, correlated outcome to drive
                // its fallback chain — and idempotency per requestID (a retried request acks
                // the existing session) is what makes timeout-retry double-start-proof.
                send(.chatStarted(await bridge.startChatAcknowledged(request, requestID: requestID)))
            } else {
                await guardedCall { try await self.bridge.startChat(request) }
            }
        case .captureTodo(let request):
            guard requireControl() else { return }
            await guardedCall { try await self.bridge.captureTodo(request) }
        case .dispatchTodo(let todo, let project):
            guard requireControl() else { return }
            await guardedCall { try await self.bridge.dispatchTodo(todo, in: project) }
        case .setTodoStatus(let todo, let status):
            guard requireControl() else { return }
            await guardedCall { try await self.bridge.remoteSetTodoStatus(todo, status) }
        case .deleteTodo(let todo):
            guard requireControl() else { return }
            await guardedCall { try await self.bridge.remoteDeleteTodo(todo) }
        case .renameSession(let id, let title):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.renameSession(id, to: title) }
        case .setFavorite(let id, let flag):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.setFavorite(id, flag) }
        case .setArchived(let id, let flag):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.setArchived(id, flag) }
        case .markSessionDone(let id):
            // The remote surface's "Mark Done" — the same declaration the Mac sidebar's makes, so
            // the host finalizes it identically (an autoship-armed chat ships). Gated client-side
            // on `WireCapabilities.canMarkSessionDone`.
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.markSessionDoneRemote(id) }
        case .deleteSession(let id):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.deleteSessionRemote(id) }
        case .integrate(let id, let mode):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.integrate(id, mode) }
        case .discard(let id):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.discard(id) }
        case .setSessionModel(let id, let model):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.setSessionModel(id, model) }
        case .setSessionEffort(let id, let effort):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.setSessionEffort(id, effort) }
        case .setSessionAuto(let id, let flag):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.setSessionAuto(id, flag) }
        case .setSessionAutoShip(let id, let flag):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.setSessionAutoShip(id, flag) }
        case .setSessionShipBranch(let id, let branch):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.setSessionShipBranch(id, branch) }
        case .cancelQueuedMessage(let id, let messageID):
            guard requireControl(id) else { return }
            await guardedCall(sessionID: id) { try await self.bridge.cancelQueuedMessage(id, messageID) }
        case .requestPairingCode:
            // "Add a device to this mesh" relayed from a phone: open a pairing window here and
            // hand back the join code. Control scope — minting a code lets a new device join the
            // group, the same authority a phone's control actions carry. Gated client-side on
            // `WireCapabilities.canMintPairingCode`. A *Mac* joining via the returned code still
            // hits the accepting-Mac confirm (`approveMacPairing`) on this host's own screen; a
            // phone joins via pure TOFU, so the relay flow completes hands-free for phones.
            guard requireControl() else { return }
            send(.pairingCode(await bridge.mintPairingCode()))
        case .cancelPairingCode:
            // The relaying phone dismissed its sheet — tear down the one-time secret so it doesn't
            // stay armed. Same gating as `requestPairingCode`.
            guard requireControl() else { return }
            await bridge.cancelPairingCode()
        case .respondMacPair(let deviceID, let approve):
            // The relaying phone answered a forwarded Mac-pair confirm. Control scope — approving a
            // Mac join grants it control of this host. First responder wins; the host ignores a
            // stale/mismatched deviceID.
            guard requireControl() else { return }
            await bridge.respondToMacPairing(deviceID: deviceID, approve: approve)
        case .registerLiveActivity(let token, _):
            // Read-only (scope ≥ view): it only lets the host push the same aggregate glance the
            // phone already renders, keeping the lock screen fresh while the phone is backgrounded
            // (UX_IOS §5.3). The activity id rides along for forward use; the host keys by device.
            await host?.registerLiveActivity(deviceID: deviceID, token: token)
        case .endLiveActivity:
            await host?.endLiveActivity(deviceID: deviceID)
        case .registerPushToStartToken(let token):
            // Read-only, like `registerLiveActivity`: it only lets the host *create* the same
            // aggregate glance the phone already renders — over APNs when work starts while the app
            // is closed, so it appears without the user opening the app first (UX_IOS §5.3).
            await host?.registerPushToStartToken(deviceID: deviceID, token: token)
        case .setForeground(let foreground):
            // Read-only: tells the host whether the app is foreground (creating its own Live Activity
            // over the socket) or backgrounded (needs the host to push-to-start it). Steers only the
            // glance path (UX_IOS §5.3); the host keys it by device.
            await host?.setDeviceForeground(deviceID: deviceID, foreground: foreground)
        case .intelligenceResult(let result):
            // Answer to a delegated intelligence call (runner §5). Only a host that advertised
            // `canDelegateIntelligence` ever solicits these; the bridge correlates by request id
            // and drops stale/unknown ids, so no extra scope gate — the result only feeds soft-AI
            // summaries/classification. Defaulted no-op on a Mac host.
            await bridge.receiveIntelligenceResult(result, from: deviceID)
        case .credentialManifest(let manifest):
            // Credential-mesh metadata (runner §6): descriptors + refresh leases, never secret
            // bytes. Approve scope — the same trust that lets a device answer approvals lets it
            // tell the host what credentials it could provide. Defaulted no-op on a Mac host.
            guard grantedScope >= .approve else {
                send(.error(WireError(code: .unauthorized, message: "scope too low"))); return
            }
            await bridge.receiveCredentialManifest(manifest, from: deviceID)
        case .credentialProvision(let envelope):
            // Sealed credential delivery (runner §6). The host can only open boxes sealed to its
            // own sealing key, so a stray envelope is inert; approve scope keeps view-only devices
            // from pushing credentials at all. Defaulted no-op on a Mac host.
            guard grantedScope >= .approve else {
                send(.error(WireError(code: .unauthorized, message: "scope too low"))); return
            }
            await bridge.receiveCredentialProvision(envelope, from: deviceID)
        case .credentialRevoke(let tombstones):
            // Mesh-wide credential deletion. Control scope — deleting the credential every
            // member runs agents with is the same trust tier as rotating it (`agentLoginBegin`).
            // Gated client-side on `canRevokeCredentials`. The bridge clears local copies,
            // records the tombstones, and fans them out; a replay is a no-op there.
            guard requireControl() else { return }
            await bridge.receiveCredentialRevocation(tombstones, from: deviceID)
        case .createProject(let request):
            // Clone-and-register (CLOUD_RUNTIME §4.3) — how a device seeds work onto a fresh
            // runner. Gated client-side on `WireCapabilities.canCreateProjects`; a control action
            // like `startChat`. The outcome — success or failure — answers as `projectCreated`
            // correlated by the request's id, so the asking UI can settle its sheet; the project
            // row itself reaches every client through the dashboard broadcast.
            guard requireControl() else { return }
            do {
                send(.projectCreated(try await bridge.createProject(request)))
            } catch {
                let message = (error as? WireError)?.message ?? "\(error)"
                send(.projectCreated(WireProjectCreated(requestID: request.requestID, error: message)))
            }
        case .castSubscribe(let sub):
            // Read-only cast feed opt-in (mesh casting; scope ≥ view like `fetchDiff`, with
            // per-channel minimums on top). Gated client-side on `WireCapabilities.canCast`.
            // Serves the catch-up pass, then live casts flow through `deliver`.
            await serveCastSubscribe(sub)
        case .composerTyping(let typing):
            // Live composer streaming: the device reports its draft and this host republishes
            // it on the `composer.typing` cast channel, stamped with the authenticated sender's
            // identity (a client can't cast — its host is the relay). Approve scope: the same
            // trust that lets the device send input lets it stream a draft. Gated client-side on
            // `WireCapabilities.canStreamComposer`.
            guard grantedScope >= .approve else {
                send(.error(WireError(
                    code: .unauthorized, message: "scope too low", sessionID: typing.sessionID)))
                return
            }
            await bridge.receiveComposerTyping(typing, from: deviceID)
        case .updateSettings(let settings):
            // Account-level settings today govern how *approvals* behave (resolve a lock-screen
            // Allow/Deny over the Nucleic Edge), so a device trusted to answer approvals is trusted
            // to opt into answering them over the relay — approve scope, matching `approvalRespond`.
            // The host persists, enforces (forces the relay method on), and re-broadcasts the
            // authoritative value as `HostMsg.settings`. Gated client-side on `canSyncSettings`.
            guard grantedScope >= .approve else {
                send(.error(WireError(code: .unauthorized, message: "scope too low"))); return
            }
            await bridge.updateSyncedSettings(settings)
        case .directOffer(let offer):
            // Direct-path upgrade (SYNC §3.3). Read-only in effect — it changes which wire
            // carries the frames, not any host state — so no scope gate beyond authentication.
            // Gated client-side on `canDirectConnect`; both-ways gate + armed mux checked here.
            guard clientDirectConnects, muxChannel != nil else {
                send(.directDecline("direct upgrade unavailable on this connection")); return
            }
            await host?.handleDirectOffer(offer, from: self)
        case .directSelect(let selected):
            guard clientDirectConnects, muxChannel != nil else {
                send(.directDecline("direct upgrade unavailable on this connection")); return
            }
            await host?.handleDirectSelect(selected, from: self)
        case .agentLoginBegin(let request):
            // Remote agent sign-in (docs/REMOTE_AGENT_LOGIN.md). Control scope — a sign-in
            // rotates the mesh-wide credential every member runs agents with, the same trust
            // tier as the other control verbs. Gated client-side on `canBrokerAgentLogin`.
            // The bridge kicks the flow off in its own task and returns promptly: the flow
            // blocks for minutes awaiting the browser round-trip, whose `agentLoginCallback`
            // arrives on THIS serial loop — awaiting it here would deadlock the attempt.
            guard requireControl() else { return }
            await bridge.beginAgentLogin(request, from: deviceID) { [weak self] msg in
                await self?.sendIfLive(msg)
            }
        case .agentLoginCallback(let callback):
            // The captured redirect for an attempt this connection began. No extra gate beyond
            // control: the code is single-use, worthless without the host-held PKCE verifier,
            // and stale/unknown requestIDs are dropped by the coordinator.
            guard requireControl() else { return }
            await bridge.deliverAgentLoginCallback(callback)
        case .agentLoginCancel(let requestID):
            guard requireControl() else { return }
            await bridge.cancelAgentLogin(requestID: requestID)
        }
    }

    /// Send one host message if the connection is still authenticated — the async-closure
    /// variant of `send` handed to long-running bridge flows (remote agent sign-in), so a
    /// challenge/result races a teardown safely instead of writing into a closed session.
    func sendIfLive(_ msg: HostMsg) {
        guard session != nil else { return }
        send(msg)
    }

    // MARK: - Direct-path replies (called by SyncHost's DirectHostUpgrader)

    func deliverDirectAnswer(_ offer: DirectPunchOffer) {
        guard session != nil else { return }
        send(.directAnswer(offer))
    }

    func deliverDirectGo() {
        guard session != nil else { return }
        send(.directGo)
    }

    func deliverDirectDecline(_ reason: String) {
        guard session != nil else { return }
        send(.directDecline(reason))
    }

    /// Gate a control-scope action; emits an `unauthorized` error and returns false if denied.
    private func requireControl(_ sessionID: SessionID? = nil) -> Bool {
        guard grantedScope >= .control else {
            send(.error(WireError(code: .unauthorized, message: "control scope required", sessionID: sessionID)))
            return false
        }
        return true
    }

    /// Gate a session-transfer verb: `.control` scope AND a mac peer (a session can only be
    /// owned/received by a Mac). Emits an `unauthorized` error and returns false if denied.
    private func requireTransferPeer() -> Bool {
        guard grantedScope >= .control, peerKind.canOwnSessions else {
            send(.error(WireError(code: .unauthorized, message: "session transfer requires a control-scope mac peer")))
            return false
        }
        return true
    }

    private func guardedCall(sessionID: SessionID? = nil, _ body: () async throws -> Void) async {
        do { try await body() }
        catch let wire as WireError { send(.error(wire)) }
        catch { send(.error(WireError(code: .backpressure, message: "\(error)", sessionID: sessionID))) }
    }

    // MARK: - Carbon shard replication serving (CARBON_SHARDING §8.1, §8.3)

    /// Per-request shard-serve state for the §8.3 ack window: ids not yet sent + frames sent
    /// but not yet acked. The message loop is serial, so pumps and acks never interleave.
    private var carbonShardQueues: [String: (projectUUID: UUID, pending: [String], unacked: Int)] = [:]

    /// Answer a `fetchCarbonManifests`: each id answers individually (data or notHeld), so a
    /// partial holder still serves what it has and the requester can fall through per id.
    /// Manifests are small and capped at 64/request — no window needed (unlike shards).
    private func serveCarbonManifestFetch(_ req: CarbonBlobsFetch) async {
        guard req.formatVersion <= CarbonFormat.version else {
            send(.carbonManifestUnavailable(CarbonBlobUnavailable(
                requestID: req.requestID, id: nil, reason: .unsupportedFormat)))
            return
        }
        for id in req.ids.prefix(CarbonBlobsFetch.maxManifestIDs) {
            if let blob = await bridge.carbonManifestBlob(id: id, projectUUID: req.projectUUID) {
                send(.carbonManifestData(CarbonBlobData(
                    requestID: req.requestID, projectUUID: req.projectUUID, id: id, blob: blob)))
            } else {
                send(.carbonManifestUnavailable(CarbonBlobUnavailable(
                    requestID: req.requestID, id: id, reason: .notHeld)))
            }
            await Task.yield()  // don't starve other connections on a long batch
        }
    }

    /// Answer a `fetchCarbonShards` under the ack window: enqueue, then pump. NOT the
    /// transcript fire-and-stream — shard frames are ~2 MiB each, so the server keeps at most
    /// `CarbonFormat.shardAckWindow` un-acked frames in flight (the transfer discipline).
    private func serveCarbonShardFetch(_ req: CarbonBlobsFetch) async {
        guard req.formatVersion <= CarbonFormat.version else {
            send(.carbonShardUnavailable(CarbonBlobUnavailable(
                requestID: req.requestID, id: nil, reason: .unsupportedFormat)))
            return
        }
        carbonShardQueues[req.requestID] = (
            req.projectUUID, Array(req.ids.prefix(CarbonBlobsFetch.maxShardIDs)), 0)
        await pumpCarbonShards(req.requestID)
    }

    private func pumpCarbonShards(_ requestID: String) async {
        guard var state = carbonShardQueues[requestID] else { return }
        while state.unacked < CarbonFormat.shardAckWindow, !state.pending.isEmpty {
            let id = state.pending.removeFirst()
            if let blob = await bridge.carbonShardBlob(id: id, projectUUID: state.projectUUID) {
                send(.carbonShardData(CarbonBlobData(
                    requestID: requestID, projectUUID: state.projectUUID, id: id, blob: blob)))
                state.unacked += 1
            } else {
                // A miss is a tiny frame — answer it outside the window so a mostly-absent
                // batch settles fast.
                send(.carbonShardUnavailable(CarbonBlobUnavailable(
                    requestID: requestID, id: id, reason: .notHeld)))
            }
            await Task.yield()
        }
        // Done only when nothing is pending AND everything sent was acked; otherwise park the
        // state for the next ack. (A requester that stops acking leaks only this small record,
        // reaped with the connection.)
        if state.pending.isEmpty && state.unacked == 0 {
            carbonShardQueues[requestID] = nil
        } else {
            carbonShardQueues[requestID] = state
        }
    }

    /// Answer a `fetchTranscript`: stream the requested events as ordered `transcriptChunk`
    /// batches, then a terminal `transcriptFetchComplete` (or `transcriptUnavailable` when we
    /// hold nothing / can't handle the format). `throughSeq` is the tip shipped — equal to the
    /// request's `afterSeq` when we're already current, so the requester learns there's no gap.
    private func serveTranscriptFetch(_ req: TranscriptFetch) async {
        guard req.formatVersion <= TranscriptSyncFormat.version else {
            send(.transcriptUnavailable(TranscriptUnavailable(
                requestID: req.requestID, sessionID: req.sessionID, reason: .unsupportedFormat)))
            return
        }
        guard let (header, revertEpoch, events) =
            await bridge.transcriptEvents(req.sessionID, afterSeq: req.afterSeq)
        else {
            send(.transcriptUnavailable(TranscriptUnavailable(
                requestID: req.requestID, sessionID: req.sessionID, reason: .notHeld)))
            return
        }
        var batchIndex = 0
        var start = events.startIndex
        while start < events.endIndex {
            let end = events.index(
                start, offsetBy: TranscriptChunk.maxEventsPerBatch, limitedBy: events.endIndex)
                ?? events.endIndex
            send(.transcriptChunk(TranscriptChunk(
                requestID: req.requestID, sessionID: req.sessionID,
                batchIndex: batchIndex, events: Array(events[start..<end]))))
            batchIndex += 1
            start = end
            await Task.yield()  // don't starve other connections on a long transcript
        }
        var headerJSON: Data?
        if req.afterSeq == 0, let header { headerJSON = try? header.encodedForWire() }
        send(.transcriptFetchComplete(TranscriptFetchComplete(
            requestID: req.requestID, sessionID: req.sessionID,
            throughSeq: events.last?.seq ?? req.afterSeq, headerJSON: headerJSON,
            revertEpoch: revertEpoch)))
    }

    /// Answer a `castSubscribe` (mesh casting): record the subscription (live casts flow from
    /// `deliver` after this), then serve one catch-up pass per (origin, channel) held — at the
    /// peer's scope, honoring its channel filter and cursors. Batches interleave with live
    /// `casts`; the client's identity-key dedup makes the overlap harmless (at-least-once
    /// delivery, exactly-once apply).
    private func serveCastSubscribe(_ sub: CastSubscribe) async {
        guard sub.formatVersion <= CastFormat.version else { return }  // future format: inert
        castSubscribed = true
        castChannels = sub.channels.map(Set.init)
        var cursors: [String: UInt64] = [:]
        for cursor in sub.cursors {
            cursors["\(cursor.originHostID)\u{1}\(cursor.channel)"] = cursor.afterSeq
        }
        for tip in await bridge.castTips() {
            if let castChannels, !castChannels.contains(tip.channel) { continue }
            guard grantedScope >= CastChannelRegistry.minimumScope(for: tip.channel) else { continue }
            // Never serve a peer its own casts back — it is the authority for them.
            guard tip.originHostID != deviceID else { continue }
            let after = cursors["\(tip.originHostID)\u{1}\(tip.channel)"] ?? 0
            for batch in await bridge.castCatchUp(
                origin: tip.originHostID, channel: tip.channel, afterSeq: after
            ) {
                send(.castCatchUp(batch))
                await Task.yield()  // don't starve other connections on a large backlog
            }
        }
    }

    // MARK: - Broadcast fan-out (per-subscription + verbosity filtering, SYNC §5.5)

    func deliver(_ broadcast: HostBroadcast) {
        guard session != nil else { return }
        switch broadcast {
        case .events(let sid, let events):
            guard let verbosity = subscriptions[sid], verbosity != .statusOnly else {
                // statusOnly clients still want terminal runFinished for the session.
                if subscriptions[sid] == .statusOnly {
                    let terminal = events.filter { if case .runFinished = $0.kind { return true } else { return false } }
                    if !terminal.isEmpty { send(.events(EventBatch(sessionID: sid, events: terminal))) }
                }
                return
            }
            let filtered = verbosity == .full ? events : events.filter { !isPartialDelta($0) }
            if !filtered.isEmpty { send(.events(EventBatch(sessionID: sid, events: filtered))) }
        case .approvalRequested(let req):
            if subscriptions[req.sessionID] != nil { send(.approvalRequested(req)) }
        case .approvalResolved(let resolved):
            send(.approvalResolved(resolved))  // harmless to all; clients dismiss by id
        case .sessionUpdated(let summary):
            send(.sessionUpdated(summary))
        case .sessionList(let list):
            send(.sessionList(list))  // a deletion: clients replace their mirrored list
        case .dashboard(let snapshot):
            send(.dashboard(snapshot))  // Home/Projects/To-Dos refresh
        case .toolSummaries(let sid, let summaries):
            // Only viewers actually subscribed to this session care (mesh session sync); scope it
            // like `.events`. Verbosity is irrelevant — a summary line is a tiny status detail, so
            // even a statusOnly subscriber gets it. An empty set carries nothing, so skip it.
            if subscriptions[sid] != nil, !summaries.isEmpty {
                send(.toolSummaries(ToolSummaryBatch(sessionID: sid, summaries: summaries)))
            }
        case .transcriptReverted(let sid, let throughSeq, let revertEpoch):
            // A revert/undo truncated the transcript. Send it to EVERY device, not just this
            // session's live subscribers — unlike `.events` (a firehose only a viewer wants), a
            // revert is a tiny structural fact any device holding a *copy* must apply. Peer Macs
            // proactively mirror sessions they aren't viewing (via `fetchTranscript`, which never
            // registers a subscription), and phones keep an offline `SessionCache`; both truncate
            // that stored copy on receipt (`applyPeerRevert`/`SessionCache.truncateEvents`) whether
            // or not the session is on screen. Scoping to subscribers left those background copies
            // holding the reverted tail — a later open re-seeded it through the grow-only merge, and
            // the mirror coordinator only ever *grows* a mirror, so nothing healed it. Broadcasting
            // unconditionally (like `.sessionUpdated`/`.sessionList`) is what makes undo converge
            // across the mesh; the message is a `(sessionID, throughSeq)` pair, so the fan-out is
            // cheap and a no-op on any device that holds nothing for `sid`.
            send(.transcriptReverted(TranscriptReverted(
                sessionID: sid, throughSeq: throughSeq, revertEpoch: revertEpoch)))
        case .macPairRequested(let req):
            // Forward a Mac-pair confirm to control-scope clients so the phone that shared the join
            // code can approve it. Old clients ignore the unknown tag; a view/approve-scope client
            // isn't trusted to grant another Mac control, so it never sees the prompt.
            if grantedScope >= .control { send(.macPairRequested(req)) }
        case .macPairResolved(let deviceID, let approved):
            if grantedScope >= .control { send(.macPairResolved(deviceID, approved)) }
        case .casts(let casts):
            // Mesh casting: only subscribed, cast-capable clients; honor the channel filter +
            // per-channel scope; and echo suppression — never send a peer casts it originated
            // (it's the authority for them; the bounce would be pure waste).
            guard clientCasts, castSubscribed else { return }
            let filtered = casts.filter { cast in
                cast.originHostID != deviceID
                    && (castChannels?.contains(cast.channel) ?? true)
                    && grantedScope >= CastChannelRegistry.minimumScope(for: cast.channel)
            }
            if !filtered.isEmpty { send(.casts(CastBatch(casts: filtered))) }
        case .settings(let settings):
            // Account-level; every device converges on the host's truth regardless of scope or
            // subscriptions. An older client ignores the unknown `settings` tag.
            send(.settings(settings))
        case .carbonHeadAdvanced(let head):
            // Carbon replication push (CARBON_SHARDING §8.2 step 1): only mirror-capable
            // clients (Macs/runners that advertised `canMirrorCarbon`) — a phone holds no
            // store and would drop it anyway; skipping saves the metered relay bytes.
            guard clientMirrorsCarbon else { return }
            send(.carbonHeadAdvanced(head))
        case .carbonLockLedgerAdvanced(let head):
            // Lock-ledger push (CARBON_SHARDING §17.5): same mirror gate — members chain-sync
            // the `lock` stream and re-fold; phones hold no store (inert, D13).
            guard clientMirrorsCarbon else { return }
            send(.carbonLockLedgerAdvanced(head))
        }
    }

    /// Push this device's relay credential (mesh P2). Called by `SyncHost` after
    /// registration, off the hello path — minting is a network round-trip that must not
    /// stall the welcome.
    func deliverRelayMembership(_ membership: WireRelayMembership) {
        send(.relayMembership(membership))
    }

    /// Share the owner's runner-pool credential (COVALENCE_RUNNER item 4). Control scope
    /// only — the credential grants pool management (provision/stop/cap), the same trust tier
    /// as the other control verbs. Called by `SyncHost` after registration.
    func deliverRunnerPoolCredential(_ credential: WireRunnerPoolCredential) {
        guard grantedScope >= .control, session != nil else { return }
        send(.runnerPoolCredential(credential))
    }

    /// Ask this device for the credentials the runner lacks (COVALENCE_RUNNER §6). Approve
    /// scope — the same trust that lets a device answer approvals lets it be asked for (and
    /// volunteer) credentials; the reply only ever carries boxes sealed to the runner's key.
    /// Called by `SyncHost` after registration.
    func deliverCredentialNeed(_ need: WireCredentialNeed) {
        guard grantedScope >= .approve, session != nil else { return }
        send(.credentialNeeded(need))
    }

    /// Enroll this connection in the host's mesh AFM queue (COVALENCE_RUNNER §5) when the
    /// peer offered execution at hello. Approve scope — the same trust bar as the other runner
    /// verbs, and a request's inputs are digests of the user's own sessions. Runs between the
    /// hello and the message loop; `withdrawIntelligenceWorker` (from `close()`) awaits the
    /// enrollment before withdrawing, so the pair is ordered even against an immediate drop.
    private func registerIntelligenceWorkerIfOffered() {
        guard !closed, intelligenceWorkerRegistration == nil, grantedScope >= .approve,
              let profile = clientIntelligenceProfile, !deviceID.isEmpty
        else { return }
        let deviceID = self.deviceID
        let connectionID = id.uuidString
        intelligenceWorkerRegistration = Task { [weak self, bridge] in
            await bridge.intelligenceWorkerConnected(
                deviceID: deviceID, connectionID: connectionID, profile: profile,
                send: { [weak self] request in
                    await self?.deliverIntelligenceRequest(request)
                })
        }
    }

    /// Withdraw this connection's worker enrollment on teardown. Keyed by the handler id, so a
    /// superseded (deduped) connection's close can't unregister its live replacement; ordered
    /// behind the enrollment task so the withdraw can never land first and leave a ghost.
    private func withdrawIntelligenceWorker() {
        guard let registration = intelligenceWorkerRegistration else { return }
        intelligenceWorkerRegistration = nil
        let deviceID = self.deviceID
        let connectionID = id.uuidString
        Task { [bridge] in
            await registration.value
            await bridge.intelligenceWorkerDisconnected(
                deviceID: deviceID, connectionID: connectionID)
        }
    }

    /// Push one delegated intelligence job at this device (COVALENCE_RUNNER §5) — the mesh
    /// AFM queue's send leg. The device answers `ClientMsg.intelligenceResult` with the same id.
    func deliverIntelligenceRequest(_ request: WireIntelligenceRequest) {
        guard grantedScope >= .approve, session != nil else { return }
        send(.intelligenceRequest(request))
    }

    /// Mirror a rotated credential back to this device (COVALENCE_RUNNER §6), sealed to that
    /// device's own sealing key — the cloud analogue of `syncClaudeLoginBack`. Approve scope,
    /// same as the need push. Called by `SyncHost` when the runner's credential changed.
    func deliverCredentialUpdate(_ envelope: SealedCredentialEnvelope) {
        guard grantedScope >= .approve, session != nil else { return }
        send(.credentialUpdate(envelope))
    }

    /// Push mesh-wide credential deletions at this device (`HostMsg.credentialRevoked`) — the
    /// post-hello convergence push and the fan-out when a deletion lands. Approve scope, same
    /// as `credentialUpdate` (it names kinds and stamps, never secret bytes, but it *acts* on
    /// the receiver's stores).
    func deliverCredentialRevocation(_ tombstones: [CredentialTombstone]) {
        guard grantedScope >= .approve, session != nil, !tombstones.isEmpty else { return }
        send(.credentialRevoked(tombstones))
    }

    /// Push the host's per-provider auth state (remote agent sign-in) — the wire projection of
    /// Settings ▸ Agents, so the device's Agent Accounts UI stays truthful. Any scope: it
    /// carries install/auth flags and an account label, never secret bytes. Called by
    /// `SyncHost` after registration and on every change.
    func deliverAgentAuthStatus(_ statuses: [WireProviderAuthStatus]) {
        guard session != nil else { return }
        send(.agentAuthStatus(statuses))
    }

    /// Push the current mesh roster to this client (mesh "join"), when membership changed while it
    /// was connected. A no-op for a client that didn't advertise `canSyncRoster` (it stays on the
    /// display-only `listPeers` path). Called by `SyncHost` fanout.
    func deliverMeshRoster(_ push: MeshRosterPush) {
        guard clientSyncsRoster, session != nil else { return }
        send(.meshRoster(push))
    }

    private func isPartialDelta(_ event: AgentEvent) -> Bool {
        switch event.kind {
        case .toolCallInputDelta: return true
        case .assistantText(let c), .thinking(let c): return c.isPartial
        default: return false
        }
    }

    // MARK: - Sealing

    private func send(_ msg: HostMsg) {
        guard let session else { return }
        if let frame = try? session.seal(msg) { connection.send(frame) }
    }
    private func sendIfPossible(_ msg: HostMsg) { send(msg) }

    private func open(_ frame: Data) throws -> ClientMsg {
        guard let session else { throw WireError(code: .malformed, message: "no session") }
        return try session.open(ClientMsg.self, from: frame)
    }
}

extension ConnectionHandler: Equatable, Hashable {
    nonisolated static func == (lhs: ConnectionHandler, rhs: ConnectionHandler) -> Bool { lhs.id == rhs.id }
    nonisolated func hash(into hasher: inout Hasher) { hasher.combine(id) }
}
