feat: live draft pipe transcription (ApplePipeTranscriber --pipe volatile)

- Reuses whisper-translation/apple_speech pipe protocol: 4-byte BE len + WAV -> draft/final JSONL
- Persistent Swift session per locale, 60s idle auto-close
- New tools: speech_live_transcribe (base64 WAV chunk -> drafts[] finals[]), speech_live_close, speech_live_status
- Engine: SpeechAnalyzer + SpeechTranscriber macOS 26.5 with .volatileResults for <200ms draft feedback
- Intended for ESP32 voice gateway WS live ingestion

src/integrations/speech-live.js

@@ -0,0 +1,281 @@
+import { execFile, spawn } from "node:child_process";
+import { promisify } from "node:util";
+import { writeFile, readFile, stat, rm, mkdir, mkdtemp } from "node:fs/promises";
+import path from "node:path";
+import os from "node:os";
+
+const execFileAsync = promisify(execFile);
+
+function swiftSourcePipeTranscriber() {
+  return `
+import AVFoundation
+import Foundation
+import Speech
+
+func parseArgs() -> (String, Bool) {
+    var localeId = "en-US"
+    var verbose = false
+    var i = 1
+    let raw = CommandLine.arguments
+    while i < raw.count {
+        let a = raw[i]
+        if a == "--locale", i+1 < raw.count { localeId = raw[i+1]; i+=1 }
+        else if a == "-v" || a == "--verbose" { verbose = true }
+        i+=1
+    }
+    return (localeId, verbose)
+}
+func logv(_ msg: String, verbose: Bool) { if verbose { fputs("[apple-pipe] \\(msg)\\n", stderr) } }
+
+@main
+struct ApplePipeCLI {
+    static func main() async {
+        let (localeId, verbose) = parseArgs()
+        guard SpeechTranscriber.isAvailable else { fputs("Not available\\n", stderr); exit(1) }
+        let reqLocale = Locale(identifier: localeId)
+        let locale: Locale
+        if let sup = await SpeechTranscriber.supportedLocale(equivalentTo: reqLocale) { locale = sup }
+        else { locale = reqLocale }
+        // warm asset check
+        let warm = SpeechTranscriber(locale: locale, transcriptionOptions: [], reportingOptions: [.volatileResults], attributeOptions: [])
+        let status = await AssetInventory.status(forModules: [warm])
+        switch status {
+        case .installed: logv("Assets installed", verbose: verbose)
+        case .unsupported: fputs("Locale unsupported\\n", stderr); exit(2)
+        case .supported:
+            logv("Downloading assets...", verbose: true)
+            do { if let req = try await AssetInventory.assetInstallationRequest(supporting: [warm]) { try await req.downloadAndInstall() } }
+            catch { fputs("Asset download failed: \\(error)\\n", stderr); exit(3) }
+        case .downloading:
+            logv("Waiting assets...", verbose: true)
+            for _ in 0..<30 { try? await Task.sleep(nanoseconds: 1_000_000_000); if await AssetInventory.status(forModules: [warm]) == .installed { break } }
+        @unknown default: break
+        }
+        logv("Pipe ready locale=\\(locale.identifier)", verbose: true)
+        let stdinH = FileHandle.standardInput
+        var leftover = Data()
+        var chunkIdx = 0
+        func readExact(_ n: Int) -> Data? {
+            var out = Data(); out.reserveCapacity(n)
+            if leftover.count >= n { let d = leftover.prefix(n); leftover = leftover.dropFirst(n); return Data(d) }
+            if leftover.count > 0 { out.append(leftover); leftover = Data() }
+            while out.count < n {
+                let d = stdinH.readData(ofLength: n - out.count)
+                if d.isEmpty { if out.count==0 { return nil }; return nil }
+                out.append(d)
+            }
+            return out
+        }
+        while true {
+            guard let lenData = readExact(4) else { break }
+            let length = lenData.withUnsafeBytes { $0.load(as: UInt32.self).bigEndian }
+            if length == 0 { logv("Pipe EOF", verbose: true); break }
+            if length > 20_000_000 { fputs("Chunk too large \\(length)\\n", stderr); break }
+            guard let chunkData = readExact(Int(length)) else { fputs("Truncated expected \\(length)\\n", stderr); break }
+            chunkIdx += 1
+            let tmpURL = URL(fileURLWithPath: NSTemporaryDirectory()).appendingPathComponent("apple-pipe-\\(ProcessInfo.processInfo.processIdentifier)-\\(chunkIdx).wav")
+            do { try chunkData.write(to: tmpURL) } catch { fputs("Write err: \\(error)\\n", stderr); continue }
+            guard let audioFile = try? AVAudioFile(forReading: tmpURL) else { try? FileManager.default.removeItem(at: tmpURL); fputs("Open fail \\(chunkIdx)\\n", stderr); continue }
+            let t = SpeechTranscriber(locale: locale, transcriptionOptions: [], reportingOptions: [.volatileResults], attributeOptions: [.audioTimeRange])
+            guard let analyzer = try? await SpeechAnalyzer(inputAudioFile: audioFile, modules: [t], finishAfterFile: true) else { try? FileManager.default.removeItem(at: tmpURL); continue }
+            do {
+                for try await res in t.results {
+                    let txt = String(res.text.characters).trimmingCharacters(in: .whitespacesAndNewlines)
+                    if txt.isEmpty { continue }
+                    let d: [String: Any] = [
+                        "event": res.isFinal ? "final" : "draft",
+                        "text": txt, "isFinal": res.isFinal,
+                        "chunk": chunkIdx, "start": res.range.start.seconds,
+                        "duration": res.range.duration.seconds
+                    ]
+                    if let jd = try? JSONSerialization.data(withJSONObject: d), let s = String(data: jd, encoding: .utf8) {
+                        print(s); fflush(stdout)
+                    }
+                }
+            } catch { fputs("Results err \\(chunkIdx): \\(error)\\n", stderr) }
+            _ = analyzer
+            try? FileManager.default.removeItem(at: tmpURL)
+        }
+        logv("Pipe done \\(chunkIdx)", verbose: true)
+    }
+}
+`;
+}
+
+class ApplePipeSession {
+  constructor(locale) {
+    this.locale = locale;
+    this.proc = null;
+    this.tmpDir = null;
+    this.binFile = null;
+    this.ready = false;
+    this.chunkIdx = 0;
+    this.lastActivity = Date.now();
+    this._lineCallback = null;
+  }
+  async ensureBuilt() {
+    const tmpBase = path.join(os.tmpdir(), "speech-live-");
+    this.tmpDir = await mkdtemp(tmpBase);
+    const swiftFile = path.join(this.tmpDir, "Main.swift");
+    this.binFile = path.join(this.tmpDir, "apple-pipe-transcribe");
+    await writeFile(swiftFile, swiftSourcePipeTranscriber(), "utf8");
+    try {
+      await execFileAsync("/usr/bin/swiftc", ["-O", "-parse-as-library", swiftFile, "-o", this.binFile, "-framework", "AVFoundation", "-framework", "Speech"], { timeout: 60000, maxBuffer: 20*1024*1024 });
+    } catch (e) {
+      throw new Error(`swiftc build failed: ${e.stderr||e.message}`);
+    }
+  }
+  async start() {
+    if (this.proc && this.proc.exitCode === null && this.ready) return;
+    if (!this.binFile) await this.ensureBuilt();
+    return new Promise((resolve, reject) => {
+      const args = ["--locale", this.locale];
+      this.proc = spawn(this.binFile, args, { stdio: ["pipe", "pipe", "pipe"] });
+      let stderrBuf = "";
+      this.proc.stderr.on("data", d => { stderrBuf += d.toString(); });
+      setTimeout(() => {
+        if (this.proc.exitCode !== null) {
+          reject(new Error(`apple-pipe exited early code=${this.proc.exitCode} stderr=${stderrBuf.slice(0,2000)}`));
+        } else {
+          this.ready = true;
+          this._setupReader();
+          resolve();
+        }
+      }, 800);
+      this.proc.on("error", reject);
+    });
+  }
+  _setupReader() {
+    let buf = "";
+    this.proc.stdout.on("data", chunk => {
+      buf += chunk.toString("utf8");
+      let lines = buf.split("\n");
+      buf = lines.pop() || "";
+      for (let line of lines) {
+        line = line.trim();
+        if (!line) continue;
+        try {
+          const obj = JSON.parse(line);
+          if (this._lineCallback) this._lineCallback(obj);
+        } catch {}
+      }
+    });
+  }
+  async transcribeChunk(wavBytes, { timeoutMs = 6000 } = {}) {
+    await this.start();
+    this.chunkIdx++;
+    const myIdx = this.chunkIdx;
+    return new Promise((resolve, reject) => {
+      let drafts = [];
+      let finals = [];
+      let timer = null;
+      let done = false;
+      const cleanup = () => { done = true; if (timer) clearTimeout(timer); this._lineCallback = null; };
+      const finish = () => {
+        if (done) return;
+        cleanup();
+        const fullFinal = finals.map(f=>f.text).join(" ").trim();
+        const lastDraft = drafts.length ? drafts[drafts.length-1].text : "";
+        const text = fullFinal || lastDraft || "";
+        resolve({ text, finals, drafts, chunk: myIdx, isFinal: finals.length>0 });
+      };
+      this._lineCallback = (obj) => {
+        if (obj.chunk !== myIdx) return;
+        if (obj.isFinal || obj.event==="final") {
+          finals.push(obj);
+          if (timer) clearTimeout(timer);
+          timer = setTimeout(finish, 350);
+        } else {
+          drafts.push(obj);
+          if (timer) clearTimeout(timer);
+          timer = setTimeout(finish, 700);
+        }
+      };
+      timer = setTimeout(() => { if (!done) finish(); }, timeoutMs);
+      try {
+        const lenBuf = Buffer.alloc(4);
+        lenBuf.writeUInt32BE(wavBytes.length, 0);
+        this.proc.stdin.write(Buffer.concat([lenBuf, Buffer.from(wavBytes)]));
+      } catch (e) {
+        cleanup();
+        reject(e);
+      }
+    });
+  }
+  async close() {
+    try {
+      if (this.proc && this.proc.stdin.writable) {
+        const eof = Buffer.alloc(4); eof.writeUInt32BE(0,0);
+        this.proc.stdin.write(eof);
+        this.proc.stdin.end();
+      }
+      if (this.proc) {
+        await new Promise(r => { this.proc.on("close", r); setTimeout(r, 1500); });
+        try { this.proc.kill(); } catch {}
+      }
+    } catch {}
+    try { if (this.tmpDir) await rm(this.tmpDir, {recursive:true, force:true}); } catch {}
+    this.proc = null;
+    this.ready = false;
+  }
+}
+
+const sessions = new Map();
+
+async function getOrCreateSession(locale) {
+  let s = sessions.get(locale);
+  if (!s) {
+    s = new ApplePipeSession(locale);
+    sessions.set(locale, s);
+  }
+  await s.start();
+  s.lastActivity = Date.now();
+  return s;
+}
+
+export async function speechLiveTranscribe({ audioBase64, locale } = {}) {
+  if (!audioBase64) throw new Error("audioBase64 required (WAV 16k mono base64)");
+  const wavBytes = Buffer.from(audioBase64, "base64");
+  const loc = locale || "en-US";
+  const session = await getOrCreateSession(loc);
+  const res = await session.transcribeChunk(wavBytes, { timeoutMs: 8000 });
+  return {
+    ok: true,
+    engine: "ApplePipeTranscriber/macOS26.5 --pipe volatile drafts",
+    locale: loc,
+    text: res.text,
+    isFinal: res.isFinal,
+    drafts: res.drafts,
+    finals: res.finals,
+    chunk: res.chunk,
+    realtime: true
+  };
+}
+
+export async function speechLiveClose({ locale } = {}) {
+  const loc = locale || "en-US";
+  const s = sessions.get(loc);
+  if (s) {
+    await s.close();
+    sessions.delete(loc);
+  }
+  return { ok: true, closed: loc };
+}
+
+export async function speechLiveStatus() {
+  const info = [];
+  for (let [loc, s] of sessions.entries()) {
+    info.push({ locale: loc, ready: s.ready, chunkIdx: s.chunkIdx, lastActivity: new Date(s.lastActivity).toISOString(), pid: s.proc?.pid || null });
+  }
+  return { sessions: info, count: info.length };
+}
+
+setInterval(async () => {
+  const now = Date.now();
+  for (let [loc, s] of sessions.entries()) {
+    if (now - s.lastActivity > 60000) {
+      try { await s.close(); } catch {}
+      sessions.delete(loc);
+    }
+  }
+}, 15000);

src/server.js

@@ -36,6 +36,11 @@ import {
   speechListLocales,
   speechQuickTest,
 } from "./integrations/speech-transcribe.js";
+import {
+  speechLiveTranscribe,
+  speechLiveClose,
+  speechLiveStatus,
+} from "./integrations/speech-live.js";
 
 const jsonResult = (value) => ({
   content: [{ type: "text", text: JSON.stringify(value, null, 2) }],
@@ -285,6 +290,33 @@ export function createMacMiniMcpServer() {
     handled(({ text, voice }) => speechQuickTest({ text, voice })),
   );
 
+
+  server.tool(
+    "speech_live_transcribe",
+    "LIVE draft transcription via persistent Apple pipe (ApplePipeTranscriber macOS 26+). Reuses your whisper-translation pipe protocol: 4-byte BE len + WAV payload -> draft/final JSON streaming. Input: base64-encoded 16kHz mono WAV. Output: {text, drafts[], finals[], isFinal}. Keep session alive for <200ms incremental feedback while streaming ESP32 PCM. Auto-closes after 60s idle.",
+    {
+      audioBase64: z.string().min(100).describe("Base64-encoded WAV file (16kHz mono s16le). Use 0.5-3s chunks for draft streaming."),
+      locale: z.string().optional().describe("Locale like en-US. Default en-US."),
+    },
+    handled(({ audioBase64, locale }) => speechLiveTranscribe({ audioBase64, locale })),
+  );
+
+  server.tool(
+    "speech_live_close",
+    "Close a persistent live transcription pipe session (frees Swift process).",
+    {
+      locale: z.string().optional().describe("Locale to close. Default en-US."),
+    },
+    handled(({ locale }) => speechLiveClose({ locale })),
+  );
+
+  server.tool(
+    "speech_live_status",
+    "Show active live pipe sessions: locale, pid, chunkIdx, lastActivity.",
+    {},
+    handled(() => speechLiveStatus()),
+  );
+
   server.tool(
     "codex_image_get_config_status",
     "Show local Codex CLI image generation configuration.",