"""Audio capture child process — isolates CoreAudio from the main app.

Why this exists: in 2026-06-03 we caught a wedged Lirelle where the MAIN
thread had blocked inside `event-tap callback → recorder._lock acquire`,
because a Timer thread holding that lock was stuck for ~19 hours inside
`stream.close()` → `HALB_Mutex::Lock` — an Apple-internal CoreAudio mutex
deadlock that's plausibly triggered by AOP (Always-On-Processor) firmware
instability on macOS 15.x. With everything in one process, a stuck
CoreAudio call paralyses the entire app: run loop dies, overlay frozen,
watchdog timers never fire, only kill-and-relaunch recovers.

Aqua Voice and Wispr Flow don't have this problem because they're
Electron apps — Chromium runs its audio service in a separate process,
and when CoreAudio wedges, only the audio service dies; the UI is fine.
This module gives us the same isolation in Python.

Wire protocol (binary, length-prefixed, parent ↔ child):

  Parent → child (commands, single byte each):
    b'A'  ARM           — start capturing into the buffer
    b'V'  ARM+VAD       — same as ARM but ALSO stream each callback's
                          chunk back via b'C' frames so the parent can
                          run voice-activity detection in real time
    b'D'  DISARM        — stop capturing; child sends back the audio frame
    b'P'  PING          — heartbeat; child replies with b'O' PONG
    b'Q'  QUIT          — graceful shutdown

  Child → parent (frames, [1 type byte][4-byte LE uint32 length][payload]):
    b'L'  payload = N float32 RMS levels (4*N bytes)
    b'C'  payload = N float32 audio samples — single callback chunk,
                    streamed in real time only while in VAD mode
    b'A'  payload = N float32 audio samples — full aggregate sent at DISARM
    b'O'  payload = empty — PONG reply to PING
    b'E'  payload = utf-8 error message

The child keeps a single lingering CoreAudio stream open across multiple
ARM/DISARM cycles (8 second idle → close), so back-to-back dictations
don't pay the ~100ms cold-start cost.
"""

from __future__ import annotations

import collections
import os
import struct
import sys
import threading
import time

# Stub heavy deps BEFORE importing anything that might pull them in transitively.
# Same logic as lumo_main.py — sounddevice doesn't need them, but defensive.
_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _root not in sys.path:
    sys.path.insert(0, _root)
try:
    from src.heavy_dep_stubs import install as _install_stubs
    _install_stubs()
except Exception:
    pass

import numpy as np
import sounddevice as sd


KEEPALIVE_SECONDS = 8.0
SAMPLE_RATE = 16000
CHANNELS = 1
LEVELS_BURST_INTERVAL_SEC = 0.05  # send levels to parent every 50ms while armed


# Serialises ALL writes to stdout. Two threads share it: the main command
# loop (writing AUDIO + PONG + ERROR frames) and the levels_pulse thread
# (writing LEVEL frames every 50ms). Without this lock, a large AUDIO
# frame (~96 KB for a 1.5s clip) chunks into multiple POSIX writes under
# PIPE_BUF=4 KB; a LEVEL write from the other thread could interleave
# mid-frame → parent reader desyncs → false "helper unresponsive". Hit
# this on the first dev smoke test.
_stdout_lock = threading.Lock()


def _write_frame(out, ftype: bytes, payload: bytes) -> None:
    """Length-prefixed write. ftype is a single byte. Payload up to 4 GiB.
    Hardened against ConnectionResetError so a parent that died doesn't
    crash the child — the parent's death is the QUIT signal we'd catch
    on the next stdin read anyway."""
    with _stdout_lock:
        try:
            out.write(ftype + struct.pack("<I", len(payload)) + payload)
            out.flush()
        except (BrokenPipeError, OSError):
            # Parent gone — bail; the stdin read will return b'' next and we exit.
            pass


def _send_error(out, msg: str) -> None:
    _write_frame(out, b"E", msg.encode("utf-8", errors="replace"))


def run() -> None:
    """Child-process entry. Runs until parent disconnects or sends QUIT."""
    # IMPORTANT: stdout MUST be binary, not the default text mode. The bytes
    # we write include arbitrary float32 audio samples — text mode would
    # mangle 0x0A bytes ("LF translation") on writes.
    stdin = sys.stdin.buffer
    stdout = sys.stdout.buffer

    # ── Shared state between the sounddevice callback (CoreAudio thread)
    #    and the main child thread (which is what's reading stdin).
    armed = False
    streaming = False  # send b'C' chunks to parent in real time (VAD mode)
    armed_lock = threading.Lock()  # also guards `streaming`
    audio_buffer: collections.deque[np.ndarray] = collections.deque()
    audio_buffer_lock = threading.Lock()
    levels: collections.deque[float] = collections.deque(maxlen=240)
    levels_lock = threading.Lock()
    stream: sd.InputStream | None = None
    stream_lock = threading.Lock()  # owns the CoreAudio stream lifecycle
    keepalive_timer: threading.Timer | None = None

    def callback(indata, frames, time_info, status):  # noqa: ARG001
        # Runs on the CoreAudio IOThread. Drop samples if not armed.
        if status:
            _send_error(stdout, f"sounddevice status: {status}")
        block = indata if indata.ndim == 1 else indata.mean(axis=1)
        rms = float(np.sqrt(np.mean(block.astype(np.float32) ** 2) + 1e-12))
        with levels_lock:
            levels.append(rms)
        with armed_lock:
            if not armed:
                return
            send_stream = streaming
        # Capture into buffer (used for the aggregate b'A' frame on DISARM).
        with audio_buffer_lock:
            audio_buffer.append(indata.copy())
        # Stream the same chunk to parent for VAD analysis if requested.
        # _write_frame is mutex-protected, safe to call from the IO thread.
        if send_stream:
            mono = block.astype(np.float32, copy=False)
            _write_frame(stdout, b"C", mono.tobytes())

    def ensure_stream() -> bool:
        """Open the stream if not already. Returns True on success."""
        nonlocal stream
        with stream_lock:
            if stream is not None:
                return True
            try:
                stream = sd.InputStream(
                    samplerate=SAMPLE_RATE,
                    channels=CHANNELS,
                    dtype="float32",
                    callback=callback,
                )
                stream.start()
                return True
            except Exception as e:
                _send_error(stdout, f"InputStream open failed: {e}")
                stream = None
                return False

    def schedule_keepalive_close() -> None:
        """(Re)arm the idle-close timer."""
        nonlocal keepalive_timer
        if keepalive_timer is not None:
            keepalive_timer.cancel()
        keepalive_timer = threading.Timer(KEEPALIVE_SECONDS, close_idle_stream)
        keepalive_timer.daemon = True
        keepalive_timer.start()

    def close_idle_stream() -> None:
        """Tear down the stream when the keepalive expires. CoreAudio
        teardown runs OUTSIDE stream_lock — if Apple's HAL mutex deadlocks
        inside stream.close() (the original reason this whole subprocess
        exists), the child process simply gets stuck on that close — but
        that's now CONTAINED to the child. Parent will heartbeat-detect
        unresponsiveness and SIGKILL+respawn us. The poisoned CoreAudio
        state stays buried in this dying child, the new child opens a
        fresh stream cleanly."""
        nonlocal stream, keepalive_timer
        with stream_lock:
            with armed_lock:
                if armed:
                    return  # ARM came back during the keepalive window
            local_stream = stream
            stream = None
            keepalive_timer = None
        if local_stream is not None:
            try:
                local_stream.stop()
                local_stream.close()
            except Exception as e:
                _send_error(stdout, f"keepalive close error: {e}")

    # ── Background thread: pulse levels to the parent every 50ms while armed.
    def levels_pulse():
        last_sent = 0.0
        while True:
            time.sleep(LEVELS_BURST_INTERVAL_SEC)
            with armed_lock:
                if not armed:
                    last_sent = 0.0
                    continue
            with levels_lock:
                if not levels:
                    continue
                # Send the most recent ~60 samples (3 seconds of 50ms ticks).
                snapshot = list(levels)[-60:]
            payload = struct.pack(f"<{len(snapshot)}f", *snapshot)
            _write_frame(stdout, b"L", payload)
            last_sent = time.time()

    threading.Thread(target=levels_pulse, daemon=True).start()

    # ── Main loop: read one command at a time from parent.
    while True:
        cmd = stdin.read(1)
        if not cmd:
            break  # parent closed pipe
        if cmd == b"A" or cmd == b"V":
            # ARM (b"A") or ARM-with-VAD-streaming (b"V"). Same prep — just
            # differs in whether we stream chunks back to the parent for
            # voice-activity detection.
            with audio_buffer_lock:
                audio_buffer.clear()
            with levels_lock:
                levels.clear()
            if not ensure_stream():
                continue
            if keepalive_timer is not None:
                keepalive_timer.cancel()
                keepalive_timer = None
            with armed_lock:
                armed = True
                streaming = (cmd == b"V")
        elif cmd == b"D":
            # DISARM — collect audio, send to parent, schedule keepalive.
            with armed_lock:
                armed = False
                streaming = False
            with audio_buffer_lock:
                chunks = list(audio_buffer)
                audio_buffer.clear()
            if chunks:
                audio = np.concatenate(chunks, axis=0)
                if audio.ndim > 1:
                    audio = audio.mean(axis=1)
                audio = audio.astype(np.float32, copy=False)
                payload = audio.tobytes()
            else:
                payload = b""
            _write_frame(stdout, b"A", payload)
            schedule_keepalive_close()
        elif cmd == b"P":
            # PING → PONG. This is what the parent uses for liveness checks;
            # an unresponsive child gets SIGKILL'd within seconds.
            _write_frame(stdout, b"O", b"")
        elif cmd == b"Q":
            break
        else:
            _send_error(stdout, f"unknown command: {cmd!r}")

    # Graceful shutdown — best effort, don't block on stuck CoreAudio.
    if keepalive_timer is not None:
        keepalive_timer.cancel()
    try:
        if stream is not None:
            stream.stop()
            stream.close()
    except Exception:
        pass


if __name__ == "__main__":
    run()
