import AVFoundation import CoreMedia import Foundation import ScreenCaptureKit // --------------------------------------------------------------------------- // MARK: - Output // --------------------------------------------------------------------------- private let outputDir: String = { let dir = NSHomeDirectory() + "/.skykoi/capture" try? FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) return dir }() private let outputPath = outputDir + "/audio-live.jsonl" private var outputFd: Int32 = open(outputPath, O_WRONLY | O_APPEND | O_CREAT, 0o644) func emit(_ dict: [String: Any]) { guard let data = try? JSONSerialization.data(withJSONObject: dict, options: []), let line = String(data: data, encoding: .utf8) else { return } let full = line + "\n" if outputFd >= 0, let bytes = full.data(using: .utf8) { bytes.withUnsafeBytes { ptr in _ = write(outputFd, ptr.baseAddress!, ptr.count) } } } // Live interim transcription for the mic. The TUI's visible-hearing ticker // reads mic-partial.json to stream the wearer's words as they speak (the // color-coded 👂 line). The Windows listener writes this same file; emitting // it here gives macOS identical live-transcription behavior (previously the // daemon dropped every interim hypothesis, so the Mac ticker only ever showed // finalized utterances — never the live stream). private let micPartialPath = outputDir + "/mic-partial.json" private let partialLock = NSLock() private var lastPartialAt: Double = 0 func emitMicPartial(_ text: String) { let t = text.trimmingCharacters(in: .whitespacesAndNewlines) guard !t.isEmpty else { return } let now = Date().timeIntervalSince1970 partialLock.lock() if now - lastPartialAt < 0.1 { partialLock.unlock(); return } // ~100ms throttle lastPartialAt = now partialLock.unlock() let dict: [String: Any] = ["ts": tsNow(), "text": String(t.prefix(300))] guard let data = try? JSONSerialization.data(withJSONObject: dict, options: []) else { return } try? data.write(to: URL(fileURLWithPath: micPartialPath), options: .atomic) } func tsNow() -> Int64 { Int64(Date().timeIntervalSince1970 * 1000) } // Process-level record of the last message received from the live-STT // websocket. Audio always flows to Soniox (silence included), so a healthy // link always chatters. The per-streamer watchdog dies with its streamer // (2026-07-07: a reconnect storm after "Soniox 400: Audio is too long" // wedged the streamer and the koi stayed deaf for 3 hours while mic levels // kept flowing) — the deadman in Main reads this and exits for a clean // launchd restart instead. let sttMsgLock = NSLock() var lastSttMsgAt: Double = Date().timeIntervalSince1970 func noteSttMessage() { sttMsgLock.lock(); lastSttMsgAt = Date().timeIntervalSince1970; sttMsgLock.unlock() } // --------------------------------------------------------------------------- // MARK: - Capture opt-ins (shared with the TS runtime's settings file) // --------------------------------------------------------------------------- // System-audio capture uses ScreenCaptureKit (SCStream), which macOS 26 nags // about on EVERY launch ("bypass the system private window picker and directly // access your screen and audio") — there is no way to make the OS remember an // SCStream grant permanently. So it is OPT-IN, default OFF: the mic (durable // TCC grant via the stable code-signing identity) covers normal use with no // recurring prompt. Enable with SKYKOI_CAPTURE_SYSTEM_AUDIO=1 or // {"systemAudio": true} in ~/.skykoi/capture/settings.json. func systemAudioEnabled() -> Bool { if ProcessInfo.processInfo.environment["SKYKOI_CAPTURE_SYSTEM_AUDIO"] == "1" { return true } let path = NSHomeDirectory() + "/.skykoi/capture/settings.json" guard let data = try? Data(contentsOf: URL(fileURLWithPath: path)), let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any] else { return false } return (obj["systemAudio"] as? Bool) ?? false } // --------------------------------------------------------------------------- // MARK: - API key // --------------------------------------------------------------------------- struct WhisperConfig { let apiKey: String let url: String let model: String } func loadWhisperConfig() -> WhisperConfig? { // Check CLI args: --api-key [--url ] [--model ] let args = CommandLine.arguments var cliKey: String? var cliUrl: String? var cliModel: String? for i in 0.. [String: String] { guard let content = try? String(contentsOfFile: path, encoding: .utf8) else { return [:] } var result: [String: String] = [:] for line in content.split(separator: "\n") { let t = line.trimmingCharacters(in: .whitespaces) if t.hasPrefix("#") { continue } if let eq = t.firstIndex(of: "=") { let key = String(t[t.startIndex.. String? { let path = NSHomeDirectory() + "/.skykoi/credentials/provisioned-keys.json" guard let data = try? Data(contentsOf: URL(fileURLWithPath: path)), let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any], let keys = obj["keys"] as? [String: Any], let soniox = keys["soniox"] as? [String: Any], let key = soniox["apiKey"] as? String, !key.isEmpty else { return nil } return key } /// Resolve the CURRENT Soniox key, freshest-source-first, re-read every call: /// a hand-set env/`.env` key wins (never rotates), otherwise the gateway's /// live provisioned key. Called on each WebSocket connect so key rotation is /// picked up automatically. func resolveCurrentSonioxKey() -> String? { let env = ProcessInfo.processInfo.environment if let k = env["SONIOX_API_KEY"], !k.isEmpty { return k } let dotenvPath = NSHomeDirectory() + "/.skykoi/.env" if let content = try? String(contentsOfFile: dotenvPath, encoding: .utf8) { for line in content.split(separator: "\n") { let t = line.trimmingCharacters(in: .whitespaces) if t.hasPrefix("SONIOX_API_KEY=") { let v = String(t.dropFirst("SONIOX_API_KEY=".count)).trimmingCharacters(in: .init(charactersIn: "\"'")) if !v.isEmpty { return v } } } } return provisionedSonioxKey() } func loadSonioxConfig() -> SonioxConfig? { let args = CommandLine.arguments var cliKey: String? for i in 0.. [String: String] { guard let content = try? String(contentsOfFile: path, encoding: .utf8) else { return [:] } var result: [String: String] = [:] for line in content.split(separator: "\n") { let t = line.trimmingCharacters(in: .whitespaces) if t.hasPrefix("#") { continue } if let eq = t.firstIndex(of: "=") { let key = String(t[t.startIndex.." endpoint marker per utterance. We accumulate final tokens and /// emit one transcript per utterance, tagged with the dominant speaker label. final class SonioxStreamer: AudioSink { private let cfg: SonioxConfig private let source: String private var session = URLSession(configuration: .default) private var task: URLSessionWebSocketTask? private let lock = NSLock() private var ready = false private var closed = false private var converter: AVAudioConverter? private let targetFormat = AVAudioFormat(commonFormat: .pcmFormatInt16, sampleRate: 16000, channels: 1, interleaved: true)! private var segText = "" private var segCounts: [String: Int] = [:] private var backoff: Double = 0.8 private var lastErrLog: Double = 0 private var lastMsgTs: Double = Date().timeIntervalSince1970 // last message FROM Soniox (watchdog) private var healthTimer: DispatchSourceTimer? // Per-utterance audio for downstream voice-embedding diarization. Soniox's own // speaker labels collapse to one speaker on a single far-field mic, so we persist // each utterance's PCM and let the bridge re-diarize by voice fingerprint. private var uttPCM = Data() private let pcmLock = NSLock() private var uttCounter = 0 private static let uttDir: String = { let d = (NSHomeDirectory() as NSString).appendingPathComponent(".skykoi/capture/utt") try? FileManager.default.removeItem(atPath: d) // clear stale clips from a prior run try? FileManager.default.createDirectory(atPath: d, withIntermediateDirectories: true) return d }() init(cfg: SonioxConfig, source: String) { self.cfg = cfg self.source = source connect() startHealthTimer() } // Watchdog: Soniox sends periodic messages whenever audio is flowing (even // silence). If it goes silent too long, the WebSocket is wedged (e.g. a dead // URLSession after sleep/wake or a network change) and the receive loop may // never fire .failure — so we self-heal by forcing a fresh reconnect. private func startHealthTimer() { let t = DispatchSource.makeTimerSource(queue: .global()) t.schedule(deadline: .now() + 15, repeating: 15) t.setEventHandler { [weak self] in self?.healthCheck() } t.resume() healthTimer = t } private var silentReconnects = 0 private func healthCheck() { if closed { return } let now = Date().timeIntervalSince1970 guard now - lastMsgTs > 35 else { return } silentReconnects += 1 // In-process WebSocket reconnects CANNOT recover a process-level network wedge // (a URLSession stuck after sleep/wake or a network change) — fresh sessions in // the same process stay silent. After a few silent reconnects on the mic, exit so // launchd (KeepAlive) restarts us with fresh networking, which DOES recover. A real // received message resets silentReconnects to 0, so healthy streams never trip this. if source == "mic" && silentReconnects >= 3 { emit(["type": "error", "ts": tsNow(), "message": "Soniox wedged: \(silentReconnects) silent reconnects — exiting for a clean launchd restart"]) exit(0) } if now - lastErrLog > 30 { emit(["type": "error", "ts": tsNow(), "message": "Soniox watchdog: silent \(Int(now - lastMsgTs))s, reconnecting (\(silentReconnects))"]) lastErrLog = now } lock.lock(); ready = false; lock.unlock() task?.cancel(with: .goingAway, reason: nil) backoff = 0.8 lastMsgTs = now connect() } private let connectLock = NSLock() private func connect() { if closed { return } // Serialize the session swap + task creation. connect() fires from init, the // reconnect backoff, AND the watchdog; concurrent calls raced on `session` and // crashed with 'Task created in a session that has been invalidated'. connectLock.lock() reconnectPending = false // this is the connect that pending flight promised segText = ""; segCounts = [:] session.invalidateAndCancel() // a wedged URLSession can't be revived session = URLSession(configuration: .default) // always reconnect on a FRESH session // NOTE: do NOT touch lastMsgTs here — it must only advance when a message // is actually RECEIVED. Resetting it on connect would let a failing // reconnect loop keep the watchdog from ever tripping (the original bug). let t = session.webSocketTask(with: URLRequest(url: URL(string: cfg.url)!)) task = t connectLock.unlock() t.resume() // Re-resolve the key on EVERY connect — Soniox keys rotate hourly, so a // key cached at startup would strand the daemon deaf after ~1h. Falls // back to the startup key if the file read momentarily fails. let liveKey = resolveCurrentSonioxKey() ?? cfg.apiKey var config: [String: Any] = [ "api_key": liveKey, "model": cfg.model, "audio_format": "pcm_s16le", "sample_rate": 16000, "num_channels": 1, "enable_speaker_diarization": true, "enable_endpoint_detection": true, // Pin recognition to English (parity with the Windows listener) so // English speech isn't misdetected as another language. "language_hints": (ProcessInfo.processInfo.environment["SONIOX_LANGUAGE_HINTS"] ?? "en") .split(separator: ",").map { $0.trimmingCharacters(in: .whitespaces) }.filter { !$0.isEmpty }, ] if !cfg.terms.isEmpty { config["context"] = ["terms": cfg.terms] } // bias toward "koi" if let d = try? JSONSerialization.data(withJSONObject: config), let s = String(data: d, encoding: .utf8) { t.send(.string(s)) { _ in } } lock.lock(); ready = true; lock.unlock() receiveLoop() } private var reconnectPending = false private func reconnect() { lock.lock(); ready = false; lock.unlock() if closed { return } // Single-flight: a server error cancels the task AND fails its pending // receive — both paths land here. Letting each schedule its own connect // makes every fresh session get invalidated by the other's connect in a // self-sustaining storm (observed 2026-07-07 after a Soniox 400 // "Audio is too long": the storm wedged the streamer, its watchdog died // with it, and the koi stayed deaf for 3 hours). connectLock.lock() if reconnectPending { connectLock.unlock(); return } reconnectPending = true connectLock.unlock() let delay = backoff backoff = min(backoff * 2, 15.0) DispatchQueue.global().asyncAfter(deadline: .now() + delay) { [weak self] in self?.connect() } } private func receiveLoop() { receiveLoop(on: task) } private func receiveLoop(on t: URLSessionWebSocketTask?) { t?.receive { [weak self] result in guard let self = self, !self.closed else { return } // A superseded task (its session was invalidated by a newer connect) // must not steer reconnects or double-receive on the live task. self.connectLock.lock(); let isCurrent = t === self.task; self.connectLock.unlock() guard isCurrent else { return } switch result { case .failure: self.reconnect() case .success(let message): self.lastMsgTs = Date().timeIntervalSince1970 // proof the link is alive noteSttMessage() // feeds the process-level deadman self.silentReconnects = 0 // a real message → connection healthy switch message { case .string(let text): self.handle(text) case .data(let data): if let text = String(data: data, encoding: .utf8) { self.handle(text) } @unknown default: break } self.receiveLoop(on: t) } } } private func handle(_ text: String) { guard let data = text.data(using: .utf8), let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any] else { return } if let code = obj["error_code"] as? Int { let now = Date().timeIntervalSince1970 if now - lastErrLog > 5 { // throttle: an outage shouldn't flood the jsonl emit(["type": "error", "ts": tsNow(), "message": "Soniox \(code): \((obj["error_message"] as? String) ?? "")"]) lastErrLog = now } // 429 = per-MINUTE org quota. Retrying every ≤15s keeps the org // pinned at the limit forever (observed 2026-07-05: 400+ 429s over // 75 minutes, koi deaf the whole time, every device sharing the key // starved). Park the backoff at 60-90s so the quota can actually // recover; other server errors keep the normal fast backoff. if code == 429 { lock.lock(); backoff = 60.0 + Double.random(in: 0...30); lock.unlock() } // ANY server error (429 quota, 408 timeout, 5xx) means the // stream is unusable — close and reconnect with backoff. task?.cancel(); reconnect() return } guard let tokens = obj["tokens"] as? [[String: Any]] else { return } lock.lock(); backoff = 0.8; lock.unlock() // Accumulate this batch's interim (uncommitted) hypothesis so the mic's // live partial can be streamed to the hearing ticker before the tokens // finalize — matching the Windows listener. var interimTail = "" for tok in tokens { let isFinal = (tok["is_final"] as? Bool) ?? false let txt = (tok["text"] as? String) ?? "" if !isFinal { // interim hypothesis if txt != "" && txt != "" { interimTail += txt } continue } if txt == "" || txt == "" { flushSegment(); continue } segText += txt var sp = "" if let s = tok["speaker"] as? String { sp = s } else if let n = tok["speaker"] as? Int { sp = String(n) } if !sp.isEmpty { segCounts[sp, default: 0] += max(1, txt.count) } } // Stream the live interim for the wearer's mic (committed segment so far // + the current interim tail), throttled, so the visible-hearing ticker // shows words as they are spoken on macOS just like on Windows. if source == "mic" { emitMicPartial(segText + interimTail) } } private func flushSegment() { let trimmed = segText.trimmingCharacters(in: .whitespacesAndNewlines) let speaker = segCounts.max(by: { $0.value < $1.value })?.key segText = ""; segCounts = [:] guard !trimmed.isEmpty, !isBlankAudio(trimmed) else { return } let eventType = source == "mic" ? "mic_transcript" : "audio_transcript" var ev: [String: Any] = ["type": eventType, "ts": tsNow(), "text": trimmed, "source": source, "isFinal": true] if let sp = speaker { ev["speaker"] = sp } if source == "mic" { // persist this utterance's audio for voice-embedding diarization // removeAll(keepingCapacity: FALSE) hands uttPCM a brand-new buffer, so `snapshot` // becomes the SOLE owner of the old one. Writing it on THIS thread (not a background // queue) means nothing else ever touches that buffer concurrently — the previous // async write raced append()'s in-place mutation and crashed in Data's COW machinery. pcmLock.lock(); let snapshot = uttPCM; uttPCM.removeAll(keepingCapacity: false); pcmLock.unlock() if snapshot.count >= 16000 { // >= ~0.5s of 16k mono int16 — enough to fingerprint uttCounter += 1 let name = "\(Int(Date().timeIntervalSince1970 * 1000))-\(uttCounter).pcm" let path = (Self.uttDir as NSString).appendingPathComponent(name) if (try? snapshot.write(to: URL(fileURLWithPath: path))) != nil { ev["audioPath"] = path } if uttCounter % 20 == 0 { pruneUttDir() } } } emit(ev) } private func pruneUttDir() { let fm = FileManager.default guard let files = try? fm.contentsOfDirectory(atPath: Self.uttDir) else { return } let now = Date().timeIntervalSince1970 for f in files { let p = (Self.uttDir as NSString).appendingPathComponent(f) if let a = try? fm.attributesOfItem(atPath: p), let m = a[.modificationDate] as? Date, now - m.timeIntervalSince1970 > 30 { try? fm.removeItem(atPath: p) } } } func append(_ buffer: AVAudioPCMBuffer) { guard let payload = convertToPCM16(buffer) else { return } if source == "mic" { // keep a rolling ~6s window for per-utterance voice ID pcmLock.lock() uttPCM.append(payload) let maxBytes = 16000 * 2 * 6 if uttPCM.count > maxBytes { uttPCM.removeFirst(uttPCM.count - maxBytes) } pcmLock.unlock() } lock.lock(); let isReady = ready; lock.unlock() guard isReady, let task = task else { return } task.send(.data(payload)) { _ in } } private func convertToPCM16(_ buffer: AVAudioPCMBuffer) -> Data? { if buffer.format.sampleRate == targetFormat.sampleRate, buffer.format.channelCount == 1, buffer.format.commonFormat == .pcmFormatInt16, let ch = buffer.int16ChannelData { return Data(bytes: ch[0], count: Int(buffer.frameLength) * 2) } if converter == nil { converter = AVAudioConverter(from: buffer.format, to: targetFormat) } guard let converter = converter else { return nil } let ratio = targetFormat.sampleRate / buffer.format.sampleRate let cap = AVAudioFrameCount(Double(buffer.frameLength) * ratio + 1024) guard let out = AVAudioPCMBuffer(pcmFormat: targetFormat, frameCapacity: cap) else { return nil } var fed = false var err: NSError? let status = converter.convert(to: out, error: &err) { _, outStatus in if fed { outStatus.pointee = .noDataNow; return nil } fed = true; outStatus.pointee = .haveData; return buffer } if status == .error { return nil } guard out.frameLength > 0, let ch = out.int16ChannelData else { return nil } return Data(bytes: ch[0], count: Int(out.frameLength) * 2) } func stop() { closed = true healthTimer?.cancel(); healthTimer = nil task?.cancel(with: .goingAway, reason: nil) } } // --------------------------------------------------------------------------- // MARK: - Continuous on-device audio buffer (offline recovery store) // --------------------------------------------------------------------------- // Persists ALL captured audio as 60s 16k-mono WAV chunks under // ~/.skykoi/capture/audio-buffer/ so that anything the live STT missed // (rate limits, network loss — e.g. the 2026-07-05 Soniox 429 storm that // deafened the koi for 75 minutes) can be re-transcribed by the runtime's // recovery worker the moment STT is reachable again. Retention: 24h / 6GB. func wavData(pcm16: Data, sampleRate: Int) -> Data { var d = Data() func le32(_ v: UInt32) { withUnsafeBytes(of: v.littleEndian) { d.append(contentsOf: $0) } } func le16(_ v: UInt16) { withUnsafeBytes(of: v.littleEndian) { d.append(contentsOf: $0) } } d.append("RIFF".data(using: .ascii)!); le32(UInt32(36 + pcm16.count)) d.append("WAVE".data(using: .ascii)!); d.append("fmt ".data(using: .ascii)!) le32(16); le16(1); le16(1); le32(UInt32(sampleRate)); le32(UInt32(sampleRate * 2)); le16(2); le16(16) d.append("data".data(using: .ascii)!); le32(UInt32(pcm16.count)) d.append(pcm16) return d } final class BufferWriter: AudioSink { private let source: String private let queue = DispatchQueue(label: "audio-buffer", qos: .utility) private var converter: AVAudioConverter? private let targetFormat = AVAudioFormat( commonFormat: .pcmFormatInt16, sampleRate: 16000, channels: 1, interleaved: true)! private var pcm = Data() private var chunkStartMs: Int64 = 0 private let chunkBytes = 16000 * 2 * 60 // 60s of 16k mono int16 private let maxAgeSec: Double = 24 * 3600 // keep 24 hours private let maxTotalBytes: Int64 = 6 << 30 // 6 GB cap across sources static let dir: String = { let d = (NSHomeDirectory() as NSString).appendingPathComponent(".skykoi/capture/audio-buffer") try? FileManager.default.createDirectory(atPath: d, withIntermediateDirectories: true) return d }() init(source: String) { self.source = source } func append(_ buffer: AVAudioPCMBuffer) { guard let payload = convert(buffer) else { return } queue.async { [weak self] in guard let self = self else { return } if self.pcm.isEmpty { // Stamp the chunk with the wall-clock time of its FIRST sample. self.chunkStartMs = Int64(Date().timeIntervalSince1970 * 1000) - Int64(payload.count / 32) } self.pcm.append(payload) if self.pcm.count >= self.chunkBytes { self.flush() } } } func stop() { queue.sync { self.flush() } } private func flush() { guard pcm.count >= 16000 else { pcm = Data(); return } // <0.5s — not worth a file let data = pcm; pcm = Data() let path = (Self.dir as NSString).appendingPathComponent("\(source)-\(chunkStartMs).wav") try? wavData(pcm16: data, sampleRate: 16000).write(to: URL(fileURLWithPath: path)) prune() } private func prune() { let fm = FileManager.default guard let files = try? fm.contentsOfDirectory(atPath: Self.dir) else { return } let now = Date().timeIntervalSince1970 var entries: [(path: String, mtime: Double, size: Int64)] = [] for f in files where f.hasSuffix(".wav") { let p = (Self.dir as NSString).appendingPathComponent(f) guard let a = try? fm.attributesOfItem(atPath: p) else { continue } let m = (a[.modificationDate] as? Date)?.timeIntervalSince1970 ?? now let s = (a[.size] as? NSNumber)?.int64Value ?? 0 if now - m > maxAgeSec { try? fm.removeItem(atPath: p); continue } entries.append((p, m, s)) } var total = entries.reduce(Int64(0)) { $0 + $1.size } if total > maxTotalBytes { for e in entries.sorted(by: { $0.mtime < $1.mtime }) { try? fm.removeItem(atPath: e.path) total -= e.size if total <= maxTotalBytes { break } } } } private func convert(_ buffer: AVAudioPCMBuffer) -> Data? { if buffer.format.sampleRate == targetFormat.sampleRate, buffer.format.channelCount == 1, buffer.format.commonFormat == .pcmFormatInt16, let ch = buffer.int16ChannelData { return Data(bytes: ch[0], count: Int(buffer.frameLength) * 2) } if converter == nil { converter = AVAudioConverter(from: buffer.format, to: targetFormat) } guard let converter = converter else { return nil } let ratio = targetFormat.sampleRate / buffer.format.sampleRate let cap = AVAudioFrameCount(Double(buffer.frameLength) * ratio + 1024) guard let out = AVAudioPCMBuffer(pcmFormat: targetFormat, frameCapacity: cap) else { return nil } var fed = false var err: NSError? let status = converter.convert(to: out, error: &err) { _, outStatus in if fed { outStatus.pointee = .noDataNow; return nil } fed = true; outStatus.pointee = .haveData; return buffer } if status == .error { return nil } guard out.frameLength > 0, let ch = out.int16ChannelData else { return nil } return Data(bytes: ch[0], count: Int(out.frameLength) * 2) } } /// Fan one audio source out to several sinks (live STT + the recovery buffer). final class TeeSink: AudioSink { private let sinks: [AudioSink] init(_ sinks: [AudioSink]) { self.sinks = sinks } func append(_ buffer: AVAudioPCMBuffer) { for s in sinks { s.append(buffer) } } func stop() { for s in sinks { s.stop() } } } /// Pick the STT sink for a source: Soniox streaming if configured, else the /// legacy Whisper batch recorder. Every source is ALSO teed into the /// on-device audio buffer so nothing is ever lost to an STT outage. func makeSink(source: String) -> AudioSink { let stt: AudioSink if let s = sonioxConfig { stt = SonioxStreamer(cfg: s, source: source) } else { stt = AudioRecorder(source: source) } return TeeSink([stt, BufferWriter(source: source)]) } // --------------------------------------------------------------------------- // MARK: - CMSampleBuffer → AVAudioPCMBuffer // --------------------------------------------------------------------------- func createPCMBuffer(from sampleBuffer: CMSampleBuffer) -> AVAudioPCMBuffer? { guard let formatDesc = CMSampleBufferGetFormatDescription(sampleBuffer), let asbd = CMAudioFormatDescriptionGetStreamBasicDescription(formatDesc) else { return nil } let numSamples = CMSampleBufferGetNumSamples(sampleBuffer) guard numSamples > 0, let format = AVAudioFormat(streamDescription: asbd), let buffer = AVAudioPCMBuffer(pcmFormat: format, frameCapacity: AVAudioFrameCount(numSamples)) else { return nil } buffer.frameLength = AVAudioFrameCount(numSamples) let status = CMSampleBufferCopyPCMDataIntoAudioBufferList( sampleBuffer, at: 0, frameCount: Int32(numSamples), into: buffer.mutableAudioBufferList) return status == noErr ? buffer : nil } // --------------------------------------------------------------------------- // MARK: - Whisper API upload // --------------------------------------------------------------------------- func transcribeFile(at url: URL, source: String) async { guard let audioData = try? Data(contentsOf: url) else { return } if audioData.count < 1000 { return } // skip tiny files let boundary = "----SkykoiBoundary\(Int(Date().timeIntervalSince1970))" var request = URLRequest(url: URL(string: whisperConfig.url)!) request.httpMethod = "POST" request.setValue("Bearer \(whisperConfig.apiKey)", forHTTPHeaderField: "Authorization") request.setValue("multipart/form-data; boundary=\(boundary)", forHTTPHeaderField: "Content-Type") request.timeoutInterval = 30 var body = Data() func appendUTF8(_ s: String) { body.append(s.data(using: .utf8)!) } appendUTF8("--\(boundary)\r\nContent-Disposition: form-data; name=\"file\"; filename=\"segment.wav\"\r\nContent-Type: audio/wav\r\n\r\n") body.append(audioData) appendUTF8("\r\n--\(boundary)\r\nContent-Disposition: form-data; name=\"model\"\r\n\r\n\(whisperConfig.model)\r\n") appendUTF8("--\(boundary)\r\nContent-Disposition: form-data; name=\"response_format\"\r\n\r\njson\r\n") // Bias toward the assistant's name so "koi" never comes back as coy/koy. appendUTF8("--\(boundary)\r\nContent-Disposition: form-data; name=\"prompt\"\r\n\r\nThe user may address their assistant named Koi (e.g. \"hey Koi\").\r\n") appendUTF8("--\(boundary)--\r\n") request.httpBody = body do { let (data, response) = try await URLSession.shared.data(for: request) let code = (response as? HTTPURLResponse)?.statusCode ?? 0 if code == 200 { if let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any], let text = json["text"] as? String { let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines) if !trimmed.isEmpty && !isBlankAudio(trimmed) { let eventType = source == "mic" ? "mic_transcript" : "audio_transcript" emit(["type": eventType, "ts": tsNow(), "text": trimmed, "source": source, "isFinal": true]) } } } else if code == 429 { // Rate limited — back off silently try? await Task.sleep(nanoseconds: 5_000_000_000) } else { let errText = (String(data: data, encoding: .utf8) ?? "").prefix(200) emit(["type": "error", "ts": tsNow(), "message": "Whisper API \(code): \(errText)"]) } } catch { emit(["type": "error", "ts": tsNow(), "message": "Whisper request failed: \(error.localizedDescription)"]) } } func isBlankAudio(_ text: String) -> Bool { let lower = text.lowercased() return lower.contains("[blank_audio]") || lower.contains("(blank audio)") || lower == "you" || lower == "." || lower.count < 3 } // --------------------------------------------------------------------------- // MARK: - Audio segment recorder // --------------------------------------------------------------------------- class AudioRecorder: AudioSink { let source: String let speechRMS: Float = 0.01 let silenceRMS: Float = 0.005 let silenceFlushDelay: Double = 1.5 let maxSegmentSeconds: Double = 15.0 private var buffers: [AVAudioPCMBuffer] = [] private var totalFrames: AVAudioFrameCount = 0 private var format: AVAudioFormat? private let queue = DispatchQueue(label: "recorder", qos: .userInitiated) private var checkTimer: DispatchSourceTimer? private var inflightCount = 0 private let maxInflight = 3 private var hadSpeech = false private var silenceStartTime: CFAbsoluteTime = 0 private var segmentStartTime: CFAbsoluteTime = 0 init(source: String) { self.source = source let timer = DispatchSource.makeTimerSource(queue: queue) timer.schedule(deadline: .now() + .milliseconds(200), repeating: .milliseconds(200)) timer.setEventHandler { [weak self] in self?.checkFlush() } timer.resume() checkTimer = timer } private func computeRMS(_ buffer: AVAudioPCMBuffer) -> Float { var sumSq: Float = 0 var count = 0 if let data = buffer.floatChannelData?[0] { for i in 0.. 0 ? sqrt(sumSq / Float(count)) : 0 } func append(_ buffer: AVAudioPCMBuffer) { queue.async { [weak self] in guard let self = self else { return } if self.format == nil { self.format = buffer.format } let rms = self.computeRMS(buffer) let now = CFAbsoluteTimeGetCurrent() if rms >= self.speechRMS { if !self.hadSpeech { self.hadSpeech = true self.segmentStartTime = now } self.silenceStartTime = 0 } else if rms < self.silenceRMS && self.hadSpeech && self.silenceStartTime == 0 { self.silenceStartTime = now } self.buffers.append(buffer) self.totalFrames += buffer.frameLength } } private func checkFlush() { guard totalFrames > 0 else { return } let now = CFAbsoluteTimeGetCurrent() if hadSpeech { let maxExceeded = (now - segmentStartTime) >= maxSegmentSeconds let silenceExceeded = silenceStartTime > 0 && (now - silenceStartTime) >= silenceFlushDelay if maxExceeded || silenceExceeded { flush() } } else { // Discard ambient noise buffers older than 5s to prevent unbounded growth if let fmt = format { let maxFrames = AVAudioFrameCount(5.0 * fmt.sampleRate) if totalFrames > maxFrames { buffers = [] totalFrames = 0 } } } } func stop() { checkTimer?.cancel() queue.sync { flush() } } private func flush() { guard totalFrames > 0, let format = format else { return } if inflightCount >= maxInflight { return } let currentBuffers = buffers buffers = [] totalFrames = 0 hadSpeech = false silenceStartTime = 0 segmentStartTime = 0 // Compute overall RMS to skip all-silence segments var sumSq: Float = 0 var sampleCount = 0 for buf in currentBuffers { if let data = buf.floatChannelData?[0] { for i in 0.. 0 ? sqrt(sumSq / Float(sampleCount)) : 0 if rms < silenceRMS { return } let tempPath = NSTemporaryDirectory() + "skykoi-\(source)-\(Int(Date().timeIntervalSince1970 * 1000)).wav" let tempURL = URL(fileURLWithPath: tempPath) do { let settings: [String: Any] = [ AVFormatIDKey: Int(kAudioFormatLinearPCM), AVSampleRateKey: format.sampleRate, AVNumberOfChannelsKey: format.channelCount, AVLinearPCMBitDepthKey: 16, AVLinearPCMIsFloatKey: false, AVLinearPCMIsBigEndianKey: false, ] let file = try AVAudioFile(forWriting: tempURL, settings: settings) for buf in currentBuffers { try file.write(from: buf) } } catch { return } inflightCount += 1 Task { [weak self] in await transcribeFile(at: tempURL, source: self?.source ?? "unknown") try? FileManager.default.removeItem(at: tempURL) self?.queue.async { self?.inflightCount -= 1 } } } } // --------------------------------------------------------------------------- // MARK: - Mic capture // --------------------------------------------------------------------------- class MicCapture { private var engine = AVAudioEngine() private var sink: AudioSink? private var bufCount = 0 private var levelMax: Float = 0 private var lastBufferAt = Date() private var restartPending = false private var watchdog: DispatchSourceTimer? private func statusString(_ s: AVAuthorizationStatus) -> String { switch s { case .authorized: return "authorized" case .denied: return "denied" case .restricted: return "restricted" case .notDetermined: return "notDetermined" @unknown default: return "unknown" } } func start() { // Report the mic permission THIS process actually has. A daemon with no // grant still "starts" the engine but only receives silence, so this is the // one reliable signal that the mic is truly usable. (We do NOT call // requestAccess here — in a background context it auto-denies in a few ms // and can suppress AVAudioEngine's own implicit consent prompt. Letting the // engine trigger the prompt is the path that actually surfaces "Allow".) let status = AVCaptureDevice.authorizationStatus(for: .audio) emit(["type": "mic_permission", "ts": tsNow(), "status": statusString(status)]) beginCapture(reason: "startup") // A meeting app grabbing or switching the input device fires a // configuration change and silently kills the tap — the "koi went deaf // mid-meeting" bug (2026-07-03). Rebuild the whole capture (fresh engine + // fresh sink, since the Soniox converter is locked to the first buffer // format) whenever it happens. object: nil on purpose — the engine is // recreated on every restart, so binding to one instance would detach. NotificationCenter.default.addObserver( forName: .AVAudioEngineConfigurationChange, object: nil, queue: nil ) { [weak self] _ in self?.scheduleRestart(reason: "input configuration change") } // Belt: CoreAudio delivers buffers continuously (silence included), so a // stalled tap or a stopped engine means the mic is DEAD, not quiet. let t = DispatchSource.makeTimerSource(queue: .main) t.schedule(deadline: .now() + 30, repeating: 30) t.setEventHandler { [weak self] in guard let self = self, !self.restartPending else { return } if !self.engine.isRunning { self.scheduleRestart(reason: "engine stopped") } else if Date().timeIntervalSince(self.lastBufferAt) > 60 { self.scheduleRestart(reason: "buffer stall (>60s)") } } t.resume() watchdog = t } private func beginCapture(reason: String) { sink?.stop() sink = makeSink(source: "mic") engine = AVAudioEngine() let inputNode = engine.inputNode let nativeFormat = inputNode.outputFormat(forBus: 0) lastBufferAt = Date() inputNode.installTap(onBus: 0, bufferSize: 8192, format: nativeFormat) { [weak self] buffer, _ in guard let self = self else { return } self.lastBufferAt = Date() self.sink?.append(buffer) // Periodic mic level — proves real (non-silent) audio is arriving, which // a denied mic (zeroed buffers) never produces. AEC-independent. if let ch = buffer.floatChannelData { let n = Int(buffer.frameLength) if n > 0 { var sum: Float = 0 let data = ch[0] for i in 0..= 30 { // ~ every 5s at 48kHz/8192 emit(["type": "mic_level", "ts": tsNow(), "rms": self.levelMax]) self.bufCount = 0; self.levelMax = 0 } } do { try engine.start() emit(["type": "mic_started", "ts": tsNow(), "reason": reason, "sampleRate": nativeFormat.sampleRate, "channels": nativeFormat.channelCount]) } catch { emit(["type": "error", "ts": tsNow(), "message": "Mic engine failed (\(reason)): \(error.localizedDescription)"]) } } private func scheduleRestart(reason: String) { DispatchQueue.main.async { [weak self] in guard let self = self, !self.restartPending else { return } self.restartPending = true emit(["type": "mic_restarting", "ts": tsNow(), "reason": reason]) // Let the route settle before rebuilding. DispatchQueue.main.asyncAfter(deadline: .now() + 1.5) { [weak self] in guard let self = self else { return } self.engine.inputNode.removeTap(onBus: 0) self.engine.stop() self.beginCapture(reason: reason) self.restartPending = false } } } func stop() { watchdog?.cancel() engine.inputNode.removeTap(onBus: 0) engine.stop() sink?.stop() } } // --------------------------------------------------------------------------- // MARK: - System audio capture via ScreenCaptureKit // --------------------------------------------------------------------------- class SystemAudioCapture: NSObject, SCStreamOutput, SCStreamDelegate { private var stream: SCStream? private var sink: AudioSink? private let audioQueue = DispatchQueue(label: "system-audio", qos: .userInitiated) func start() { sink = makeSink(source: "system") Task { await beginCapture() } } func stop() { Task { try? await stream?.stopCapture() } sink?.stop() } private func beginCapture() async { do { let content = try await SCShareableContent.excludingDesktopWindows(false, onScreenWindowsOnly: false) guard let display = content.displays.first else { emit(["type": "error", "ts": tsNow(), "message": "No display for audio capture"]) return } let filter = SCContentFilter(display: display, excludingWindows: []) let config = SCStreamConfiguration() config.capturesAudio = true config.excludesCurrentProcessAudio = true config.sampleRate = 16000 config.channelCount = 1 config.width = 2 config.height = 2 config.minimumFrameInterval = CMTime(value: 10, timescale: 1) config.showsCursor = false stream = SCStream(filter: filter, configuration: config, delegate: self) try stream?.addStreamOutput(self, type: .audio, sampleHandlerQueue: audioQueue) try await stream?.startCapture() emit(["type": "system_audio_started", "ts": tsNow()]) } catch { emit(["type": "error", "ts": tsNow(), "message": "System audio capture failed: \(error.localizedDescription)"]) } } func stream(_ stream: SCStream, didOutputSampleBuffer sampleBuffer: CMSampleBuffer, of type: SCStreamOutputType) { guard type == .audio else { return } guard let pcmBuffer = createPCMBuffer(from: sampleBuffer) else { return } sink?.append(pcmBuffer) } func stream(_ stream: SCStream, didStopWithError error: Error) { emit(["type": "error", "ts": tsNow(), "message": "System audio stopped: \(error.localizedDescription)"]) DispatchQueue.main.asyncAfter(deadline: .now() + 3) { [weak self] in Task { await self?.beginCapture() } } } } // --------------------------------------------------------------------------- // MARK: - Main // --------------------------------------------------------------------------- // Prefer Soniox real-time streaming (diarized, ~50ms); fall back to Whisper // batch only if no Soniox key is configured. sonioxConfig = loadSonioxConfig() if sonioxConfig == nil { guard let config = loadWhisperConfig() else { emit(["type": "error", "ts": tsNow(), "message": "No SONIOX_API_KEY, OPENAI_API_KEY or GROQ_API_KEY found in env or ~/.skykoi/.env — audio transcription disabled"]) exit(1) } whisperConfig = config } emit(["type": "audio_daemon_started", "ts": tsNow(), "stt": sonioxConfig != nil ? "soniox" : "whisper"]) let mic = MicCapture() let system = SystemAudioCapture() mic.start() // System audio is opt-in (default off) so the ScreenCaptureKit permission // nag never recurs for normal use. The mic grant is durable (stable cert). let systemAudioOn = systemAudioEnabled() if systemAudioOn { system.start() } else { emit(["type": "system_audio_disabled", "ts": tsNow(), "message": "system audio capture is opt-in (SKYKOI_CAPTURE_SYSTEM_AUDIO=1 or settings.json systemAudio:true)"]) } // Process-level STT deadman: if the live-STT link is silent for 10 minutes, // exit(0) so launchd (KeepAlive) restarts us with fresh networking and fresh // streamer state. Unlike the per-streamer watchdog, this cannot die with a // wedged streamer object. 10 min never fires across a legit 60–90s 429 // backoff window; during a real outage one restart every 10 min is cheap. // Also self-heals the post-sleep/wake URLSession wedge on long naps. var sttDeadman: DispatchSourceTimer? if sonioxConfig != nil { let t = DispatchSource.makeTimerSource(queue: .global()) t.schedule(deadline: .now() + 600, repeating: 60) t.setEventHandler { sttMsgLock.lock(); let last = lastSttMsgAt; sttMsgLock.unlock() let silent = Date().timeIntervalSince1970 - last if silent > 600 { emit(["type": "error", "ts": tsNow(), "message": "STT deadman: no Soniox message for \(Int(silent))s — exiting for a clean launchd restart"]) if outputFd >= 0 { close(outputFd) } exit(0) } } t.resume() sttDeadman = t } let sigSources: [DispatchSourceSignal] = [SIGTERM, SIGINT].map { sig in signal(sig, SIG_IGN) let src = DispatchSource.makeSignalSource(signal: sig, queue: .main) src.setEventHandler { mic.stop() system.stop() if outputFd >= 0 { close(outputFd) } exit(0) } src.resume() return src } withExtendedLifetime((sigSources, mic, system, sttDeadman)) { CFRunLoopRun() }