STT-Assistant/tests/test_assistant_speaker.py

"""Tests for Phase E: assistant speaker cache, display resolution, naming, and ctl."""

import itertools
import json
import socket
from pathlib import Path

from stt_daemon import events
from stt_daemon.assistant.control import _Client, ControlServer
from stt_daemon.assistant.service import AssistantService
from stt_daemon.config import Config
from stt_daemon.ctl import COMMAND_MAP, build_message

_sequence = itertools.count(1000)


class Clock:
    def __init__(self, start=1000.0):
        self.t = float(start)

    def __call__(self):
        return self.t


class FakeSubscriber:
    def events(self):
        return iter(())

    def close(self):
        pass


class FakeAudit:
    def __init__(self):
        self.records: list[dict] = []

    def record(self, **fields):
        self.records.append(fields)


def _service(tmp_path, audit=None):
    cfg = Config()
    cfg.speaker.store_dir = str(tmp_path / "speakers")
    clock = Clock()
    relayed: list[dict] = []
    service = AssistantService(
        cfg,
        FakeSubscriber(),
        time_fn=clock,
        audit=audit,
        relay=relayed.append,
    )
    return service, relayed, clock


def _speakers_event(clock, speakers):
    return {
        "v": 1,
        "type": events.SPEAKERS,
        "seq": next(_sequence),
        "ts": clock(),
        "speakers": speakers,
    }


def _transcript(clock, text, **fields):
    event = {
        "v": 1,
        "type": events.TRANSCRIPT,
        "seq": next(_sequence),
        "ts": clock(),
        "text": text,
    }
    event.update(fields)
    return event


_ROSTER = [
    {
        "id": "spk_0002",
        "name": "",
        "status": "provisional",
        "cumulative_s": 3.0,
        "sample_path": "speakers/samples/spk_0002.wav",
        "created_at": "2026-09-28T10:00:00",
    },
    {
        "id": "spk_0001",
        "name": "Efren",
        "status": "confirmed",
        "cumulative_s": 20.0,
        "sample_path": "",
        "created_at": "2026-09-28T09:00:00",
    },
]


def _cached(service, clock):
    service.handle_event(_speakers_event(clock, _ROSTER))


def test_speakers_event_caches_roster_sorted_by_id(tmp_path):
    service, _, clock = _service(tmp_path)
    assert service.speaker_list() == []

    _cached(service, clock)

    roster = service.speaker_list()
    assert [entry["id"] for entry in roster] == ["spk_0001", "spk_0002"]
    assert roster[0]["name"] == "Efren"


def test_speakers_event_is_relayed_to_viewers(tmp_path):
    service, relayed, clock = _service(tmp_path)
    event = _speakers_event(clock, _ROSTER)
    service.handle_event(event)

    assert len(relayed) == 1
    assert relayed[0]["type"] == events.SPEAKERS


def test_transcript_matched_named_relays_name(tmp_path):
    service, relayed, clock = _service(tmp_path)
    event = _transcript(
        clock,
        "hello",
        seg_id=1,
        speaker_id="spk_0001",
        speaker_status="matched",
        speaker="Efren",
        speaker_score=0.8,
    )
    service.handle_event(event)

    assert relayed[0]["speaker"] == "Efren"
    assert relayed[0]["speaker_id"] == "spk_0001"
    assert relayed[0]["speaker_status"] == "matched"
    assert relayed[0]["speaker_score"] == 0.8
    assert event["speaker"] == "Efren"


def test_transcript_unmatched_cases_relay_unknown(tmp_path):
    cases = [
        {"speaker_id": "spk_0002", "speaker_status": "matched", "speaker": "spk_0002"},
        {"speaker_id": "spk_0002", "speaker_status": "uncertain", "speaker": "spk_0002?"},
        {"speaker_id": None, "speaker_status": "unknown", "speaker": "?"},
        {"speaker_id": None, "speaker_status": "none", "speaker": "?"},
    ]
    for fields in cases:
        service, relayed, clock = _service(tmp_path)
        event = _transcript(clock, "hello", seg_id=1, speaker_score=None, **fields)
        service.handle_event(event)

        assert relayed[0]["speaker"] == "Unknown"
        assert relayed[0]["speaker_id"] == fields["speaker_id"]
        assert relayed[0]["speaker_status"] == fields["speaker_status"]


def test_speaker_name_known_id_writes_inbox_request(tmp_path):
    audit = FakeAudit()
    service, _, clock = _service(tmp_path, audit=audit)
    _cached(service, clock)

    assert service.speaker_name("spk_0002", "Ana") is True

    inbox = Path(service.cfg.speaker.store_dir) / "inbox"
    files = list(inbox.glob("*.json"))
    assert len(files) == 1
    assert json.loads(files[0].read_text(encoding="utf-8")) == {
        "speaker_id": "spk_0002",
        "name": "Ana",
    }
    assert audit.records == [
        {"mode": "speaker", "action": "name", "speaker_id": "spk_0002", "name": "Ana"}
    ]


def test_speaker_name_unknown_id_writes_nothing(tmp_path):
    audit = FakeAudit()
    service, _, clock = _service(tmp_path, audit=audit)
    _cached(service, clock)

    assert service.speaker_name("spk_9999", "Nobody") is False

    inbox = Path(service.cfg.speaker.store_dir) / "inbox"
    assert not inbox.exists() or list(inbox.glob("*.json")) == []
    assert audit.records == []


def test_speaker_name_inbox_is_private(tmp_path):
    service, _, clock = _service(tmp_path)
    _cached(service, clock)

    assert service.speaker_name("spk_0002", "Ana") is True

    inbox = Path(service.cfg.speaker.store_dir) / "inbox"
    assert (inbox.stat().st_mode & 0o077) == 0


def test_chat_context_uses_display_labels(tmp_path):
    service, _, clock = _service(tmp_path)
    service.toggle_chat()
    service.handle_event(
        _transcript(
            clock, "hi", speaker_id="spk_0001", speaker_status="matched", speaker="Efren"
        )
    )
    service.handle_event(
        _transcript(clock, "yo", speaker_id=None, speaker_status="none", speaker="?")
    )

    assert service.chat_history == ["Efren: hi", "Unknown: yo"]
    assert service._pending_chat == ["Efren: hi", "Unknown: yo"]
    assert service.triage_history == ["hi", "yo"]


def _dispatch(service, message):
    cfg = Config()
    server = ControlServer(cfg, service)
    left, right = socket.socketpair()
    try:
        client = _Client(left, 256)
        server._dispatch(client, message)
        return json.loads(bytes(client.queue.popleft()).decode("utf-8"))
    finally:
        left.close()
        right.close()


def test_control_speaker_list_reply_shape(tmp_path):
    service, _, clock = _service(tmp_path)
    _cached(service, clock)

    reply = _dispatch(service, {"type": "speaker_list"})

    assert reply["type"] == "speaker_list"
    assert [entry["id"] for entry in reply["speakers"]] == ["spk_0001", "spk_0002"]


def test_control_speaker_name_ack(tmp_path):
    service, _, clock = _service(tmp_path)
    _cached(service, clock)

    reply = _dispatch(
        service, {"type": "speaker_name", "speaker_id": "spk_0001", "name": "Efren"}
    )

    assert reply["type"] == "ack"
    assert reply["command"] == "speaker_name"


def test_control_speaker_name_unknown_id(tmp_path):
    service, _, clock = _service(tmp_path)
    _cached(service, clock)

    reply = _dispatch(
        service, {"type": "speaker_name", "speaker_id": "spk_9999", "name": "Nobody"}
    )

    assert reply["type"] == "error"
    assert reply["code"] == "unknown_speaker"


def test_control_speaker_name_empty_name(tmp_path):
    service, _, clock = _service(tmp_path)
    _cached(service, clock)

    reply = _dispatch(
        service, {"type": "speaker_name", "speaker_id": "spk_0001", "name": ""}
    )

    assert reply["type"] == "error"
    assert reply["code"] == "bad_message"


def test_ctl_command_map_and_build_message():
    assert COMMAND_MAP["speaker-list"] == "speaker_list"
    assert COMMAND_MAP["speaker-name"] == "speaker_name"
    assert build_message("speaker-list") == {"type": "speaker_list"}
    assert build_message("speaker-name", rest=["spk_0001", "Efren"]) == {
        "type": "speaker_name",
        "speaker_id": "spk_0001",
        "name": "Efren",
    }