driver-red: rewrite to verified RCP2 — WebSocket :9998/rcp, rcp_config handshake, rcp_get/set, cur.val responses, per-channel CDL params, unsolicited notifications; sim rewritten as WS server; gang handles CDL-only cameras (no live LUT on RCP2) w/ representability check; Lut3D.isApproximatelyIdentity — 187 tests

Package.swift

@@ -20,6 +20,7 @@ let package = Package(
             "ForgeCamera",
             .product(name: "NIO", package: "swift-nio"),
             .product(name: "NIOHTTP1", package: "swift-nio"),
+            .product(name: "NIOWebSocket", package: "swift-nio"),
         ]),
         // Vendor drivers
         .target(name: "ForgeCameraARRI", dependencies: [
@@ -30,6 +31,8 @@ let package = Package(
         .target(name: "ForgeCameraRED", dependencies: [
             "ForgeCamera",
             .product(name: "NIO", package: "swift-nio"),
+            .product(name: "NIOHTTP1", package: "swift-nio"),
+            .product(name: "NIOWebSocket", package: "swift-nio"),
         ]),
         .target(name: "ForgeCameraSony", dependencies: [
             "ForgeCamera",

Sources/ForgeCamera/GradeGang.swift

@@ -82,6 +82,26 @@ public actor GradeGang {
             guard let driver = drivers[target] else { continue }
             let caps = driver.capabilities
             do {
+                if caps.lut3dSizes.isEmpty {
+                    // CDL-only camera (RED over RCP2: no live LUT upload).
+                    // Split CDL out; if the baked remainder is non-identity the
+                    // camera cannot reproduce the grade — hard error so the
+                    // operator knows monitor ≠ grade.
+                    guard caps.supportsNativeCDL else {
+                        throw CameraError.unsupportedLook("camera accepts neither LUTs nor CDL")
+                    }
+                    let look = Flattener.flatten(
+                        stack: grade,
+                        latticeSize: 17, // cheap lattice, only used for the identity check
+                        splitLeadingCDL: true)
+                    guard look.cdl != nil else {
+                        throw CameraError.unsupportedLook("grade has no leading CDL; CDL-only camera cannot represent it")
+                    }
+                    if let lut = look.lut, !lut.isApproximatelyIdentity() {
+                        throw CameraError.unsupportedLook("grade has nodes beyond CDL; CDL-only camera cannot represent them")
+                    }
+                    try await driver.push(look: FlattenedLook(cdl: look.cdl, lut: nil, latticeSize: 0))
+                } else {
                     // Prefer exact requested lattice; else largest supported size
                     // *below* it (downgrade only — baking a 17³-quality grade into a
                     // 99³ lattice wastes upload time and implies false precision).
@@ -95,6 +115,7 @@ public actor GradeGang {
                         latticeSize: size,
                         splitLeadingCDL: caps.supportsNativeCDL)
                     try await driver.push(look: look)
+                }
             } catch {
                 failures.append((slot: target, error: error))
             }

Sources/ForgeCameraRED/RedDriver.swift

@@ -1,44 +1,59 @@
 import Foundation
+import NIO
+import NIOHTTP1
+import NIOWebSocket
 import ForgeCamera
 import ForgeGrade
 import ForgeColor
 
-#if canImport(Glibc)
+/// RCP2 param IDs (wire form: RCP_PARAM_ prefix stripped).
-import Glibc
+/// Verified against RED parameter docs via docs/research/red-rcp2-protocol.md.
-#endif
-
-/// RED parameter names — single source of truth, correct against RED SDK
-/// (RCP2) when hardware/SDK agreement lands. Sim mirrors these.
 enum RedParams {
-    static let recordState = "RECORD_STATE"
+    static let recordState = "RECORD_STATE"      // cur.val: 0 idle, 1 recording, set "2" = toggle
     static let clipName = "CLIP_NAME"
     static let iso = "ISO"
-    static let colorTemp = "COLOR_TEMP"
+    static let colorTemp = "COLOR_TEMPERATURE"
     static let tint = "TINT"
-    static let sensorFPS = "SENSOR_FPS"
+    static let sensorFPS = "SENSOR_FRAME_RATE"
     static let timecode = "TIMECODE"
-    static let ipp2CDL = "IPP2_CDL"
+    static let cdlEnable = "CDL_ENABLE"
+    static let cdlSlopeR = "CDL_SLOPE_RED"
+    static let cdlSlopeG = "CDL_SLOPE_GREEN"
+    static let cdlSlopeB = "CDL_SLOPE_BLUE"
+    static let cdlOffsetR = "CDL_OFFSET_RED"
+    static let cdlOffsetG = "CDL_OFFSET_GREEN"
+    static let cdlOffsetB = "CDL_OFFSET_BLUE"
+    static let cdlPowerR = "CDL_POWER_RED"
+    static let cdlPowerG = "CDL_POWER_GREEN"
+    static let cdlPowerB = "CDL_POWER_BLUE"
+    static let cdlSaturation = "CDL_SATURATION"
 }
 
-/// DSMC3 driver: RCP2-style length-prefixed JSON over TCP.
+/// DSMC3 driver — real RCP2: WebSocket ws://<ip>:9998/rcp, JSON text frames.
-/// Request/response serialized through the actor — one in-flight request.
+/// Handshake: rcp_config w/ client name. CDL pushed as per-channel rcp_set ops.
+/// NO live LUT upload (RCP2 limitation) — capabilities.lut3dSizes empty; grade
+/// pipeline keeps camera-native CDL only, remainder must stay identity.
 public actor RedDriver: CameraDriver {
     public nonisolated let capabilities = CameraCapabilities(
         vendor: .red,
         supportsNativeCDL: true,
-        lut3dSizes: [17, 33],
+        lut3dSizes: [],
         metadataFields: [.clipName, .exposureIndex, .whiteBalance, .tint, .fps, .timecode])
 
     private let host: String
     private let port: Int
     private let pollInterval: Duration
 
-    private var socketFD: Int32 = -1
+    private var group: MultiThreadedEventLoopGroup?
+    private var channel: Channel?
     private var pollTask: Task<Void, Never>?
+
+    /// Latest values per param, merged into CameraState on any change.
+    private var paramCache: [String: String] = [:]
     private var lastState = CameraState(connection: .disconnected)
     private var stateContinuations: [UUID: AsyncStream<CameraState>.Continuation] = [:]
 
-    public init(host: String, port: Int, pollInterval: Duration = .milliseconds(150)) {
+    public init(host: String, port: Int = 9998, pollInterval: Duration = .milliseconds(150)) {
         self.host = host
         self.port = port
         self.pollInterval = pollInterval
@@ -69,111 +84,71 @@ public actor RedDriver: CameraDriver {
         }
     }
 
-    // MARK: Socket transport (blocking, actor-serialized, short timeouts)
+    // MARK: WebSocket transport
-
-    private func openSocket() throws {
-        let fd = socket(AF_INET, Int32(SOCK_STREAM.rawValue), 0)
-        guard fd >= 0 else { throw CameraError.connectionFailed("socket() failed") }
-
-        var tv = timeval(tv_sec: 3, tv_usec: 0)
-        setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, socklen_t(MemoryLayout<timeval>.size))
-        setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, socklen_t(MemoryLayout<timeval>.size))
-
-        var addr = sockaddr_in()
-        addr.sin_family = sa_family_t(AF_INET)
-        addr.sin_port = in_port_t(UInt16(port).bigEndian)
-        guard inet_pton(AF_INET, host, &addr.sin_addr) == 1 else {
-            close(fd)
-            throw CameraError.connectionFailed("bad host \(host)")
-        }
-        let rc = withUnsafePointer(to: &addr) { ptr in
-            ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in
-                Glibc.connect(fd, sa, socklen_t(MemoryLayout<sockaddr_in>.size))
-            }
-        }
-        guard rc == 0 else {
-            close(fd)
-            throw CameraError.connectionFailed("connect to \(host):\(port) failed")
-        }
-        socketFD = fd
-    }
-
-    private func closeSocket() {
-        if socketFD >= 0 {
-            close(socketFD)
-            socketFD = -1
-        }
-    }
-
-    private func sendAll(_ data: Data) throws {
-        var sent = 0
-        try data.withUnsafeBytes { (buf: UnsafeRawBufferPointer) in
-            while sent < buf.count {
-                let n = write(socketFD, buf.baseAddress!.advanced(by: sent), buf.count - sent)
-                guard n > 0 else { throw CameraError.connectionFailed("write failed") }
-                sent += n
-            }
-        }
-    }
-
-    private func recvExact(_ count: Int) throws -> Data {
-        var out = Data(capacity: count)
-        var remaining = count
-        var chunk = [UInt8](repeating: 0, count: 64 * 1024)
-        while remaining > 0 {
-            let n = read(socketFD, &chunk, min(remaining, chunk.count))
-            guard n > 0 else { throw CameraError.connectionFailed("read failed/timeout") }
-            out.append(contentsOf: chunk[0..<n])
-            remaining -= n
-        }
-        return out
-    }
-
-    /// Send framed JSON request, await framed JSON reply.
-    private func request(_ obj: [String: Any]) throws -> [String: Any] {
-        guard socketFD >= 0 else { throw CameraError.connectionFailed("not connected") }
-        let payload = try JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys])
-        var frame = Data()
-        var lenBE = UInt32(payload.count).bigEndian
-        withUnsafeBytes(of: &lenBE) { frame.append(contentsOf: $0) }
-        frame.append(payload)
-        try sendAll(frame)
-
-        let header = try recvExact(4)
-        // Byte-wise assembly — Data slices may be misaligned for a u32 load.
-        let length = header.reduce(UInt32(0)) { ($0 << 8) | UInt32($1) }
-        let body = try recvExact(Int(length))
-        guard let reply = try JSONSerialization.jsonObject(with: body) as? [String: Any] else {
-            throw CameraError.connectionFailed("bad reply JSON")
-        }
-        return reply
-    }
-
-    private func getParam(_ name: String) throws -> String {
-        let reply = try request(["type": "get", "param": name])
-        guard reply["ok"] as? Bool == true, let value = reply["value"] as? String else {
-            throw CameraError.connectionFailed("get \(name) failed")
-        }
-        return value
-    }
-
-    private func setParam(_ name: String, _ value: String) throws {
-        let reply = try request(["type": "set", "param": name, "value": value])
-        guard reply["ok"] as? Bool == true else {
-            throw CameraError.pushFailed("set \(name) failed")
-        }
-    }
-
-    // MARK: CameraDriver
 
     public func connect() async throws {
-        closeSocket()
+        await teardown()
-        try openSocket()
+        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
-        let reply = try request(["type": "handshake"])
+        self.group = group
-        guard reply["ok"] as? Bool == true else {
+
-            closeSocket()
+        let upgradeResult: EventLoopPromise<Channel> = group.next().makePromise(of: Channel.self)
-            throw CameraError.connectionFailed("handshake rejected")
+        let driver = self
+
+        let upgrader = NIOWebSocketClientUpgrader(
+            requestKey: Data((0..<16).map { _ in UInt8.random(in: 0...255) }).base64EncodedString(),
+            upgradePipelineHandler: { channel, _ in
+                channel.pipeline.addHandler(RedClientWSHandler(driver: driver)).map {
+                    upgradeResult.succeed(channel)
                 }
+            })
+
+        let bootstrap = ClientBootstrap(group: group)
+            .channelInitializer { channel in
+                let upgradeConfig = NIOHTTPClientUpgradeConfiguration(
+                    upgraders: [upgrader],
+                    completionHandler: { _ in
+                        channel.pipeline.removeHandler(name: "sendInitialRequest", promise: nil)
+                    })
+                return channel.pipeline.addHTTPClientHandlers(withClientUpgrade: upgradeConfig).flatMap {
+                    channel.pipeline.addHandler(
+                        SendUpgradeRequestHandler(host: self.host, port: self.port, path: "/rcp"),
+                        name: "sendInitialRequest")
+                }
+            }
+
+        do {
+            let ch = try await bootstrap.connect(host: host, port: port).get()
+            _ = ch
+            // Wait for upgrade completion (or fail after timeout).
+            let upgraded: Channel = try await withThrowingTaskGroup(of: Channel.self) { grp in
+                grp.addTask {
+                    try await upgradeResult.futureResult.get()
+                }
+                grp.addTask {
+                    try await Task.sleep(for: .seconds(3))
+                    throw CameraError.connectionFailed("WS upgrade timeout")
+                }
+                let first = try await grp.next()!
+                grp.cancelAll()
+                return first
+            }
+            channel = upgraded
+        } catch {
+            upgradeResult.fail(error)
+            await teardown()
+            throw CameraError.connectionFailed("connect \(host):\(port): \(error)")
+        }
+
+        // RCP2 handshake.
+        try send([
+            "type": "rcp_config",
+            "strings_decoded": 0,
+            "json_minified": 1,
+            "include_cacheable_flags": 0,
+            "encoding_type": "html",
+            "client": ["name": "Forge"],
+        ])
+
         emit(CameraState(connection: .connected))
         startPolling()
     }
@@ -181,65 +156,183 @@ public actor RedDriver: CameraDriver {
     public func disconnect() async {
         pollTask?.cancel()
         pollTask = nil
-        closeSocket()
+        await teardown()
         emit(CameraState(connection: .disconnected))
     }
 
-    public func push(look: FlattenedLook) async throws {
+    private func teardown() async {
-        if let lut = look.lut, !capabilities.lut3dSizes.contains(lut.size) {
+        if let ch = channel {
-            throw CameraError.unsupportedLook("LUT size \(lut.size) not in \(capabilities.lut3dSizes)")
+            _ = try? await ch.close()
         }
-        if let cdl = look.cdl {
+        channel = nil
-            let obj: [String: Any] = [
+        if let g = group {
-                "slope": [cdl.slope.x, cdl.slope.y, cdl.slope.z],
+            try? await g.shutdownGracefully()
-                "offset": [cdl.offset.x, cdl.offset.y, cdl.offset.z],
-                "power": [cdl.power.x, cdl.power.y, cdl.power.z],
-                "saturation": cdl.saturation,
-            ]
-            let json = String(data: try JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys]), encoding: .utf8)!
-            try setParam(RedParams.ipp2CDL, json)
         }
-        if let lut = look.lut {
+        group = nil
-            let reply = try request(["type": "lut", "size": lut.size, "table": lut.table])
-            guard reply["ok"] as? Bool == true else {
-                throw CameraError.pushFailed("LUT upload rejected")
     }
+
+    private func send(_ obj: [String: Any]) throws {
+        guard let ch = channel else {
+            throw CameraError.connectionFailed("not connected")
+        }
+        let data = try JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys])
+        var buf = ch.allocator.buffer(capacity: data.count)
+        buf.writeBytes(data)
+        let frame = WebSocketFrame(fin: true, opcode: .text, maskKey: randomMask(), data: buf)
+        ch.writeAndFlush(NIOAny(frame), promise: nil)
+    }
+
+    private func randomMask() -> WebSocketMaskingKey {
+        WebSocketMaskingKey([UInt8.random(in: 0...255), .random(in: 0...255), .random(in: 0...255), .random(in: 0...255)])!
+    }
+
+    // MARK: Inbound (called from WS handler)
+
+    func handleInbound(_ data: Data) {
+        guard let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
+              let id = obj["id"] as? String else { return }
+        // Value under cur.val (int/double/string).
+        guard let cur = obj["cur"] as? [String: Any], let val = cur["val"] else { return }
+        let stringValue: String
+        if let s = val as? String {
+            stringValue = s
+        } else if let i = val as? Int {
+            stringValue = String(i)
+        } else if let d = val as? Double {
+            stringValue = String(d)
+        } else {
+            return
+        }
+        paramCache[id] = stringValue
+        rebuildState()
+    }
+
+    private func rebuildState() {
+        let newState = CameraState(
+            connection: .connected,
+            isRecording: paramCache[RedParams.recordState] == "1",
+            clipName: (paramCache[RedParams.clipName]?.isEmpty ?? true) ? nil : paramCache[RedParams.clipName],
+            metadata: CameraMetadata(
+                exposureIndex: paramCache[RedParams.iso].flatMap(Int.init),
+                whiteBalance: paramCache[RedParams.colorTemp].flatMap(Int.init),
+                tint: paramCache[RedParams.tint].flatMap(Int.init),
+                fps: paramCache[RedParams.sensorFPS].flatMap(Double.init),
+                timecode: paramCache[RedParams.timecode]))
+        if newState != lastState {
+            emit(newState)
         }
     }
 
-    // MARK: Polling
+    func handleChannelClosed() {
+        if lastState.connection == .connected {
+            emit(CameraState(connection: .disconnected))
+        }
+    }
+
+    // MARK: Polling (rcp_get; responses arrive via handleInbound)
 
     private func startPolling() {
         pollTask?.cancel()
         pollTask = Task {
+            let polled = [
+                RedParams.recordState, RedParams.clipName, RedParams.iso,
+                RedParams.colorTemp, RedParams.tint, RedParams.sensorFPS, RedParams.timecode,
+            ]
             while !Task.isCancelled {
-                pollOnce()
+                for p in polled {
+                    try? self.send(["type": "rcp_get", "id": p])
+                }
                 try? await Task.sleep(for: pollInterval)
             }
         }
     }
 
-    private func pollOnce() {
+    // MARK: Look push
-        do {
+
-            let rec = try getParam(RedParams.recordState) == "1"
+    public func push(look: FlattenedLook) async throws {
-            let clip = try getParam(RedParams.clipName)
+        // RCP2 cannot upload LUTs. Non-identity LUT = grade won't match on camera.
-            let newState = CameraState(
+        if let lut = look.lut, !lut.isApproximatelyIdentity() {
-                connection: .connected,
+            throw CameraError.unsupportedLook(
-                isRecording: rec,
+                "RED RCP2 cannot receive live 3D LUTs — use CDL-only grade or pre-load LUT on camera")
-                clipName: clip.isEmpty ? nil : clip,
-                metadata: CameraMetadata(
-                    exposureIndex: Int(try getParam(RedParams.iso)),
-                    whiteBalance: Int(try getParam(RedParams.colorTemp)),
-                    tint: Int(try getParam(RedParams.tint)),
-                    fps: Double(try getParam(RedParams.sensorFPS)),
-                    timecode: try getParam(RedParams.timecode)))
-            if newState != lastState {
-                emit(newState)
         }
-        } catch {
+        guard let cdl = look.cdl else {
-            if lastState.connection == .connected {
+            throw CameraError.unsupportedLook("nothing to push (no CDL)")
-                emit(CameraState(connection: .disconnected))
+        }
+        let sets: [(String, Any)] = [
+            (RedParams.cdlSlopeR, Double(cdl.slope.x)),
+            (RedParams.cdlSlopeG, Double(cdl.slope.y)),
+            (RedParams.cdlSlopeB, Double(cdl.slope.z)),
+            (RedParams.cdlOffsetR, Double(cdl.offset.x)),
+            (RedParams.cdlOffsetG, Double(cdl.offset.y)),
+            (RedParams.cdlOffsetB, Double(cdl.offset.z)),
+            (RedParams.cdlPowerR, Double(cdl.power.x)),
+            (RedParams.cdlPowerG, Double(cdl.power.y)),
+            (RedParams.cdlPowerB, Double(cdl.power.z)),
+            (RedParams.cdlSaturation, Double(cdl.saturation)),
+            (RedParams.cdlEnable, 1),
+        ]
+        for (param, value) in sets {
+            try send(["type": "rcp_set", "id": param, "value": value])
         }
     }
+
+}
+
+// MARK: - NIO handlers
+
+/// Sends the initial HTTP upgrade request when the channel becomes active.
+final class SendUpgradeRequestHandler: ChannelInboundHandler, RemovableChannelHandler, @unchecked Sendable {
+    typealias InboundIn = HTTPClientResponsePart
+    typealias OutboundOut = HTTPClientRequestPart
+
+    private let host: String
+    private let port: Int
+    private let path: String
+
+    init(host: String, port: Int, path: String) {
+        self.host = host
+        self.port = port
+        self.path = path
+    }
+
+    func channelActive(context: ChannelHandlerContext) {
+        var headers = HTTPHeaders()
+        headers.add(name: "Host", value: "\(host):\(port)")
+        headers.add(name: "Content-Length", value: "0")
+        let head = HTTPRequestHead(version: .http1_1, method: .GET, uri: path, headers: headers)
+        context.write(wrapOutboundOut(.head(head)), promise: nil)
+        context.writeAndFlush(wrapOutboundOut(.end(nil)), promise: nil)
+        context.fireChannelActive()
+    }
+}
+
+/// Inbound WS frames -> driver actor.
+final class RedClientWSHandler: ChannelInboundHandler, @unchecked Sendable {
+    typealias InboundIn = WebSocketFrame
+
+    private let driver: RedDriver
+
+    init(driver: RedDriver) {
+        self.driver = driver
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        let frame = unwrapInboundIn(data)
+        switch frame.opcode {
+        case .text:
+            var buf = frame.unmaskedData
+            guard let bytes = buf.readBytes(length: buf.readableBytes) else { return }
+            let payload = Data(bytes)
+            Task { await driver.handleInbound(payload) }
+        case .connectionClose:
+            context.close(promise: nil)
+        default:
+            break
+        }
+    }
+
+    func channelInactive(context: ChannelHandlerContext) {
+        Task { await driver.handleChannelClosed() }
+        context.fireChannelInactive()
     }
 }

Sources/ForgeColor/Lut3D.swift

@@ -42,6 +42,26 @@ public struct Lut3D: Sendable {
         return SIMD3(table[i], table[i + 1], table[i + 2])
     }
 
+    /// True if lattice ≈ identity within tolerance (sampled grid corners).
+    /// Used to detect "nothing left after CDL split" flatten results.
+    public func isApproximatelyIdentity(tolerance: Float = 1.0 / 512) -> Bool {
+        let denom = Float(size - 1)
+        let step = max(1, size / 4)
+        for b in stride(from: 0, to: size, by: step) {
+            for g in stride(from: 0, to: size, by: step) {
+                for r in stride(from: 0, to: size, by: step) {
+                    let idx = ((b * size + g) * size + r) * 3
+                    if abs(table[idx] - Float(r) / denom) > tolerance ||
+                        abs(table[idx + 1] - Float(g) / denom) > tolerance ||
+                        abs(table[idx + 2] - Float(b) / denom) > tolerance {
+                        return false
+                    }
+                }
+            }
+        }
+        return true
+    }
+
     /// Trilinear interpolation; input clamped to [0,1].
     public func sample(_ rgb: SIMD3<Float>) -> SIMD3<Float> {
         let maxIdx = Float(size - 1)

Sources/ForgeSim/RedSimulator.swift

@@ -1,85 +1,150 @@
 import Foundation
 import NIO
+import NIOHTTP1
+import NIOWebSocket
 
-/// RCP2-style TCP simulator: 4-byte big-endian length prefix + JSON message.
+/// RCP2-faithful RED camera simulator.
-/// Message types: handshake, get {param}, set {param,value}, lut {size,table}.
+/// Wire format verified against real Komodo clients (docs/research/red-rcp2-protocol.md):
-/// Replies mirror type with {ok:true} or value payloads.
+/// - WebSocket server at /rcp (real camera: port 9998)
-/// NOTE: real RCP2 framing/params verified against RED SDK in hardware phase;
+/// - Text frames, one JSON object each
-/// this encodes the documented shape (length-prefixed JSON over TCP).
+/// - Handshake: client sends {"type":"rcp_config",...}, sim replies ack
+/// - {"type":"rcp_get","id":P} -> {"id":P,"cur":{"val":...}}
+/// - {"type":"rcp_set","id":P,"value":V} -> stores, replies current value
+/// - pushNotification() sends unsolicited {"id":P,"cur":{"val":...}} to all clients
 public actor RedSimulator {
+
+    /// Param values are ints, doubles, or strings on the wire.
+    public enum ParamValue: Sendable, Equatable {
+        case int(Int)
+        case double(Double)
+        case string(String)
+
+        var jsonValue: Any {
+            switch self {
+            case .int(let i): return i
+            case .double(let d): return d
+            case .string(let s): return s
+            }
+        }
+
+        static func from(_ any: Any) -> ParamValue? {
+            if let i = any as? Int { return .int(i) }
+            if let d = any as? Double { return .double(d) }
+            if let s = any as? String { return .string(s) }
+            return nil
+        }
+    }
+
     private var group: MultiThreadedEventLoopGroup?
     private var channel: Channel?
     public private(set) var boundPort: Int?
 
-    public private(set) var handshakeCount = 0
+    public private(set) var params: [String: ParamValue] = [
-    public private(set) var params: [String: String] = [
+        "RECORD_STATE": .int(0),
-        "RECORD_STATE": "0",
+        "CLIP_NAME": .string(""),
-        "CLIP_NAME": "",
+        "ISO": .int(800),
-        "ISO": "800",
+        "COLOR_TEMPERATURE": .int(5600),
-        "COLOR_TEMP": "5600",
+        "TINT": .int(0),
-        "TINT": "0",
+        "SENSOR_FRAME_RATE": .string("24"),
-        "SENSOR_FPS": "24",
+        "TIMECODE": .string("00:00:00:00"),
-        "TIMECODE": "00:00:00:00",
+        "CDL_ENABLE": .int(0),
-        "MODEL": "V-RAPTOR-SIM",
+        "CAMERA_TYPE": .string("V-RAPTOR-SIM"),
     ]
-    public struct UploadedLut: Sendable {
+    /// Raw rcp_config messages received (handshake verification; decode in consumers).
-        public let size: Int
+    public private(set) var receivedConfigsData: [Data] = []
-        public let entryCount: Int
+
-    }
+    private var clients: [ObjectIdentifier: Channel] = [:]
-    public private(set) var uploadedLuts: [UploadedLut] = []
 
     public init() {}
 
-    public func setParam(_ key: String, value: String) {
+    public func setParam(_ key: String, value: ParamValue) {
         params[key] = value
     }
 
-    /// Data-in/Data-out across the actor boundary (JSON dicts aren't Sendable).
+    /// Send unsolicited notification to all connected clients (camera-initiated update).
-    func handleFrame(_ payload: Data) -> Data {
+    public func pushNotification(param: String, value: ParamValue) {
-        guard let obj = try? JSONSerialization.jsonObject(with: payload) as? [String: Any] else {
+        params[param] = value
-            return (try? JSONSerialization.data(withJSONObject: ["ok": false, "error": "bad JSON"])) ?? Data()
+        let payload: [String: Any] = ["id": param, "type": "rcp_cur_int", "cur": ["val": value.jsonValue]]
+        guard let data = try? JSONSerialization.data(withJSONObject: payload, options: [.sortedKeys]) else { return }
+        for ch in clients.values {
+            Self.sendText(data, over: ch)
         }
-        let reply = handle(message: obj)
-        return (try? JSONSerialization.data(withJSONObject: reply, options: [.sortedKeys])) ?? Data()
     }
 
-    func handle(message: [String: Any]) -> [String: Any] {
+    // MARK: Frame handling
-        switch message["type"] as? String {
+
-        case "handshake":
+    func clientConnected(_ ch: Channel) {
-            handshakeCount += 1
+        clients[ObjectIdentifier(ch)] = ch
-            return ["type": "handshake", "ok": true, "model": params["MODEL"] ?? ""]
-        case "get":
-            guard let param = message["param"] as? String, let value = params[param] else {
-                return ["type": "get", "ok": false, "error": "unknown param"]
     }
-            return ["type": "get", "ok": true, "param": param, "value": value]
+
-        case "set":
+    func clientDisconnected(_ ch: Channel) {
-            guard let param = message["param"] as? String, let value = message["value"] as? String else {
+        clients.removeValue(forKey: ObjectIdentifier(ch))
-                return ["type": "set", "ok": false, "error": "bad set"]
     }
-            params[param] = value
+
-            return ["type": "set", "ok": true, "param": param]
+    func handleText(_ data: Data, from ch: Channel) {
-        case "lut":
+        guard let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
-            guard let size = message["size"] as? Int,
+              let type = obj["type"] as? String else { return }
-                  let table = message["table"] as? [Any],
+
-                  table.count == size * size * size * 3 else {
+        switch type {
-                return ["type": "lut", "ok": false, "error": "bad lut"]
+        case "rcp_config":
+            receivedConfigsData.append(data)
+            reply(["type": "rcp_config_ack", "ok": true], to: ch)
+        case "rcp_get":
+            guard let id = obj["id"] as? String else { return }
+            if let value = params[id] {
+                reply(["id": id, "type": "rcp_cur_int", "cur": ["val": value.jsonValue]], to: ch)
+            } else {
+                reply(["id": id, "type": "rcp_error", "error": "unknown param"], to: ch)
             }
-            uploadedLuts.append(UploadedLut(size: size, entryCount: table.count))
+        case "rcp_set":
-            return ["type": "lut", "ok": true]
+            guard let id = obj["id"] as? String,
+                  let raw = obj["value"], let value = ParamValue.from(raw) else { return }
+            params[id] = value
+            reply(["id": id, "type": "rcp_cur_int", "cur": ["val": value.jsonValue]], to: ch)
         default:
-            return ["ok": false, "error": "unknown type"]
+            reply(["type": "rcp_error", "error": "unsupported type \(type)"], to: ch)
         }
     }
 
+    private func reply(_ obj: [String: Any], to ch: Channel) {
+        guard let data = try? JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys]) else { return }
+        Self.sendText(data, over: ch)
+    }
+
+    private static func sendText(_ data: Data, over ch: Channel) {
+        var buf = ch.allocator.buffer(capacity: data.count)
+        buf.writeBytes(data)
+        let frame = WebSocketFrame(fin: true, opcode: .text, data: buf)
+        ch.writeAndFlush(NIOAny(frame), promise: nil)
+    }
+
+    // MARK: Lifecycle
+
     public func start(port: Int) async throws {
         let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
         self.group = group
         let sim = self
+
+        let upgrader = NIOWebSocketServerUpgrader(
+            shouldUpgrade: { channel, head in
+                // Real camera serves /rcp.
+                guard head.uri == "/rcp" else {
+                    return channel.eventLoop.makeSucceededFuture(nil)
+                }
+                return channel.eventLoop.makeSucceededFuture(HTTPHeaders())
+            },
+            upgradePipelineHandler: { channel, _ in
+                Task { await sim.clientConnected(channel) }
+                return channel.pipeline.addHandler(RedWSHandler(sim: sim))
+            })
+
         let bootstrap = ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
             .childChannelInitializer { channel in
-                channel.pipeline.addHandler(RCPFrameHandler(sim: sim))
+                let config = NIOHTTPServerUpgradeConfiguration(
+                    upgraders: [upgrader],
+                    completionHandler: { _ in })
+                return channel.pipeline.configureHTTPServerPipeline(withServerUpgrade: config)
             }
         let channel = try await bootstrap.bind(host: "127.0.0.1", port: port).get()
         self.channel = channel
@@ -87,6 +152,10 @@ public actor RedSimulator {
     }
 
     public func stop() async throws {
+        for ch in clients.values {
+            _ = try? await ch.close()
+        }
+        clients.removeAll()
         try await channel?.close()
         try await group?.shutdownGracefully()
         channel = nil
@@ -95,41 +164,43 @@ public actor RedSimulator {
     }
 }
 
-/// Length-prefixed JSON framing: u32 BE length + payload.
+/// Inbound WebSocket frames -> sim actor.
-final class RCPFrameHandler: ChannelInboundHandler, @unchecked Sendable {
+final class RedWSHandler: ChannelInboundHandler, @unchecked Sendable {
-    typealias InboundIn = ByteBuffer
+    typealias InboundIn = WebSocketFrame
-    typealias OutboundOut = ByteBuffer
+    typealias OutboundOut = WebSocketFrame
 
     private let sim: RedSimulator
-    private var buffer = ByteBuffer()
 
     init(sim: RedSimulator) {
         self.sim = sim
     }
 
     func channelRead(context: ChannelHandlerContext, data: NIOAny) {
-        var incoming = unwrapInboundIn(data)
+        let frame = unwrapInboundIn(data)
-        buffer.writeBuffer(&incoming)
+        switch frame.opcode {
-
+        case .text:
-        while true {
+            var buf = frame.unmaskedData
-            guard buffer.readableBytes >= 4,
+            guard let bytes = buf.readBytes(length: buf.readableBytes) else { return }
-                  let length = buffer.getInteger(at: buffer.readerIndex, as: UInt32.self),
-                  buffer.readableBytes >= 4 + Int(length) else { break }
-            buffer.moveReaderIndex(forwardBy: 4)
-            guard let bytes = buffer.readBytes(length: Int(length)) else { break }
             let payload = Data(bytes)
+            let ch = context.channel
+            Task { await sim.handleText(payload, from: ch) }
+        case .connectionClose:
+            let ch = context.channel
+            Task { await sim.clientDisconnected(ch) }
+            context.close(promise: nil)
+        case .ping:
+            var buf = frame.unmaskedData
+            let pong = WebSocketFrame(fin: true, opcode: .pong, data: buf)
+            context.writeAndFlush(wrapOutboundOut(pong), promise: nil)
+            _ = buf
+        default:
+            break
+        }
+    }
 
-            let channel = context.channel
+    func channelInactive(context: ChannelHandlerContext) {
-            let loop = context.eventLoop
+        let ch = context.channel
-            Task {
+        Task { await sim.clientDisconnected(ch) }
-                let replyData = await self.sim.handleFrame(payload)
+        context.fireChannelInactive()
-                loop.execute {
-                    var out = channel.allocator.buffer(capacity: replyData.count + 4)
-                    out.writeInteger(UInt32(replyData.count))
-                    out.writeBytes(replyData)
-                    channel.writeAndFlush(out, promise: nil)
-                }
-            }
-        }
     }
 }

Tests/ForgeCameraREDTests/RedDriverTests.swift

@@ -6,6 +6,10 @@ import ForgeColor
 import ForgeSim
 @testable import ForgeCameraRED
 
+/// Tests against verified RCP2 wire protocol:
+/// WebSocket ws://<ip>:9998/rcp, rcp_config handshake, rcp_get/rcp_set,
+/// responses {"id":..., "cur":{"val":...}}, per-channel CDL params,
+/// no live LUT upload (docs/research/red-rcp2-protocol.md).
 final class RedDriverTests: XCTestCase {
 
     var sim: RedSimulator!
@@ -25,16 +29,16 @@ final class RedDriverTests: XCTestCase {
         return RedDriver(host: "127.0.0.1", port: port, pollInterval: pollInterval)
     }
 
-    // Capabilities: RED = native CDL (IPP2), 17/33 LUTs.
+    // RED: native CDL yes, live LUT upload NO (RCP2 has no LUT-push op).
     func testCapabilities() async {
         let driver = await makeDriver()
         XCTAssertEqual(driver.capabilities.vendor, .red)
         XCTAssertTrue(driver.capabilities.supportsNativeCDL)
-        XCTAssertTrue(driver.capabilities.lut3dSizes.contains(33))
+        XCTAssertTrue(driver.capabilities.lut3dSizes.isEmpty)
     }
 
-    // Connect performs RCP-style handshake, emits connected.
+    // Connect: WS upgrade on /rcp + rcp_config handshake with client name.
-    func testConnectHandshake() async throws {
+    func testConnectSendsRcpConfig() async throws {
         let driver = await makeDriver()
         var iterator = driver.state.makeAsyncIterator()
         try await driver.connect()
@@ -46,8 +50,22 @@ final class RedDriverTests: XCTestCase {
             }
         }
         XCTAssertTrue(connected)
-        let handshakes = await sim.handshakeCount
+        // Config delivery is async post-upgrade; allow brief settle.
-        XCTAssertEqual(handshakes, 1)
+        var configsData: [Data] = []
+        for _ in 0..<20 {
+            configsData = await sim.receivedConfigsData
+            if !configsData.isEmpty { break }
+            try await Task.sleep(for: .milliseconds(25))
+        }
+        XCTAssertEqual(configsData.count, 1)
+        guard let first = configsData.first else {
+            await driver.disconnect()
+            return
+        }
+        let config = try JSONSerialization.jsonObject(with: first) as? [String: Any]
+        XCTAssertEqual(config?["type"] as? String, "rcp_config")
+        let client = config?["client"] as? [String: Any]
+        XCTAssertEqual(client?["name"] as? String, "Forge")
         await driver.disconnect()
     }
 
@@ -59,25 +77,25 @@ final class RedDriverTests: XCTestCase {
         } catch {}
     }
 
-    // Rec state + clip name via param polling.
+    // Rec state: RECORD_STATE cur.val 1 -> recording, clip name via CLIP_NAME.
     func testRecStateEvents() async throws {
         let driver = await makeDriver()
         var iterator = driver.state.makeAsyncIterator()
         try await driver.connect()
 
-        await sim.setParam("RECORD_STATE", value: "1")
+        await sim.setParam("RECORD_STATE", value: .int(1))
-        await sim.setParam("CLIP_NAME", value: "B001_C001_0710AB")
+        await sim.setParam("CLIP_NAME", value: .string("B001_C001_0710AB"))
+        // Params arrive one frame at a time — wait for a state with BOTH rec + clip.
         var sawRec = false
-        for _ in 0..<30 {
+        for _ in 0..<60 {
-            if let s = await iterator.next(), s.isRecording {
+            if let s = await iterator.next(), s.isRecording, s.clipName == "B001_C001_0710AB" {
-                XCTAssertEqual(s.clipName, "B001_C001_0710AB")
                 sawRec = true
                 break
             }
         }
         XCTAssertTrue(sawRec)
 
-        await sim.setParam("RECORD_STATE", value: "0")
+        await sim.setParam("RECORD_STATE", value: .int(0))
         var sawStop = false
         for _ in 0..<30 {
             if let s = await iterator.next(), !s.isRecording {
@@ -89,24 +107,25 @@ final class RedDriverTests: XCTestCase {
         await driver.disconnect()
     }
 
-    // Metadata params surface.
+    // Metadata via real param names: ISO, COLOR_TEMPERATURE, TINT, SENSOR_FRAME_RATE, TIMECODE.
     func testMetadata() async throws {
-        await sim.setParam("ISO", value: "1600")
+        await sim.setParam("ISO", value: .int(1600))
-        await sim.setParam("COLOR_TEMP", value: "4500")
+        await sim.setParam("COLOR_TEMPERATURE", value: .int(4500))
-        await sim.setParam("TINT", value: "3")
+        await sim.setParam("TINT", value: .int(3))
-        await sim.setParam("SENSOR_FPS", value: "23.98")
+        await sim.setParam("SENSOR_FRAME_RATE", value: .string("23.98"))
-        await sim.setParam("TIMECODE", value: "15:30:00:00")
+        await sim.setParam("TIMECODE", value: .string("15:30:00:00"))
 
         let driver = await makeDriver()
         var iterator = driver.state.makeAsyncIterator()
         try await driver.connect()
 
+        // Wait for a state where ALL polled params have landed.
         var got = false
-        for _ in 0..<30 {
+        for _ in 0..<60 {
-            if let s = await iterator.next(), let md = s.metadata, md.exposureIndex == 1600 {
+            if let s = await iterator.next(), let md = s.metadata,
-                XCTAssertEqual(md.whiteBalance, 4500)
+               md.exposureIndex == 1600, md.whiteBalance == 4500,
-                XCTAssertEqual(md.tint, 3)
+               md.tint == 3, md.timecode == "15:30:00:00",
-                XCTAssertEqual(md.timecode, "15:30:00:00")
+               let fps = md.fps, abs(fps - 23.98) < 0.001 {
                 got = true
                 break
             }
@@ -115,37 +134,99 @@ final class RedDriverTests: XCTestCase {
         await driver.disconnect()
     }
 
-    // Push: CDL to IPP2 slots + LUT upload.
+    // CDL push: 10 rcp_set ops (9 SOP channels + saturation) + CDL_ENABLE.
-    func testPushLookWithNativeCDL() async throws {
+    func testPushCDLSetsPerChannelParams() async throws {
         let driver = await makeDriver()
         try await driver.connect()
 
-        let cdl = CDL(slope: SIMD3(1.1, 1.0, 0.9), offset: SIMD3(0.02, 0, 0), power: .one, saturation: 1.3)
+        let cdl = CDL(
-        let look = FlattenedLook(cdl: cdl, lut: Lut3D.identity(size: 33), latticeSize: 33)
+            slope: SIMD3(1.1, 1.0, 0.9),
-        try await driver.push(look: look)
+            offset: SIMD3(0.02, 0.0, -0.01),
+            power: SIMD3(0.95, 1.0, 1.05),
+            saturation: 1.3)
+        try await driver.push(look: FlattenedLook(cdl: cdl, lut: nil, latticeSize: 33))
 
-        let cdlValue = await sim.params["IPP2_CDL"]
+        // rcp_set frames are fire-and-forget; wait for sim to process all 11.
-        XCTAssertNotNil(cdlValue)
+        var params: [String: RedSimulator.ParamValue] = [:]
-        let obj = try JSONSerialization.jsonObject(with: Data(cdlValue!.utf8)) as? [String: Any]
+        for _ in 0..<40 {
-        let slope = obj?["slope"] as? [Double]
+            params = await sim.params
-        XCTAssertEqual(slope?[0] ?? 0, 1.1, accuracy: 1e-5)
+            if params["CDL_SATURATION"] != nil, case .int(1)? = params["CDL_ENABLE"] { break }
-
+            try await Task.sleep(for: .milliseconds(25))
-        let luts = await sim.uploadedLuts
+        }
-        XCTAssertEqual(luts.count, 1)
+        func f(_ key: String) -> Float? {
-        XCTAssertEqual(luts[0].size, 33)
+            if case .double(let d)? = params[key] { return Float(d) }
+            if case .int(let i)? = params[key] { return Float(i) }
+            return nil
+        }
+        XCTAssertEqual(f("CDL_SLOPE_RED") ?? 0, 1.1, accuracy: 1e-4)
+        XCTAssertEqual(f("CDL_SLOPE_GREEN") ?? 0, 1.0, accuracy: 1e-4)
+        XCTAssertEqual(f("CDL_SLOPE_BLUE") ?? 0, 0.9, accuracy: 1e-4)
+        XCTAssertEqual(f("CDL_OFFSET_RED") ?? 0, 0.02, accuracy: 1e-4)
+        XCTAssertEqual(f("CDL_OFFSET_BLUE") ?? 0, -0.01, accuracy: 1e-4)
+        XCTAssertEqual(f("CDL_POWER_RED") ?? 0, 0.95, accuracy: 1e-4)
+        XCTAssertEqual(f("CDL_POWER_BLUE") ?? 0, 1.05, accuracy: 1e-4)
+        XCTAssertEqual(f("CDL_SATURATION") ?? 0, 1.3, accuracy: 1e-4)
+        if case .int(let enabled)? = params["CDL_ENABLE"] {
+            XCTAssertEqual(enabled, 1)
+        } else {
+            XCTFail("CDL_ENABLE not set")
+        }
         await driver.disconnect()
     }
 
-    // Unsupported LUT size rejected.
+    // LUT in look -> rejected: RCP2 cannot upload LUTs live.
-    func testUnsupportedLutRejected() async throws {
+    func testPushWithLutRejected() async throws {
         let driver = await makeDriver()
         try await driver.connect()
+        let look = FlattenedLook(cdl: .identity, lut: Lut3D.build(size: 17) { $0 * 0.9 }, latticeSize: 17)
         do {
-            try await driver.push(look: FlattenedLook(cdl: nil, lut: Lut3D.identity(size: 65), latticeSize: 65))
+            try await driver.push(look: look)
             XCTFail("expected unsupportedLook")
         } catch let e as CameraError {
             guard case .unsupportedLook = e else { return XCTFail("wrong error") }
         }
         await driver.disconnect()
     }
+
+    // Identity LUT in look tolerated (flattener artifact), CDL still pushed.
+    func testPushWithIdentityLutAccepted() async throws {
+        let driver = await makeDriver()
+        try await driver.connect()
+        let cdl = CDL(slope: SIMD3(1.2, 1, 1), offset: .zero, power: .one, saturation: 1)
+        let look = FlattenedLook(cdl: cdl, lut: Lut3D.identity(size: 17), latticeSize: 17)
+        try await driver.push(look: look)
+        var pushed = false
+        for _ in 0..<40 {
+            let params = await sim.params
+            if case .double(let d)? = params["CDL_SLOPE_RED"], abs(Float(d) - 1.2) < 1e-4 {
+                pushed = true
+                break
+            }
+            try await Task.sleep(for: .milliseconds(25))
+        }
+        XCTAssertTrue(pushed, "CDL not pushed")
+        await driver.disconnect()
+    }
+
+    // Unsolicited camera notification (rec start w/o poll) surfaces in state stream.
+    func testUnsolicitedNotification() async throws {
+        let driver = await makeDriver(pollInterval: .seconds(10)) // poll effectively off
+        var iterator = driver.state.makeAsyncIterator()
+        try await driver.connect()
+        // Drain the connected event.
+        _ = await iterator.next()
+
+        await sim.pushNotification(param: "RECORD_STATE", value: .int(1))
+        await sim.pushNotification(param: "CLIP_NAME", value: .string("B001_C009"))
+
+        var sawRec = false
+        for _ in 0..<10 {
+            if let s = await iterator.next(), s.isRecording {
+                sawRec = true
+                break
+            }
+        }
+        XCTAssertTrue(sawRec)
+        await driver.disconnect()
+    }
 }

Tests/ForgeCameraTests/GangTests.swift

@@ -109,6 +109,42 @@ final class GangTests: XCTestCase {
         } catch {}
     }
 
+    // CDL-only camera (RED/RCP2, lut3dSizes empty): CDL-only grade pushes as bare CDL.
+    func testCDLOnlyCameraPushesCDLOnlyGrade() async throws {
+        let red = RecordingDriver(nativeCDL: true, lutSizes: [])
+        let gang = GradeGang(latticeSize: 33)
+        // Grade = single CDL node — representable on a CDL-only camera.
+        let grade = GradeStack(nodes: [
+            GradeNode(kind: .cdl(CDL(slope: SIMD3(1.3, 1, 1), offset: .zero, power: .one, saturation: 1.1))),
+        ])
+        await gang.setGrade(grade, for: "R")
+        await gang.register(slot: "R", driver: red)
+        try await gang.pushGrade(from: "R")
+
+        let pushes = await red.pushedLooks
+        XCTAssertEqual(pushes.count, 1)
+        XCTAssertNotNil(pushes[0].cdl)
+        XCTAssertNil(pushes[0].lut)
+        XCTAssertEqual(pushes[0].cdl?.slope.x ?? 0, 1.3, accuracy: 1e-5)
+    }
+
+    // CDL-only camera + grade with non-CDL nodes: hard error (camera can't show it).
+    func testCDLOnlyCameraRejectsComplexGrade() async throws {
+        let red = RecordingDriver(nativeCDL: true, lutSizes: [])
+        let gang = GradeGang(latticeSize: 33)
+        await gang.setGrade(makeGrade(), for: "R") // CDL + saturation node
+        await gang.register(slot: "R", driver: red)
+        do {
+            try await gang.pushGrade(from: "R")
+            XCTFail("expected partial failure")
+        } catch let e as GangError {
+            guard case .partialFailure(let failures) = e else { return XCTFail("wrong error") }
+            XCTAssertEqual(failures.count, 1)
+        }
+        let pushes = await red.pushedLooks
+        XCTAssertTrue(pushes.isEmpty)
+    }
+
     // Push failure on one camera doesn't block others (best-effort multi-push).
     func testPushPartialFailure() async throws {
         let good = RecordingDriver(nativeCDL: true)