STT-Assistant/tests/test_debug.py

"""Tests for stt_daemon.debug.DebugSink (per-file fakes, no conftest.py)."""

import json
import logging
import os
import threading
from datetime import datetime
from pathlib import Path

import pytest

from stt_daemon.debug import DebugSink


def _read_lines(path):
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]


def test_emit_writes_valid_jsonl_per_day(tmp_path):
    sink = DebugSink(str(tmp_path / "debug"), "daemon")
    sink.emit("heartbeat", queue_depth=0, device_ok=True)
    sink.emit("capture_reopen", attempt=1)

    day = datetime.now().strftime("%Y-%m-%d")
    records = _read_lines(tmp_path / "debug" / f"{day}.jsonl")
    assert len(records) == 2
    assert records[0]["proc"] == "daemon"
    assert records[0]["type"] == "heartbeat"
    assert records[0]["queue_depth"] == 0
    assert records[0]["device_ok"] is True
    assert "ts" in records[0]
    assert records[1]["type"] == "capture_reopen"


def test_forbidden_fields_never_persisted(tmp_path):
    sink = DebugSink(str(tmp_path / "debug"), "assistant")
    sink.emit("heartbeat", text="secret transcript", audio="clip.wav", prompt="sys", rms=0.1)

    day = datetime.now().strftime("%Y-%m-%d")
    records = _read_lines(tmp_path / "debug" / f"{day}.jsonl")
    assert len(records) == 1
    assert "text" not in records[0]
    assert "audio" not in records[0]
    assert "prompt" not in records[0]
    assert records[0]["rms"] == 0.1


def test_invalid_proc_rejected(tmp_path):
    with pytest.raises(ValueError):
        DebugSink(str(tmp_path), "worker")


@pytest.mark.parametrize("proc", ["daemon", "assistant", "supervisor"])
def test_all_procs_accepted(tmp_path, proc):
    sink = DebugSink(str(tmp_path / proc), proc)
    sink.emit("heartbeat")
    day = datetime.now().strftime("%Y-%m-%d")
    records = _read_lines(tmp_path / proc / f"{day}.jsonl")
    assert records[0]["proc"] == proc


def test_disabled_path_creates_no_files(tmp_path):
    # Disabled path = caller holds None instead of a sink (zero overhead).
    debug = None
    target = tmp_path / "debug"
    if debug is not None:
        debug.emit("heartbeat")
    assert not target.exists()


def test_heartbeat_gating_respected_by_caller_contract(tmp_path):
    # Gating lives with the caller: heartbeat lines are skipped unless enabled,
    # anomaly lines always flow.
    class FakeCaller:
        def __init__(self, sink, heartbeat):
            self.sink = sink
            self.heartbeat = heartbeat

        def tick(self, is_heartbeat, type, **fields):
            if is_heartbeat and not self.heartbeat:
                return
            self.sink.emit(type, **fields)

    sink = DebugSink(str(tmp_path / "debug"), "daemon")
    caller = FakeCaller(sink, heartbeat=False)
    caller.tick(True, "heartbeat", queue_depth=0)
    caller.tick(False, "capture_reopen", attempt=1)

    day = datetime.now().strftime("%Y-%m-%d")
    records = _read_lines(tmp_path / "debug" / f"{day}.jsonl")
    assert [r["type"] for r in records] == ["capture_reopen"]

    caller.heartbeat = True
    caller.tick(True, "heartbeat", queue_depth=0)
    records = _read_lines(tmp_path / "debug" / f"{day}.jsonl")
    assert [r["type"] for r in records] == ["capture_reopen", "heartbeat"]


def test_mkdir_failure_disables_without_raising(tmp_path, caplog):
    blocker = tmp_path / "blocker"
    blocker.write_text("not a dir")
    with caplog.at_level(logging.WARNING):
        sink = DebugSink(str(blocker / "debug"), "daemon")
    assert sink._disabled
    assert "mkdir failed" in caplog.text
    sink.emit("heartbeat")  # no-op, must not raise


def test_open_failure_warns_not_raises_and_breaker_latches(tmp_path, caplog, monkeypatch):
    sink = DebugSink(str(tmp_path / "debug"), "daemon")

    def _boom(path, flags, mode=0o644):
        raise OSError("disk gone")

    monkeypatch.setattr(os, "open", _boom)
    with caplog.at_level(logging.WARNING):
        sink.emit("heartbeat", queue_depth=1)
    assert "write failure" in caplog.text

    # Breaker latched: even with os.open restored, emits are silent no-ops.
    monkeypatch.undo()
    caplog.clear()
    with caplog.at_level(logging.WARNING):
        sink.emit("heartbeat", queue_depth=2)
    assert caplog.text == ""
    day = datetime.now().strftime("%Y-%m-%d")
    assert not (tmp_path / "debug" / f"{day}.jsonl").exists()


def test_write_failure_warns_once_then_latches(tmp_path, caplog, monkeypatch):
    sink = DebugSink(str(tmp_path / "debug"), "daemon")
    sink.emit("heartbeat", seq=0)

    calls = {"n": 0}
    real_write = os.write

    def _flaky(fd, data):
        calls["n"] += 1
        if calls["n"] == 1:
            raise OSError("transient")
        return real_write(fd, data)

    monkeypatch.setattr(os, "write", _flaky)
    with caplog.at_level(logging.WARNING):
        sink.emit("heartbeat", seq=1)
    assert "write failure" in caplog.text

    monkeypatch.undo()
    caplog.clear()
    with caplog.at_level(logging.WARNING):
        sink.emit("heartbeat", seq=2)
    assert caplog.text == ""

    day = datetime.now().strftime("%Y-%m-%d")
    records = _read_lines(tmp_path / "debug" / f"{day}.jsonl")
    assert [r.get("seq") for r in records] == [0]


def test_concurrent_emits_produce_no_partial_lines(tmp_path):
    sink = DebugSink(str(tmp_path / "debug"), "daemon")
    threads = 8
    per_thread = 100

    def _work(tid):
        for i in range(per_thread):
            sink.emit("heartbeat", tid=tid, seq=i, pad="x" * 200)

    workers = [threading.Thread(target=_work, args=(t,)) for t in range(threads)]
    for w in workers:
        w.start()
    for w in workers:
        w.join()

    day = datetime.now().strftime("%Y-%m-%d")
    path = tmp_path / "debug" / f"{day}.jsonl"
    with open(path, encoding="utf-8") as f:
        raw = f.readlines()
    assert len(raw) == threads * per_thread
    seen = set()
    for line in raw:
        record = json.loads(line)  # raises on any partial/interleaved line
        seen.add((record["tid"], record["seq"]))
    assert len(seen) == threads * per_thread


def test_cap_skips_past_limit_with_single_warning(tmp_path, caplog):
    sink = DebugSink(str(tmp_path / "debug"), "daemon", max_bytes_per_day=200)
    with caplog.at_level(logging.WARNING):
        for i in range(20):
            sink.emit("heartbeat", seq=i, pad="y" * 50)
    warnings = [r for r in caplog.records if "cap reached" in r.message]
    assert len(warnings) == 1

    day = datetime.now().strftime("%Y-%m-%d")
    size = Path(tmp_path / "debug" / f"{day}.jsonl").stat().st_size
    assert size <= 200


def test_cap_counter_resets_on_rollover(tmp_path):
    clock = {"now": datetime(2026, 9, 29, 23, 59, 59)}
    sink = DebugSink(str(tmp_path / "debug"), "daemon", max_bytes_per_day=200,
                     now_fn=lambda: clock["now"])
    for i in range(20):  # exhaust day one's cap
        sink.emit("heartbeat", seq=i, pad="z" * 50)

    clock["now"] = datetime(2026, 9, 30, 0, 0, 1)
    sink.emit("heartbeat", seq="next-day")
    records = _read_lines(tmp_path / "debug" / "2026-09-30.jsonl")
    assert [r.get("seq") for r in records] == ["next-day"]


def test_close_is_noop(tmp_path):
    sink = DebugSink(str(tmp_path / "debug"), "supervisor")
    sink.emit("spawn", pid=123)
    sink.close()
    sink.emit("exit", code=0)
    day = datetime.now().strftime("%Y-%m-%d")
    records = _read_lines(tmp_path / "debug" / f"{day}.jsonl")
    assert [r["type"] for r in records] == ["spawn", "exit"]