feat(ios): native receiver app (Swift, min iOS 17)

A native iOS receiver so an iPhone can act as the second receiver, speaking the existing signaling + RTP protocol (no C++ changes) and mirroring the Android receiver (Phase 8) source-to-source. - RTP core (header/packet, jitter buffer, H.264 depacketizer) ported from the Android receiver - BSD-socket signaling server (dual-stack, most-recent-peer, never-throwing sends) + NSBonjourServices - VideoToolbox H.264 decode (in-band SPS/PPS, real-time, rebuilds on size change) -> AVSampleBufferDisplayLayer - PLI keyframe recovery (500 ms) + pendingOffer for late surface attach - XcodeGen project + bootstrap.sh; XCTest port of the Android suite + new coverage - .gitignore for generated artifacts; CHANGELOG; PHASES + MEMORY updated Status: authored; on-device validation pending a Mac + Xcode 26 + iPhone 16.
2026-09-10 21:31:53 +02:00
parent 4f93c7cd20
commit c43dd3ca4a
31 changed files with 2282 additions and 1 deletions
@@ -0,0 +1,88 @@
+import SwiftUI
+import UIKit
+
+/// Owns the pipeline and mirrors its status to the UI. The pipeline posts its
+/// callbacks to the main thread, so the @Published mutations happen there.
+final class ReceiverController: ObservableObject {
+    @Published var status = "Starting…"
+    @Published var showStatus = true
+
+    private let pipeline: ReceiverPipeline
+    private let pixelSize: CGSize
+
+    init() {
+        let localIP = LocalAddress.primaryIPv4()
+        let size = ReceiverController.screenPixelSize()
+        pixelSize = size
+        pipeline = ReceiverPipeline(
+            localIP: localIP,
+            displaySize: { [weak self] in self?.pixelSize ?? .zero },
+            onStatus: { [weak self] text in self?.apply(status: text) },
+            onFirstFrame: { [weak self] in self?.apply(showStatus: false) },
+            onVideoSize: { _ in })
+    }
+
+    func start() { pipeline.start() }
+    func stop() { pipeline.stop() }
+    func bindSink(_ sink: RenderSink) { pipeline.attachSink(sink) }
+
+    private func apply(status: String? = nil, showStatus: Bool? = nil) {
+        if let s = status { self.status = s }
+        if let v = showStatus { self.showStatus = v }
+    }
+
+    static func screenPixelSize() -> CGSize {
+        let scale = UIScreen.main.scale
+        let b = UIScreen.main.bounds
+        return CGSize(width: b.width * scale, height: b.height * scale)
+    }
+}
+
+struct ContentView: View {
+    @StateObject private var controller = ReceiverController()
+    @Environment(\.scenePhase) private var scenePhase
+
+    var body: some View {
+        ZStack {
+            Color.black.ignoresSafeArea()
+            VideoSurfaceView { sink in controller.bindSink(sink) }
+                .ignoresSafeArea()
+
+            if controller.showStatus {
+                VStack {
+                    Spacer()
+                    Text(controller.status)
+                        .font(.footnote)
+                        .foregroundStyle(.white)
+                        .multilineTextAlignment(.center)
+                        .padding(.horizontal, 24)
+                        .padding(.vertical, 12)
+                        .background(.black.opacity(0.55), in: RoundedRectangle(cornerRadius: 10))
+                    Spacer()
+                    Spacer()
+                }
+                .transition(.opacity)
+            }
+        }
+        .onAppear { controller.start() }
+        .onChange(of: scenePhase) { _, phase in
+            switch phase {
+            case .active: controller.start()
+            case .inactive, .background: controller.stop()
+            @unknown default: break
+            }
+        }
+    }
+}
+
+@main
+struct ReceiverApp: App {
+    var body: some Scene {
+        WindowGroup {
+            ContentView()
+                .preferredColorScheme(.dark)
+                .statusBarHidden(true)
+                .persistentSystemOverlays(.hidden)
+        }
+    }
+}
@@ -0,0 +1,34 @@
+import SwiftUI
+import AVFoundation
+
+/// Hosts the `AVSampleBufferDisplayLayer` and exposes it as a `RenderSink`.
+/// The layer keeps the full screen and letterboxes via `.resizeAspect`, so the
+/// decoded stream (already downscaled to the display size by the sender) is
+/// shown 1:1 with no distortion.
+struct VideoSurfaceView: UIViewRepresentable {
+    let onSink: (RenderSink) -> Void
+
+    func makeCoordinator() -> Coordinator {
+        Coordinator()
+    }
+
+    func makeUIView(context: Context) -> UIView {
+        let view = UIView(frame: .zero)
+        view.backgroundColor = .black
+
+        let layer = AVSampleBufferDisplayLayer()
+        layer.videoGravity = .resizeAspect
+        view.layer.addSublayer(layer)
+
+        let sink = AVSampleBufferRenderSink(layer: layer)
+        context.coordinator.sink = sink
+        onSink(sink)
+        return view
+    }
+
+    func updateUIView(_ uiView: UIView, context: Context) {}
+
+    final class Coordinator {
+        var sink: AVSampleBufferRenderSink?
+    }
+}
@@ -0,0 +1,57 @@
+import CoreMedia
+import Foundation
+
+/// Builds a `CMVideoFormatDescription` for H.264 from in-band SPS + PPS.
+///
+/// VideoToolbox needs the parameter sets up front; the sender repeats them
+/// before every keyframe, so any keyframe carries a complete set. This is the
+/// Core Foundation recipe for an H.264 "config" format description.
+enum H264FormatDescription {
+    static func create(sps: [UInt8], pps: [UInt8]) -> CMVideoFormatDescription? {
+        let pointersArray = CFArrayCreateMutable(kCFAllocatorDefault, 0, &kCFTypeArrayCallBacks)
+        for ps in [sps, pps] {
+            guard let descriptor = makeParameterSetDescriptor(ps) else {
+                CFRelease(pointersArray)
+                return nil
+            }
+            CFArrayAppendValue(pointersArray, descriptor.takeUnretainedValue())
+            CFRelease(descriptor) // the array now owns it
+        }
+
+        let key = kCMFormatDescriptionExtension_SampleDescriptionPointers as CFString
+        let attrs = CFDictionaryCreateMutable(kCFAllocatorDefault, 0, &kCFDictionaryKeyCallBacks, &kCFDictionaryValueCallBacks)
+        CFDictionarySetValue(attrs, key, pointersArray)
+
+        var config: Unmanaged<CMVideoFormatDescription>?
+        let status = CMVideoFormatDescriptionCreate(
+            kCFAllocatorDefault,
+            kCMVideoCodecType_H264,
+            0, 0, 0,
+            attrs,
+            &config)
+        CFRelease(attrs)
+        CFRelease(pointersArray)
+        guard status == noErr, let c = config else { return nil }
+        return c.takeRetainedValue()
+    }
+
+    private static func makeParameterSetDescriptor(_ ps: [UInt8]) -> Unmanaged<CMVideoFormatDescription>? {
+        let cfData = Data(ps) as CFData
+        let oneElement = CFArrayCreateMutable(kCFAllocatorDefault, 0, &kCFTypeArrayCallBacks)
+        CFArrayAppendValue(oneElement, cfData)
+        let key = kCMFormatDescriptionExtension_SampleDescriptionPointers as CFString
+        let attrs = CFDictionaryCreateMutable(kCFAllocatorDefault, 0, &kCFDictionaryKeyCallBacks, &kCFDictionaryValueCallBacks)
+        CFDictionarySetValue(attrs, key, oneElement)
+
+        var desc: Unmanaged<CMVideoFormatDescription>?
+        let status = CMVideoFormatDescriptionCreateForCodecType(
+            kCFAllocatorDefault,
+            kCMVideoCodecType_H264,
+            attrs,
+            &desc)
+        CFRelease(oneElement)
+        CFRelease(attrs)
+        guard status == noErr, let d = desc else { return nil }
+        return d
+    }
+}
@@ -0,0 +1,174 @@
+import VideoToolbox
+import CoreMedia
+import CoreGraphics
+import Foundation
+
+/// Hardware H.264 decoder using VideoToolbox. The stream is self-describing:
+/// the sender repeats SPS/PPS in-band at every keyframe, so no out-of-band
+/// codec data is needed. On a keyframe the in-band SPS/PPS (re)establish the
+/// stream size; when that size changes, the decode session is recreated — the
+/// Android C2 "in-band SPS reconfigure" pattern, which also avoids the startup
+/// squish (we never report a placeholder size before the first real keyframe).
+///
+/// The decode output callback may run on a worker thread, so session and
+/// format-description access is guarded by a lock.
+final class H264VideoToolboxDecoder {
+    private let renderSink: RenderSink
+    private let stateLock = NSLock()
+    private var session: VTDecompressionSession?
+    private var formatDescription: CMVideoFormatDescription?
+    private var streamSize = CGSize.zero
+    private var realSizeSeen = false
+
+    init(renderSink: RenderSink) {
+        self.renderSink = renderSink
+    }
+
+    /// The decoded resolution, once the first keyframe configured the session
+    /// (nil before that — the caller must not drive layout off a placeholder).
+    func outputSize() -> CGSize? {
+        guard realSizeSeen, streamSize != .zero else { return nil }
+        return streamSize
+    }
+
+    /// Decodes one access unit (Annex-B) and renders the output. Returns false
+    /// when a decode error occurred and the caller should request a keyframe.
+    func decode(accessUnit annexB: [UInt8], rtpTimestamp: Int, isKeyFrame: Bool) -> Bool {
+        if isKeyFrame {
+            guard let sets = NalExtractor.parameterSets(annexB) else { return false }
+            configureIfNeeded(sps: sets.sps, pps: sets.pps)
+        }
+        stateLock.lock()
+        let session = self.session
+        let cd = self.formatDescription
+        stateLock.unlock()
+        guard let session, let cd else { return false }
+        guard let avcc = AvccConverter.toAvcc(annexB) else { return false }
+
+        let pts = CMTime(value: CMTimeValue(rtpTimestamp), timescale: 90000)
+        guard let blockBuffer = makeBlockBuffer(avcc) else { return false }
+
+        var infoFlags: VTDecodeInfoFlags = []
+        let status = VTDecompressionSessionDecodeFrame(
+            session,
+            blockBuffer,
+            isKeyFrame ? kVTDecodeFrameFlags_EnableFastPath : 0,
+            &infoFlags,
+            pts)
+        CFRelease(blockBuffer)
+
+        if status != noErr {
+            // Recoverable: the next keyframe (SPS/PPS + IDR) re-primes it.
+            teardownSession()
+            return false
+        }
+        realSizeSeen = true
+        return true
+    }
+
+    func release() {
+        stateLock.lock()
+        teardownSessionLocked()
+        if let cd = formatDescription { CFRelease(cd) }
+        formatDescription = nil
+        stateLock.unlock()
+        renderSink.detach()
+        streamSize = .zero
+        realSizeSeen = false
+    }
+
+    // MARK: - Internals
+
+    private func configureIfNeeded(sps: [UInt8], pps: [UInt8]) {
+        guard let cd = H264FormatDescription.create(sps: sps, pps: pps) else { return }
+        var dims = CMVideoDimensions()
+        guard CMVideoFormatDescriptionGetDimensions(cd, &dims) == noErr else {
+            CFRelease(cd)
+            return
+        }
+        let size = CGSize(width: CGFloat(dims.width), height: CGFloat(dims.height))
+        stateLock.lock()
+        // Same size and a live session: keep it (the sender only changes the
+        // stream when the receiver's display size changes).
+        if session != nil && size == streamSize {
+            CFRelease(cd)
+            stateLock.unlock()
+            return
+        }
+        teardownSessionLocked()
+
+        var outSession: VTDecompressionSession?
+        let status = VTDecompressionSessionCreate(
+            kCFAllocatorDefault,
+            cd,
+            vtOutputCallback,
+            Unmanaged.passUnretained(self).toOpaque(),
+            &outSession)
+        guard status == noErr, let newSession = outSession else {
+            CFRelease(cd)
+            stateLock.unlock()
+            return
+        }
+        // Low-latency decode: emit as soon as the frame is complete.
+        VTSessionSetProperty(newSession, kVTDecompressionPropertyKey_RealTime, kCFBooleanTrue)
+
+        renderSink.setFormatDescription(cd)
+        formatDescription = cd // we hold the +1 from H264FormatDescription.create
+        session = newSession
+        streamSize = size
+        realSizeSeen = false
+        stateLock.unlock()
+    }
+
+    // Caller holds stateLock.
+    private func teardownSessionLocked() {
+        if let s = session {
+            VTDecompressionSessionInvalidate(s)
+            CFRelease(s)
+        }
+        session = nil
+    }
+
+    // Caller does not hold the lock.
+    private func teardownSession() {
+        stateLock.lock()
+        teardownSessionLocked()
+        stateLock.unlock()
+    }
+
+    private func makeBlockBuffer(_ bytes: [UInt8]) -> CMBlockBuffer? {
+        let cfData = Data(bytes) as CFData
+        var blockBuffer: CMBlockBuffer?
+        let status = CMBlockBufferCreateWithData(kCFAllocatorDefault, cfData, &blockBuffer)
+        guard status == noErr, let bb = blockBuffer else { return nil }
+        return bb
+    }
+
+    // Runs on whatever thread VideoToolbox uses for the output callback.
+    private func handleOutput(_ pixelBuffer: CVPixelBuffer?, _ presentationTime: CMTime?) {
+        stateLock.lock()
+        let cd = formatDescription
+        stateLock.unlock()
+        guard let cd, let pixelBuffer else { return }
+        let pts = presentationTime ?? CMTime(value: 0, timescale: 600)
+        var sampleBuffer: CMSampleBuffer?
+        let status = CMSampleBufferCreate(kCFAllocatorDefault, nil, 0, pts, .invalid, 1, 0, nil, &sampleBuffer)
+        guard status == noErr, let sb = sampleBuffer else { return }
+        guard CMSampleBufferSetDataBufferFromPixelBuffer(sb, pixelBuffer) == noErr else {
+            CFRelease(sb)
+            return
+        }
+        renderSink.enqueue(sb)
+    }
+}
+
+/// C-compatible decode output callback; recovers the decoder from the refCon.
+private func vtOutputCallback(_ refCon: UnsafeMutableRawPointer?,
+                              _ pixelBuffer: CVPixelBuffer?,
+                              _ presentationTime: CMTime?,
+                              _ duration: CMTime?,
+                              _ infoFlags: VTDecodeInfoFlags) {
+    guard let refCon = refCon else { return }
+    let decoder = Unmanaged<H264VideoToolboxDecoder>.fromOpaque(refCon).takeUnretainedValue()
+    decoder.handleOutput(pixelBuffer, presentationTime: presentationTime)
+}
@@ -0,0 +1,61 @@
+import AVFoundation
+import CoreMedia
+
+/// A render target for decoded frames. The decoder enqueues a `CMSampleBuffer`
+/// (wrapping a CVPixelBuffer) per frame; the sink renders it. Mirrors the role
+/// of the Android `Surface` + MediaCodec surface-mode rendering.
+protocol RenderSink: AnyObject {
+    var isAttached: Bool { get }
+    func attach()
+    func detach()
+    /// Updates the layer's video format description (called when the in-band
+    /// SPS/PPS establish a (new) stream size).
+    func setFormatDescription(_ formatDescription: CMVideoFormatDescription)
+    /// Enqueues one decoded frame for display.
+    func enqueue(_ sampleBuffer: CMSampleBuffer)
+}
+
+/// Renders decoded CVPixelBuffers via `AVSampleBufferDisplayLayer`, which does
+/// the video-scaling for us. `.resizeAspect` gives letterbox directly, so the
+/// host view can stay full-screen without a manual transform (the same lesson
+/// as the Android TextureView sizing: size the surface to the aspect, not a
+/// transform matrix).
+final class AVSampleBufferRenderSink: RenderSink {
+    private let layer: AVSampleBufferDisplayLayer
+    private var session: AVSampleBufferDisplayLayerSession?
+    private let lock = NSLock()
+    private(set) var isAttached = false
+
+    init(layer: AVSampleBufferDisplayLayer) {
+        self.layer = layer
+    }
+
+    func attach() {
+        lock.lock()
+        defer { lock.unlock() }
+        guard session == nil else { return }
+        let s = AVSampleBufferDisplayLayerSession(layer)
+        s.start()
+        session = s
+        isAttached = true
+    }
+
+    func detach() {
+        lock.lock()
+        defer { lock.unlock() }
+        session?.stop()
+        session = nil
+        isAttached = false
+    }
+
+    func setFormatDescription(_ formatDescription: CMVideoFormatDescription) {
+        layer.formatDescription = formatDescription
+    }
+
+    func enqueue(_ sampleBuffer: CMSampleBuffer) {
+        lock.lock()
+        let s = session
+        lock.unlock()
+        s?.enqueue(sampleBuffer)
+    }
+}
@@ -0,0 +1,268 @@
+import Foundation
+import CoreGraphics
+
+/// The receiver pipeline, mirroring the C++ ReceiverPipeline and the Kotlin
+/// receiver:
+///
+///   signaling server (offer → answer)
+///   UDP RTP → jitter buffer → depacketize → VideoToolbox → render sink
+///
+/// Recovery matches the C++ receiver: a damaged frame is dropped and a PLI
+/// (rate-limited to one per 500 ms) asks the sender for a keyframe.
+///
+/// Concurrency: all shared state and every decode call run on a single serial
+/// queue; the reader thread only polls/receives UDP and forwards datagrams to
+/// that queue. UI callbacks are marshalled to the main thread.
+final class ReceiverPipeline {
+    static let desiredUdpPort: UInt16 = 5004
+    static let desiredSignalingPort: UInt16 = 5005
+    static let pliMinIntervalMs: Int = 500
+    static let receiveBufferSize: Int32 = 4 * 1024 * 1024
+
+    private let localIP: String
+    private let displaySize: () -> CGSize
+    private let onStatus: (String) -> Void
+    private let onFirstFrame: () -> Void
+    private let onVideoSize: (CGSize) -> Void
+
+    private let queue = DispatchQueue(label: "sc.receiver.pipeline")
+
+    private var running = false
+    private var udp: UdpTransport?
+    private var signaling: SignalingServer?
+    private var readerThread: Thread?
+
+    private let jitter = JitterBuffer()
+    private var depacketizer = H264Depacketizer()
+    private var decoder: H264VideoToolboxDecoder?
+    private var renderSink: RenderSink?
+    private var pendingOffer: SignalingMessage?
+    private var activeSession = ""
+    private var firstFrameSeen = false
+    private var currentVideoSize = CGSize.zero
+    private var lastPliAtMs: Int64 = 0
+
+    init(localIP: String,
+         displaySize: @escaping () -> CGSize,
+         onStatus: @escaping (String) -> Void,
+         onFirstFrame: @escaping () -> Void,
+         onVideoSize: @escaping (CGSize) -> Void) {
+        self.localIP = localIP
+        self.displaySize = displaySize
+        self.onStatus = onStatus
+        self.onFirstFrame = onFirstFrame
+        self.onVideoSize = onVideoSize
+    }
+
+    /// The current decoded resolution (zero until the first keyframe).
+    func videoSize() -> CGSize {
+        return queue.sync { currentVideoSize }
+    }
+
+    // MARK: - Lifecycle
+
+    /// Binds the ports, starts the signaling server, and starts reading RTP.
+    func start() {
+        queue.async { [weak self] in
+            guard let self, !self.running else { return }
+
+            let udp = UdpTransport()
+            guard udp.bind(preferredPort: Self.desiredUdpPort, receiveBufferSize: Self.receiveBufferSize) else {
+                self.postStatus("Failed to bind the media port")
+                return
+            }
+
+            let signaling = SignalingServer(
+                onOffer: { [weak self] offer in self?.queue.async { self?.handleOffer(offer) } },
+                onPli: { _ in })
+            guard signaling.start(Self.desiredSignalingPort) else {
+                udp.close()
+                self.postStatus("Failed to start signaling")
+                return
+            }
+
+            self.running = true
+            self.udp = udp
+            self.signaling = signaling
+
+            let thread = Thread { [weak self] in self?.readLoop(udp: udp) }
+            thread.name = "rtp-reader"
+            self.readerThread = thread
+            thread.start()
+
+            let ip = self.localIP
+            let mediaPort = udp.port
+            let sigPort = signaling.port
+            self.postStatus("Listening on \(ip) (media :\(mediaPort), signaling :\(sigPort))\nWaiting for a sender… (fall back to: screencast --send --peer \(ip):\(sigPort))")
+        }
+    }
+
+    /// Points the (current or future) decoder at a render sink. A pending offer
+    /// (accepted while no sink existed) configures its decoder now and requests
+    /// a keyframe, since the sender only emits one when asked.
+    func attachSink(_ sink: RenderSink) {
+        queue.async { [weak self] in
+            guard let self else { return }
+            self.renderSink = sink
+            sink.attach()
+            guard let offer = self.pendingOffer else { return }
+            self.pendingOffer = nil
+            // The decoder configures on the first keyframe; the sink is already
+            // attached, so creation cannot fail. A late-configured decoder needs
+            // a keyframe (the sender only emits one when asked).
+            self.decoder = H264VideoToolboxDecoder(renderSink: sink)
+            self.requestPli()
+        }
+    }
+
+    /// Forgets a destroyed render sink so a later offer cannot render into it.
+    func detachSink() {
+        queue.async { [weak self] in
+            guard let self else { return }
+            self.decoder?.release()
+            self.decoder = nil
+            self.renderSink?.detach()
+            self.renderSink = nil
+        }
+    }
+
+    /// Stops listening; the pipeline can be started again.
+    func stop() {
+        queue.async { [weak self] in
+            guard let self, self.running else { return }
+            self.running = false
+            self.activeSession = ""
+            self.udp?.close()
+            self.udp = nil
+            self.readerThread?.join()
+            self.readerThread = nil
+            self.signaling?.close()
+            self.signaling = nil
+            self.decoder?.release()
+            self.decoder = nil
+            self.pendingOffer = nil
+            self.firstFrameSeen = false
+            self.currentVideoSize = .zero
+            self.postStatus("Stopped")
+        }
+    }
+
+    // MARK: - Media path
+
+    private func readLoop(udp: UdpTransport) {
+        while true {
+            switch udp.poll(timeoutMs: 100) {
+            case 1:
+                if let data = udp.receiveDatagram() {
+                    queue.async { [weak self] in self?.processDatagram(data) }
+                }
+            case 0:
+                continue // timeout: re-check on the next poll
+            default:
+                return // closed/errored: the socket was closed by stop()
+            }
+        }
+    }
+
+    private func processDatagram(_ data: [UInt8]) {
+        guard let packet = RtpPacket.parse(data) else { return }
+        for released in jitter.push(packet) {
+            handleDepacketized(released)
+        }
+    }
+
+    private func handleDepacketized(_ packet: RtpPacket) {
+        let result = depacketizer.depacketize(packet)
+
+        if let accessUnit = result.accessUnit {
+            let presentation = Int(packet.header.timestamp & 0xFFFFFFFF)
+            guard let decoder = decoder else { return }
+            if !decoder.decode(accessUnit: accessUnit, rtpTimestamp: presentation, isKeyFrame: result.isKeyFrame) {
+                // Input/decode trouble: the dropped frame corrupts the GOP
+                // until the next keyframe — ask for one.
+                requestPli()
+            }
+            if !firstFrameSeen {
+                firstFrameSeen = true
+                postFirstFrame()
+            }
+            postVideoSizeIfChanged()
+        }
+
+        if result.frameDropped {
+            requestPli()
+        }
+    }
+
+    private func postVideoSizeIfChanged() {
+        guard let size = decoder?.outputSize(), size != .zero else { return }
+        if size != currentVideoSize {
+            currentVideoSize = size
+            postVideoSize(size)
+        }
+    }
+
+    // MARK: - Signaling
+
+    private func handleOffer(_ offer: SignalingMessage) {
+        guard case let .offer(sessionId, codec, _, _, _, _, _, _) = offer else { return }
+        if codec != "h264" {
+            postStatus("Unsupported codec: \(codec)")
+            return
+        }
+        // New session: pristine decoder, reassembly state, and session.
+        decoder?.release()
+        decoder = nil
+        pendingOffer = nil
+        if renderSink != nil {
+            decoder = H264VideoToolboxDecoder(renderSink: renderSink!)
+        } else {
+            // No surface yet: park the offer; attachSink configures later.
+            pendingOffer = offer
+        }
+
+        depacketizer = H264Depacketizer()
+        jitter.clear()
+        firstFrameSeen = false
+        currentVideoSize = .zero
+        activeSession = sessionId
+
+        let size = displaySize()
+        let answer = SignalingMessage.answer(
+            sessionId: sessionId,
+            rtpAddress: "", // the sender targets the address of its own signaling connection
+            rtpPort: Int(udp?.port ?? 0),
+            displayWidth: Int(size.width),
+            displayHeight: Int(size.height))
+        signaling?.send(answer)
+
+        if decoder != nil {
+            postStatus("Session \(sessionId) negotiated — waiting for the first frame…")
+        } else {
+            postStatus("Session \(sessionId) negotiated — waiting for the display surface…")
+        }
+    }
+
+    /// Rate-limited keyframe request, callable from any thread (it always runs
+    /// on the pipeline queue in practice).
+    private func requestPli() {
+        let session = activeSession
+        guard !session.isEmpty else { return }
+        let now = Int64(Date().timeIntervalSince1970 * 1000)
+        if now - lastPliAtMs < Int64(Self.pliMinIntervalMs) { return }
+        lastPliAtMs = now
+        signaling?.send(.pli(sessionId: session))
+    }
+
+    // MARK: - UI callbacks (main thread)
+
+    private func postStatus(_ text: String) {
+        DispatchQueue.main.async { [onStatus] in onStatus(text) }
+    }
+    private func postFirstFrame() {
+        DispatchQueue.main.async { [onFirstFrame] in onFirstFrame() }
+    }
+    private func postVideoSize(_ size: CGSize) {
+        DispatchQueue.main.async { [onVideoSize] in onVideoSize(size) }
+    }
+}
@@ -0,0 +1,73 @@
+import Foundation
+
+/// Converts between Annex-B (start-code delimited) and AVCC (4-byte
+/// big-endian length prefixed) H.264 representations.
+///
+/// VideoToolbox consumes AVCC: each NAL unit is preceded by a 32-bit length.
+/// Our depacketizer emits Annex-B (the project's canonical 3-byte start codes),
+/// so the decoder feeds AVCC derived here.
+enum AvccConverter {
+    /// Splits an Annex-B access unit into its NAL units (start codes removed).
+    static func nalUnits(_ annexB: [UInt8]) -> [[UInt8]] {
+        guard annexB.count >= 4 else { return [] }
+        let n = annexB.count
+
+        // Locate every start code (3-byte `00 00 01` and 4-byte `00 00 00 01`).
+        var offsets: [Int] = []
+        var i = 0
+        while i + 2 < n {
+            if annexB[i] == 0 && annexB[i + 1] == 0 && annexB[i + 2] == 1 {
+                offsets.append(i)
+                i += 3
+                continue
+            }
+            if i + 3 < n, annexB[i + 2] == 0, annexB[i + 3] == 1 {
+                offsets.append(i)
+                i += 4
+                continue
+            }
+            i += 1
+        }
+        guard !offsets.isEmpty else { return [] }
+
+        var units: [[UInt8]] = []
+        for (idx, offset) in offsets.enumerated() {
+            let nalStart = offset + 3
+            let nalEnd = idx + 1 < offsets.count ? offsets[idx + 1] : n
+            let nal = Array(annexB[nalStart..<nalEnd])
+            if !nal.isEmpty { units.append(nal) }
+        }
+        return units
+    }
+
+    /// Returns the AVCC form of an Annex-B access unit, or nil if it has no NALs.
+    static func toAvcc(_ annexB: [UInt8]) -> [UInt8]? {
+        let units = nalUnits(annexB)
+        guard !units.isEmpty else { return nil }
+        var out: [UInt8] = []
+        out.reserveCapacity(annexB.count + units.count * 4)
+        for nal in units {
+            let len = nal.count
+            out.append(UInt8((len >> 24) & 0xFF))
+            out.append(UInt8((len >> 16) & 0xFF))
+            out.append(UInt8((len >> 8) & 0xFF))
+            out.append(UInt8(len & 0xFF))
+            out.append(contentsOf: nal)
+        }
+        return out
+    }
+
+    /// Splits an AVCC byte array (4-byte big-endian length prefixes) into NAL units.
+    static func fromAvcc(_ avcc: [UInt8]) -> [[UInt8]] {
+        var units: [[UInt8]] = []
+        var i = 0
+        let n = avcc.count
+        while i + 4 <= n {
+            let len = (Int(avcc[i]) << 24) | (Int(avcc[i + 1]) << 16) | (Int(avcc[i + 2]) << 8) | Int(avcc[i + 3])
+            guard len > 0, i + 4 + len <= n else { break }
+            units.append(Array(avcc[(i + 4)..<(i + 4 + len)]))
+            i += 4 + len
+        }
+        return units
+    }
+}
@@ -0,0 +1,160 @@
+import Foundation
+
+/// Result of feeding one packet to the depacketizer.
+struct DepacketizeResult {
+    /// Completed access unit (Annex-B with 3-byte start codes) when the frame closed undamaged.
+    var accessUnit: [UInt8]?
+    /// True when this call discarded a frame as damaged (packet loss or unsupported packetization).
+    var frameDropped = false
+    /// True when the completed access unit carries SPS/PPS (a keyframe).
+    var isKeyFrame = false
+}
+
+/// Reassembles RFC 6184 packet streams (single NAL unit packets and FU-A)
+/// into Annex-B access units. Packets must arrive in order; frames damaged by
+/// sequence gaps or missing fragments are reported via DepacketizeResult.
+///
+/// Mirrors the C++ `H264Depacketizer` (same state machine and start codes).
+final class H264Depacketizer {
+    static let fuA = 28
+
+    private var lastSequenceNumber: Int?
+    private var frameStarted = false
+    private var frameDamaged = false
+    private var frameTimestamp = 0
+    // Growable byte accumulators: keyframes reach hundreds of KB.
+    private var accessUnit: [UInt8] = []
+    private var fuActive = false
+    private var fuNal: [UInt8] = []
+
+    /// Feed one packet (in sequence order, from the jitter buffer).
+    func depacketize(_ packet: RtpPacket) -> DepacketizeResult {
+        var result = DepacketizeResult()
+
+        // Track sequence continuity: a gap means packets were lost.
+        if let last = lastSequenceNumber {
+            let expected = (last + 1) & 0xFFFF
+            if packet.header.sequenceNumber != expected {
+                fuActive = false
+                fuNal.removeAll(keepingCapacity: true)
+                if frameStarted { frameDamaged = true }
+            }
+        }
+        lastSequenceNumber = packet.header.sequenceNumber
+
+        // A timestamp change without a closing marker means the previous frame
+        // lost its tail and can no longer be recovered.
+        if frameStarted && packet.header.timestamp != frameTimestamp {
+            dropFrame()
+            result.frameDropped = true
+        }
+        if !frameStarted {
+            frameStarted = true
+            frameDamaged = false
+            frameTimestamp = packet.header.timestamp
+            accessUnit.removeAll(keepingCapacity: true)
+        }
+
+        let payload = packet.payload
+        if !payload.isEmpty {
+            let type = Int(payload[0]) & 0x1F
+            if type >= 1 && type <= 23 {
+                // Single NAL unit packet.
+                if fuActive {
+                    frameDamaged = true
+                    fuActive = false
+                    fuNal.removeAll(keepingCapacity: true)
+                }
+                appendStartCode()
+                accessUnit.append(contentsOf: payload)
+            } else if type == Self.fuA {
+                if payload.count < 2 {
+                    frameDamaged = true
+                } else {
+                    let fuHeader = Int(payload[1])
+                    let start = fuHeader & 0x80 != 0
+                    let end = fuHeader & 0x40 != 0
+                    let fragment = Array(payload[2...])
+                    if start {
+                        if fuActive {
+                            // The previous fragmented NAL lost its end packet.
+                            frameDamaged = true
+                        }
+                        fuActive = true
+                        fuNal.removeAll(keepingCapacity: true)
+                        // The FU indicator keeps the original NAL's F bit (0) and NRI,
+                        // and declares type 28; the FU header carries S/E plus the real type.
+                        fuNal.append(UInt8((Int(payload[0]) & 0xE0) | (fuHeader & 0x1F)))
+                        fuNal.append(contentsOf: fragment)
+                    } else if !fuActive {
+                        // Continuation without a start: the head of the NAL is lost.
+                        frameDamaged = true
+                    } else {
+                        fuNal.append(contentsOf: fragment)
+                        if end {
+                            appendStartCode()
+                            accessUnit.append(contentsOf: fuNal)
+                            fuActive = false
+                            fuNal.removeAll(keepingCapacity: true)
+                        }
+                    }
+                }
+            } else {
+                // Unsupported packetization mode (STAP-A, MTAP, FU-B).
+                frameDamaged = true
+            }
+        }
+
+        if !packet.header.marker {
+            return result
+        }
+
+        if fuActive {
+            // The marker arrived while a NAL was still fragmented.
+            frameDamaged = true
+            fuActive = false
+            fuNal.removeAll(keepingCapacity: true)
+        }
+
+        if !frameDamaged && !accessUnit.isEmpty {
+            let unit = accessUnit
+            result.accessUnit = unit
+            result.isKeyFrame = Self.containsParameterSets(unit)
+        } else {
+            // The frame that just ended is unusable.
+            result.frameDropped = true
+        }
+        dropFrame()
+        return result
+    }
+
+    private func appendStartCode() {
+        accessUnit.append(0)
+        accessUnit.append(0)
+        accessUnit.append(1)
+    }
+
+    /// The sender repeats SPS/PPS in-band at every keyframe; sniff for NAL types 7/8.
+    private static func containsParameterSets(_ unit: [UInt8]) -> Bool {
+        guard unit.count >= 4 else { return false }
+        var i = 0
+        while i <= unit.count - 4 {
+            if unit[i] == 0 && unit[i + 1] == 0 && unit[i + 2] == 1 {
+                let nalType = Int(unit[i + 3]) & 0x1F
+                if nalType == 7 || nalType == 8 {
+                    return true
+                }
+            }
+            i += 1
+        }
+        return false
+    }
+
+    private func dropFrame() {
+        frameStarted = false
+        frameDamaged = false
+        accessUnit.removeAll(keepingCapacity: true)
+        fuActive = false
+        fuNal.removeAll(keepingCapacity: true)
+    }
+}
@@ -0,0 +1,83 @@
+import Foundation
+
+/// Reorders RTP packets by sequence number before depacketization so that a
+/// reordering link (Wi-Fi) is not read as loss. Delivery stays in order; only
+/// aged-out or overflowing buffers release out of order, which the downstream
+/// gap detection still handles for genuine loss.
+///
+/// Mirrors the C++ `RtpJitterBuffer` (same defaults and semantics) and the
+/// Kotlin receiver.
+final class JitterBuffer {
+    private struct Entry {
+        let timeNs: Int64
+        let packet: RtpPacket
+    }
+
+    private let maxDepth: Int
+    private let maxDelayNs: Int64
+    private var buffer: [Int: Entry] = [:]
+    private var nextExpected: Int?
+
+    init(maxDepth: Int = 16, maxDelayMs: Int = 60) {
+        self.maxDepth = maxDepth
+        self.maxDelayNs = Int64(maxDelayMs) * 1_000_000
+    }
+
+    /// Insert one packet and return the packets now ready for in-order delivery.
+    func push(_ packet: RtpPacket) -> [RtpPacket] {
+        var released: [RtpPacket] = []
+        let sequence = packet.header.sequenceNumber
+        let now = Self.nowNanos()
+
+        let expected0: Int
+        if let e = nextExpected {
+            expected0 = e
+        } else {
+            expected0 = sequence
+            nextExpected = sequence
+        }
+
+        // Serial-number comparison: a distance >= 32768 means the packet is
+        // older than what we already delivered (duplicate or straggler).
+        let distance = (sequence - expected0 + 65536) % 65536
+        if distance < 32768 {
+            buffer[sequence] = Entry(timeNs: now, packet: packet)
+
+            // Release the consecutive run from the expected sequence.
+            var expected = expected0
+            while let entry = buffer[expected] {
+                released.append(entry.packet)
+                buffer.removeValue(forKey: expected)
+                expected = (expected + 1) & 0xFFFF
+            }
+            nextExpected = expected
+
+            // A missing packet stalls the run: age out the backlog (or bound
+            // the buffer) and release what is there in order, so genuine loss
+            // reaches the depacketizer's gap detection rather than blocking.
+            if !buffer.isEmpty {
+                let keys = buffer.keys.sorted()
+                let head = keys.first!
+                let headAgeNs = now - buffer[head]!.timeNs
+                if headAgeNs > maxDelayNs || buffer.count > maxDepth {
+                    for key in keys {
+                        released.append(buffer[key]!.packet)
+                    }
+                    nextExpected = (keys.last! + 1) & 0xFFFF
+                    buffer.removeAll(keepingCapacity: true)
+                }
+            }
+        }
+        return released
+    }
+
+    /// Discard everything still buffered.
+    func clear() {
+        buffer.removeAll(keepingCapacity: true)
+        nextExpected = nil
+    }
+
+    private static func nowNanos() -> Int64 {
+        return Int64(DispatchTime.now().uptimeNanoseconds)
+    }
+}
@@ -0,0 +1,29 @@
+import Foundation
+
+/// The in-band SPS (type 7) and PPS (type 8) NAL units extracted from an
+/// access unit. The sender repeats both ahead of every keyframe, so any
+/// keyframe carries them; they are the source for the VideoToolbox format
+/// description.
+struct NalSets {
+    let sps: [UInt8]
+    let pps: [UInt8]
+}
+
+/// Extracts parameter sets from an Annex-B access unit.
+enum NalExtractor {
+    static func parameterSets(_ annexB: [UInt8]) -> NalSets? {
+        var sps: [UInt8]?
+        var pps: [UInt8]?
+        for nal in AvccConverter.nalUnits(annexB) {
+            guard !nal.isEmpty else { continue }
+            let type = Int(nal[0]) & 0x1F
+            if type == 7 && sps == nil {
+                sps = nal
+            } else if type == 8 && pps == nil {
+                pps = nal
+            }
+        }
+        guard let s = sps, let p = pps else { return nil }
+        return NalSets(sps: s, pps: p)
+    }
+}
@@ -0,0 +1,56 @@
+import Foundation
+
+/// Minimal RTP header (RFC 3550) without extensions, mirroring the C++
+/// `RtpHeader` and the Kotlin receiver.
+struct RtpHeader: Equatable {
+    var version = 2
+    var padding = false
+    var extensionHeader = false
+    var csrcCount = 0
+    var marker = false
+    var payloadType = 96
+    var sequenceNumber = 0
+    var timestamp = 0
+    var ssrc = 0
+
+    /// Serializes the bare 12-byte header; requires a version-2, extension-less header.
+    func serialize() -> [UInt8] {
+        var out = [UInt8](repeating: 0, count: 12)
+        out[0] = UInt8((version & 0x0F) << 6
+                        | (padding ? 0x20 : 0)
+                        | (extensionHeader ? 0x10 : 0)
+                        | (csrcCount & 0x0F))
+        out[1] = UInt8((marker ? 0x80 : 0) | (payloadType & 0x7F))
+        out[2] = UInt8((sequenceNumber >> 8) & 0xFF)
+        out[3] = UInt8(sequenceNumber & 0xFF)
+        out[4] = UInt8((timestamp >> 24) & 0xFF)
+        out[5] = UInt8((timestamp >> 16) & 0xFF)
+        out[6] = UInt8((timestamp >> 8) & 0xFF)
+        out[7] = UInt8(timestamp & 0xFF)
+        out[8] = UInt8((ssrc >> 24) & 0xFF)
+        out[9] = UInt8((ssrc >> 16) & 0xFF)
+        out[10] = UInt8((ssrc >> 8) & 0xFF)
+        out[11] = UInt8(ssrc & 0xFF)
+        return out
+    }
+
+    /// Parses a 12-byte header from the start of a datagram.
+    static func parse(_ input: [UInt8]) -> RtpHeader? {
+        guard input.count >= 12 else { return nil }
+        let b0 = Int(input[0])
+        let b1 = Int(input[1])
+        let version = b0 >> 6
+        guard version == 2 else { return nil }
+        return RtpHeader(
+            version: version,
+            padding: b0 & 0x20 != 0,
+            extensionHeader: b0 & 0x10 != 0,
+            csrcCount: b0 & 0x0F,
+            marker: b1 & 0x80 != 0,
+            payloadType: b1 & 0x7F,
+            sequenceNumber: (Int(input[2]) << 8) | Int(input[3]),
+            timestamp: (Int(input[4]) << 24) | (Int(input[5]) << 16) | (Int(input[6]) << 8) | Int(input[7]),
+            ssrc: (Int(input[8]) << 24) | (Int(input[9]) << 16) | (Int(input[10]) << 8) | Int(input[11]),
+        )
+    }
+}
@@ -0,0 +1,38 @@
+import Foundation
+
+/// An RTP packet: 12-byte base header (plus optional CSRC/extension) and payload.
+struct RtpPacket {
+    let header: RtpHeader
+    let payload: [UInt8]
+
+    init(header: RtpHeader, payload: [UInt8]) {
+        self.header = header
+        self.payload = payload
+    }
+
+    /// Parses a full RTP datagram. Honors CSRC lists, one-level extension
+    /// headers, and RFC 3550 padding, mirroring the C++ receiver.
+    static func parse(_ input: [UInt8]) -> RtpPacket? {
+        guard input.count >= 12, let header = RtpHeader.parse(input) else { return nil }
+
+        var offset = 12 + header.csrcCount * 4
+        guard input.count >= offset else { return nil }
+
+        if header.extensionHeader {
+            guard input.count >= offset + 4 else { return nil }
+            let extensionWords = (Int(input[offset + 2]) << 8) | Int(input[offset + 3])
+            offset += 4 + extensionWords * 4
+            guard input.count >= offset else { return nil }
+        }
+
+        var payloadSize = input.count - offset
+        if header.padding {
+            // RFC 3550: the last byte holds the padding size, including itself.
+            guard payloadSize > 0 else { return nil }
+            let paddingSize = Int(input[input.count - 1])
+            guard paddingSize != 0, paddingSize <= payloadSize else { return nil }
+            payloadSize -= paddingSize
+        }
+        return RtpPacket(header: header, payload: Array(input[offset..<(offset + payloadSize)]))
+    }
+}
@@ -0,0 +1,33 @@
+import Foundation
+
+/// Splits a byte stream into newline-terminated lines, dropping CR and
+/// enforcing the 64 KB line cap — the same framing the C++ and Kotlin
+/// receivers use. Extracted so the logic is testable in isolation.
+final class LineAssembler {
+    private var pending: [UInt8] = []
+    private let maxLine = SignalingMessage.maxMessageBytes
+
+    /// Feed a chunk of received bytes; returns the complete lines it contained.
+    func feed(_ chunk: [UInt8]) -> [String] {
+        var lines: [String] = []
+        for b in chunk {
+            if b == 0x0A { // \n
+                let text = String(bytes: pending, encoding: .utf8) ?? ""
+                pending.removeAll(keepingCapacity: true)
+                if !text.isEmpty { lines.append(text) }
+            } else if b != 0x0D { // \r
+                if pending.count < maxLine {
+                    pending.append(b)
+                } else {
+                    // Hostile or broken peer: drop the oversized line.
+                    pending.removeAll(keepingCapacity: true)
+                }
+            }
+        }
+        return lines
+    }
+
+    func reset() {
+        pending.removeAll(keepingCapacity: true)
+    }
+}
@@ -0,0 +1,89 @@
+import Foundation
+
+/// JSON wire format shared with the C++ implementation: one JSON object per
+/// newline-terminated TCP line (offer / answer / pli). Mirrors the C++
+/// `SignalingMessage` and the Kotlin receiver.
+enum SignalingMessage {
+    case offer(sessionId: String,
+               codec: String,
+               width: Int,
+               height: Int,
+               frameRateNum: Int,
+               frameRateDen: Int,
+               rtpAddress: String,
+               rtpPort: Int)
+    case answer(sessionId: String,
+                rtpAddress: String,
+                rtpPort: Int,
+                displayWidth: Int,
+                displayHeight: Int)
+    case pli(sessionId: String)
+
+    static let maxMessageBytes = 64 * 1024
+
+    static func parse(_ line: String) -> SignalingMessage? {
+        guard line.count <= maxMessageBytes else { return nil }
+        guard let data = line.data(using: .utf8),
+              let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
+              let type = obj["type"] as? String else {
+            return nil
+        }
+        switch type {
+        case "offer":
+            return .offer(
+                sessionId: obj["session_id"] as? String ?? "",
+                codec: obj["codec"] as? String ?? "",
+                width: obj["width"] as? Int ?? 0,
+                height: obj["height"] as? Int ?? 0,
+                frameRateNum: obj["frame_rate_num"] as? Int ?? 30,
+                frameRateDen: obj["frame_rate_den"] as? Int ?? 1,
+                rtpAddress: obj["rtp_address"] as? String ?? "",
+                rtpPort: obj["rtp_port"] as? Int ?? 0)
+        case "answer":
+            return .answer(
+                sessionId: obj["session_id"] as? String ?? "",
+                rtpAddress: obj["rtp_address"] as? String ?? "",
+                rtpPort: obj["rtp_port"] as? Int ?? 0,
+                displayWidth: obj["display_width"] as? Int ?? 0,
+                displayHeight: obj["display_height"] as? Int ?? 0)
+        case "pli":
+            return .pli(sessionId: obj["session_id"] as? String ?? "")
+        default:
+            return nil
+        }
+    }
+
+    static func serialize(_ message: SignalingMessage) -> String {
+        let json: [String: Any]
+        switch message {
+        case let .offer(sessionId, codec, width, height, frameRateNum, frameRateDen, rtpAddress, rtpPort):
+            json = [
+                "type": "offer",
+                "session_id": sessionId,
+                "codec": codec,
+                "width": width,
+                "height": height,
+                "frame_rate_num": frameRateNum,
+                "frame_rate_den": frameRateDen,
+                "rtp_address": rtpAddress,
+                "rtp_port": rtpPort,
+            ]
+        case let .answer(sessionId, rtpAddress, rtpPort, displayWidth, displayHeight):
+            json = [
+                "type": "answer",
+                "session_id": sessionId,
+                "rtp_address": rtpAddress,
+                "rtp_port": rtpPort,
+                "display_width": displayWidth,
+                "display_height": displayHeight,
+            ]
+        case let .pli(sessionId):
+            json = ["type": "pli", "session_id": sessionId]
+        }
+        guard let data = try? JSONSerialization.data(withJSONObject: json),
+              let s = String(data: data, encoding: .utf8) else {
+            return "{}\n" // unreachable for our message types; serialize is total
+        }
+        return s + "\n"
+    }
+}
@@ -0,0 +1,212 @@
+import Foundation
+#if canImport(Darwin)
+import Darwin
+#endif
+
+/// Small socket helpers shared by the signaling server and the UDP transport.
+enum SocketUtils {
+    /// Creates a bound, listening TCP socket. Prefers a dual-stack IPv6
+    /// listener (both families), falls back to IPv4-only. Returns -1 on failure.
+    static func makeStreamListener(port: UInt16) -> Int32 {
+        for family in [AF_INET6, AF_INET] {
+            let fd = socket(family, SOCK_STREAM, 0)
+            guard fd >= 0 else { continue }
+            var yes: Int32 = 1
+            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout<Int32>.size))
+            if family == AF_INET6 {
+                var no: Int32 = 0
+                setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, socklen_t(MemoryLayout<Int32>.size))
+            }
+
+            let bound: Int32
+            if family == AF_INET6 {
+                var a = sockaddr_in6()
+                a.sin6_family = sa_family_t(AF_INET6)
+                a.sin6_port = port.bigEndian
+                a.sin6_addr = in6addr_any
+                bound = withUnsafePointer(to: &a) { p in
+                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in6>.size)) }
+                }
+            } else {
+                var a = sockaddr_in()
+                a.sin_family = sa_family_t(AF_INET)
+                a.sin_port = port.bigEndian
+                a.sin_addr = in_addr(s_addr: INADDR_ANY)
+                bound = withUnsafePointer(to: &a) { p in
+                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in>.size)) }
+                }
+            }
+            if bound != 0 { close(fd); continue }
+            if listen(fd, 16) != 0 { close(fd); continue }
+            return fd
+        }
+        return -1
+    }
+
+    /// The local port of a bound socket (the port is at byte offset 2 for both
+    /// IPv4 and IPv6 sockets).
+    static func boundPort(_ fd: Int32) -> UInt16 {
+        var a = sockaddr_storage()
+        var len = socklen_t(MemoryLayout<sockaddr_storage>.size)
+        guard getsockname(fd, &a, &len) == 0 else { return 0 }
+        return withUnsafeBytes(of: &a) { raw in
+            raw.load(fromByteOffset: 2, as: UInt16.self).bigEndian
+        }
+    }
+
+    /// A UDP socket bound to [port] (or 0 for ephemeral) for receiving, with a
+    /// generous receive buffer (the C++ sender's VBV bounds bursts, but a larger
+    /// buffer absorbs a burst on a lossy link).
+    static func makeUdpReceiver(port: UInt16, receiveBufferSize: Int32) -> Int32 {
+        for family in [AF_INET6, AF_INET] {
+            let fd = socket(family, SOCK_DGRAM, 0)
+            guard fd >= 0 else { continue }
+            var yes: Int32 = 1
+            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout<Int32>.size))
+            if family == AF_INET6 {
+                var no: Int32 = 0
+                setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &no, socklen_t(MemoryLayout<Int32>.size))
+            }
+            let bound: Int32
+            if family == AF_INET6 {
+                var a = sockaddr_in6()
+                a.sin6_family = sa_family_t(AF_INET6)
+                a.sin6_port = port.bigEndian
+                a.sin6_addr = in6addr_any
+                bound = withUnsafePointer(to: &a) { p in
+                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in6>.size)) }
+                }
+            } else {
+                var a = sockaddr_in()
+                a.sin_family = sa_family_t(AF_INET)
+                a.sin_port = port.bigEndian
+                a.sin_addr = in_addr(s_addr: INADDR_ANY)
+                bound = withUnsafePointer(to: &a) { p in
+                    p.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in>.size)) }
+                }
+            }
+            if bound != 0 { close(fd); continue }
+            setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &receiveBufferSize, socklen_t(MemoryLayout<Int32>.size))
+            return fd
+        }
+        return -1
+    }
+}
+
+/// Newline-delimited JSON signaling server (the receiver side). Keeps the most
+/// recent connection as its active peer, mirroring the C++ server: `onOffer`
+/// may answer synchronously (the sender blocks on the answer).
+final class SignalingServer {
+    private let onOffer: (SignalingMessage) -> Void
+    private let onPli: (SignalingMessage) -> Void
+    private var listenFd: Int32 = -1
+    private(set) var port: UInt16 = 0
+    private var peerFd: Int32 = -1
+    private let peerLock = NSLock()
+    private let assembler = LineAssembler()
+    private var acceptThread: Thread?
+    private var readerThread: Thread?
+    private var running = false
+
+    init(onOffer: @escaping (SignalingMessage) -> Void,
+         onPli: @escaping (SignalingMessage) -> Void) {
+        self.onOffer = onOffer
+        self.onPli = onPli
+    }
+
+    /// Binds the port (SO_REUSEADDR) and starts accepting. Returns true on success.
+    func start(_ preferredPort: UInt16) -> Bool {
+        guard let fd = SocketUtils.makeStreamListener(port: preferredPort), fd >= 0 else { return false }
+        listenFd = fd
+        port = SocketUtils.boundPort(fd)
+        running = true
+        let thread = Thread { [weak self] in self?.acceptLoop() }
+        thread.name = "signaling-accept"
+        acceptThread = thread
+        thread.start()
+        return true
+    }
+
+    /// Sends to the current peer; never throws. A lost control message is
+    /// recoverable — the session re-negotiates or the next keyframe arrives —
+    /// but an exception here would kill the RTP reader thread that reaches
+    /// send() from requestPli(). Mirrors the C++ server, which ignores write
+    /// failures.
+    func send(_ message: SignalingMessage) {
+        guard let bytes = SignalingMessage.serialize(message).data(using: .utf8) else { return }
+        peerLock.lock()
+        let fd = peerFd
+        peerLock.unlock()
+        guard fd >= 0 else { return }
+        bytes.withUnsafeBytes { raw in
+            _ = send(fd, raw.baseAddress, raw.count, Int32(MSG_NOSIGNAL))
+        }
+    }
+
+    func close() {
+        running = false
+        peerLock.lock()
+        let peer = peerFd
+        peerFd = -1
+        peerLock.unlock()
+        if peer >= 0 { close(peer) }
+        if listenFd >= 0 { close(listenFd) }
+        listenFd = -1
+    }
+
+    private func acceptLoop() {
+        while running {
+            var addr = sockaddr()
+            var len = socklen_t(MemoryLayout<sockaddr>.size)
+            let client = accept(listenFd, &addr, &len)
+            if client < 0 {
+                if !running { break }
+                continue
+            }
+            // Most-recent-connection-wins: close the previous peer.
+            peerLock.lock()
+            let old = peerFd
+            peerFd = client
+            peerLock.unlock()
+            if old >= 0 { close(old) }
+
+            let thread = Thread { [weak self] in self?.readLoop(fd: client) }
+            thread.name = "signaling-reader"
+            readerThread = thread
+            thread.start()
+        }
+    }
+
+    private func readLoop(fd: Int32) {
+        assembler.reset()
+        var buffer = [UInt8](repeating: 0, count: 4096)
+        while running {
+            let read = buffer.withUnsafeMutableBytes { raw in
+                recv(fd, raw.baseAddress, raw.count, 0)
+            }
+            guard read > 0 else { break }
+            for line in assembler.feed(Array(buffer.prefix(read))) {
+                dispatch(line)
+            }
+        }
+        // Peer closed; if it is still our active peer, mark it gone.
+        peerLock.lock()
+        if peerFd == fd {
+            peerFd = -1
+        }
+        peerLock.unlock()
+    }
+
+    private func dispatch(_ line: String) {
+        guard let message = SignalingMessage.parse(line) else { return }
+        // A broken callback must not kill the reader thread.
+        switch message {
+        case .offer:
+            onOffer(message)
+        case .pli:
+            onPli(message)
+        case .answer:
+            break // the receiver never receives answers
+        }
+    }
+}
@@ -0,0 +1,41 @@
+import Foundation
+#if canImport(Darwin)
+import Darwin
+#endif
+
+/// Finds the primary IPv4 address (e.g. on Wi-Fi) for the status overlay and
+/// the `--peer` hint. Best-effort; the sender reaches us by IP over the LAN.
+enum LocalAddress {
+    static func primaryIPv4() -> String {
+        var result = "unknown"
+        var fallback = "unknown"
+        var ptr: UnsafeMutablePointer<ifaddrs>?
+        guard getifaddrs(&ptr) == 0 else { return result }
+        defer { freeifaddrs(ptr) }
+
+        var current = ptr
+        while let iface = current {
+            let next = iface.pointee.ifa_next
+            current = next
+
+            guard let sa = iface.pointee.ifa_addr else { continue }
+            guard sa.pointee.sa_family == sa_family_t(AF_INET) else { continue }
+            if Int32(iface.pointee.ifa_flags) & IFF_LOOPBACK == 0 {
+                let inaddr = sa.assumingMemoryBound(to: sockaddr_in.self).pointee
+                var host = [CChar](repeating: 0, count: Int(INET_ADDRSTRLEN))
+                if inet_ntop(AF_INET, &inaddr.sin_addr, &host, socklen_t(INET_ADDRSTRLEN)) != nil {
+                    let ip = String(cString: host)
+                    if ip.hasPrefix("169.254") {
+                        continue // link-local; prefer a routable address
+                    }
+                    fallback = ip
+                    let name = String(cString: iface.pointee.ifa_name)
+                    if name.hasPrefix("en") || name.hasPrefix("wlan") {
+                        return ip // Wi-Fi / Ethernet: good enough for the hint
+                    }
+                }
+            }
+        }
+        return fallback
+    }
+}
@@ -0,0 +1,59 @@
+import Foundation
+#if canImport(Darwin)
+import Darwin
+#endif
+
+/// A UDP socket for receiving RTP. Binds to a preferred port (or an ephemeral
+/// port when it is busy), then delivers complete datagrams one at a time.
+/// Mirrors the C++ `UdpRtpTransport` receive path and the Kotlin `DatagramSocket`
+/// reader. Uses poll(2) with a timeout so `close()` from another thread cannot
+/// strand a blocked recv (a close does not reliably unblock a POSIX recvfrom).
+final class UdpTransport {
+    private var fd: Int32 = -1
+    private(set) var port: UInt16 = 0
+
+    /// Binds to [preferredPort] (or an ephemeral port when it is busy).
+    @discardableResult
+    func bind(preferredPort: UInt16, receiveBufferSize: Int32) -> Bool {
+        guard fd < 0 else { return true }
+        for p in [preferredPort, UInt16(0)] {
+            if let f = SocketUtils.makeUdpReceiver(port: p, receiveBufferSize: receiveBufferSize), f >= 0 {
+                fd = f
+                port = SocketUtils.boundPort(fd)
+                return true
+            }
+        }
+        return false
+    }
+
+    /// Waits up to [timeoutMs] for a datagram. Returns 1 when one is ready to
+    /// read, 0 on timeout, and -1 when the socket is closed/errored.
+    func poll(timeoutMs: Int32) -> Int32 {
+        guard fd >= 0 else { return -1 }
+        var pfd = pollfd(fd: fd, events: poll_events_t(POLLIN), revents: 0)
+        let r = withUnsafeMutablePointer(to: &pfd) { poll($0, 1, timeoutMs) }
+        if r > 0 {
+            if pfd.revents & poll_events_t(POLLERR) != 0 { return -1 }
+            return 1
+        }
+        return r == 0 ? 0 : -1
+    }
+
+    /// Reads one ready datagram. Returns nil on error or empty read.
+    func receiveDatagram() -> [UInt8]? {
+        guard fd >= 0 else { return nil }
+        var buf = [UInt8](repeating: 0, count: 4096)
+        let read = buf.withUnsafeMutableBytes { raw in
+            recvfrom(fd, raw.baseAddress, raw.count, 0, nil, nil)
+        }
+        guard read > 0 else { return nil }
+        return Array(buf.prefix(read))
+    }
+
+    func close() {
+        if fd >= 0 {
+            close(fd)
+            fd = -1
+        }
+    }
+}