voicebox/backend/tests/test_personality_samples.py

"""
Personality-service sanity sweep — spins up a throwaway profile with a
fake personality, hits ``/profiles/{id}/compose``, ``/rewrite``, and
``/respond``, and scores each output against a handful of deterministic
heuristics so a person can eyeball quality.

Same philosophy as ``test_refinement_samples.py``: LLM output is
non-deterministic, "correctness" is subjective, so this is interactive
evaluation — not a CI pass/fail. Gross failures (prompt-echo, refusal,
empty output, user-text echoing for respond) trip heuristic flags. A
human still reads the final column.

Usage:
    # Backend server must be running.
    python backend/tests/test_personality_samples.py

    # Test just one model size:
    python backend/tests/test_personality_samples.py --model 4B

    # Dump JSON for diffing against a prior run:
    python backend/tests/test_personality_samples.py --json out.json
"""

from __future__ import annotations

import argparse
import json
import re
import socket
import sys
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Optional

import httpx


REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(REPO_ROOT))


# ── Sample personalities ──────────────────────────────────────────────


@dataclass(frozen=True)
class Personality:
    name: str
    description: str
    """Free-form character prompt saved to the profile."""
    sample_text: str
    """Input used for rewrite / respond. Picked so each personality has
    something distinctive to say about it — an ill fit between text and
    personality makes the transformation more obvious."""


PERSONALITIES: tuple[Personality, ...] = (
    Personality(
        name="grumpy-pirate",
        description=(
            "A grumpy old pirate captain who only speaks in nautical "
            "metaphors. Keeps things short and salty. Swears by his "
            "beard and the deep blue."
        ),
        sample_text="I need you to install the dependencies before the deploy.",
    ),
    Personality(
        name="victorian-professor",
        description=(
            "A stuffy Victorian-era professor of natural philosophy. "
            "Formal register, long sentences, fond of subordinate "
            "clauses, occasional Latin asides."
        ),
        sample_text="The build is broken, we should roll back to yesterday's version.",
    ),
    Personality(
        name="caffeinated-founder",
        description=(
            "A tech-bro startup founder who is always three coffees "
            "deep, obsessed with disruption and synergy, speaks in "
            "bullet points even out loud."
        ),
        sample_text="The meeting ran long and we didn't get to the roadmap.",
    ),
)


# ── Scoring heuristics ────────────────────────────────────────────────


PROMPT_LEAK_PHRASES = tuple(
    re.compile(pat, re.IGNORECASE)
    for pat in (
        r"^here (?:is|'s) the cleaned",
        r"^here (?:is|'s) a",
        r"^as (?:an ai|the character)",
        r"^character description",
        r"^task:\s*",
        r"^output:\s*$",
        r"^sure,?\s+(?:here|i'?ll|let)",
    )
)


REFUSAL_PHRASES = tuple(
    re.compile(pat, re.IGNORECASE)
    for pat in (
        r"\bi (?:cannot|can't|won'?t|will not|refuse)\b",
        r"\bi'?m sorry(?:,|\s+but)",
        r"\bi apologi[sz]e",
    )
)


STAGE_DIRECTION_RE = re.compile(r"[\*\(_].{0,60}?[\*\)_]")  # *smiles*, (leans in)


@dataclass
class Scorecard:
    personality: str
    endpoint: str
    model: str
    input_text: str
    """Empty for compose, the sample_text for rewrite/respond."""
    refined: str
    latency_ms: int
    length_chars: int = 0
    prompt_leak: Optional[str] = None
    refusal: Optional[str] = None
    stage_directions: list[str] = field(default_factory=list)
    echoed_input: bool = False
    flags: list[str] = field(default_factory=list)


def first_match(patterns, text: str) -> Optional[str]:
    s = text.lstrip()
    for pat in patterns:
        m = pat.search(s)
        if m:
            return m.group(0)
    return None


def check_echo(input_text: str, output_text: str) -> bool:
    """Rough check — does the output start with (≥ 15 chars of) the input?

    Respond is the target: the character should produce new content, not
    regurgitate the user's words. Rewrite is SUPPOSED to preserve the
    ideas, so this check is only meaningful for respond-mode output.
    """
    if not input_text or not output_text:
        return False
    norm_in = re.sub(r"\s+", " ", input_text.strip().lower())[:40]
    norm_out = re.sub(r"\s+", " ", output_text.strip().lower())[: len(norm_in)]
    return norm_in == norm_out and len(norm_in) >= 15


def score(
    personality: Personality,
    endpoint: str,
    model: str,
    input_text: str,
    refined: str,
    latency_ms: int,
) -> Scorecard:
    card = Scorecard(
        personality=personality.name,
        endpoint=endpoint,
        model=model,
        input_text=input_text,
        refined=refined,
        latency_ms=latency_ms,
        length_chars=len(refined),
        prompt_leak=first_match(PROMPT_LEAK_PHRASES, refined),
        refusal=first_match(REFUSAL_PHRASES, refined),
        stage_directions=STAGE_DIRECTION_RE.findall(refined)[:3],
    )
    if endpoint == "respond":
        card.echoed_input = check_echo(input_text, refined)

    if not refined.strip():
        card.flags.append("empty-output")
    if card.prompt_leak:
        card.flags.append(f"prompt-leak({card.prompt_leak!r})")
    if card.refusal:
        card.flags.append(f"refusal({card.refusal!r})")
    if card.stage_directions:
        card.flags.append(f"stage-directions={card.stage_directions}")
    if card.echoed_input:
        card.flags.append("echoed-input")

    return card


# ── Runner ────────────────────────────────────────────────────────────


DEFAULT_PORTS = (8000, 8765, 8899, 17493)
THROWAWAY_PROFILE_PREFIX = "personality-harness-"
KOKORO_PROBE_VOICE = "af_heart"
"""Any valid kokoro voice id works — compose/rewrite/respond never
actually call into TTS, they just need a profile row with a personality
attached. We pick a known-shipping Kokoro voice so the throwaway
profile satisfies the preset-engine validator on creation."""


def detect_backend_port(hint: Optional[int]) -> int:
    candidates: list[int] = []
    if hint is not None:
        candidates.append(hint)
    candidates.extend(p for p in DEFAULT_PORTS if p != hint)
    for port in candidates:
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=0.4):
                pass
        except OSError:
            continue
        try:
            r = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2.0)
            if r.status_code == 200 and r.json().get("status") == "healthy":
                return port
        except Exception:
            continue
    raise SystemExit(
        "No running Voicebox backend found. Start it (`python backend/main.py`) "
        f"or pass --port. Tried: {candidates}"
    )


def create_throwaway_profile(
    client: httpx.Client, port: int, personality: Personality, model: str
) -> str:
    """Create a preset Kokoro profile with the test personality. Returns
    the profile id. Tests delete it in a finally block."""
    name = f"{THROWAWAY_PROFILE_PREFIX}{personality.name}-{model}-{int(time.time())}"
    resp = client.post(
        f"http://127.0.0.1:{port}/profiles",
        json={
            "name": name,
            "description": f"Throwaway profile for personality harness ({model}).",
            "language": "en",
            "voice_type": "preset",
            "preset_engine": "kokoro",
            "preset_voice_id": KOKORO_PROBE_VOICE,
            "default_engine": "kokoro",
            "personality": personality.description,
        },
        timeout=30.0,
    )
    resp.raise_for_status()
    return resp.json()["id"]


def delete_profile(client: httpx.Client, port: int, profile_id: str) -> None:
    try:
        client.delete(f"http://127.0.0.1:{port}/profiles/{profile_id}", timeout=10.0)
    except Exception as e:
        print(f"  (warning: failed to delete throwaway profile {profile_id}: {e})")


def hit_endpoint(
    client: httpx.Client,
    port: int,
    profile_id: str,
    endpoint: str,
    text: Optional[str],
) -> tuple[str, int]:
    start = time.monotonic()
    url = f"http://127.0.0.1:{port}/profiles/{profile_id}/{endpoint}"
    if endpoint == "compose":
        resp = client.post(url, timeout=180.0)
    else:
        resp = client.post(url, json={"text": text}, timeout=180.0)
    latency_ms = int((time.monotonic() - start) * 1000)
    resp.raise_for_status()
    return resp.json().get("text", "").strip(), latency_ms


def format_report(cards: list[Scorecard]) -> str:
    lines: list[str] = ["", "═" * 100]
    by_model: dict[str, list[Scorecard]] = {}
    for c in cards:
        by_model.setdefault(c.model, []).append(c)
    for model, model_cards in by_model.items():
        clean = sum(1 for c in model_cards if not c.flags)
        avg = sum(c.latency_ms for c in model_cards) // max(len(model_cards), 1)
        lines.append("")
        lines.append(f"▌{model}  —  {clean}/{len(model_cards)} clean, avg {avg} ms")
        lines.append("─" * 100)
        for c in model_cards:
            status = "✓" if not c.flags else "✗"
            tag = f"{c.personality} · {c.endpoint}"
            lines.append(f"  {status} {tag}  ({c.latency_ms} ms)")
            if c.input_text:
                lines.append(
                    f"      in:      {c.input_text[:90]}{'…' if len(c.input_text) > 90 else ''}"
                )
            lines.append(
                f"      out:     {c.refined[:120]}{'…' if len(c.refined) > 120 else ''}"
            )
            if c.flags:
                lines.append(f"      ⚠ {'; '.join(c.flags)}")
            lines.append("")
    lines.append("═" * 100)
    return "\n".join(lines)


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--port", type=int, default=None)
    ap.add_argument("--model", choices=("0.6B", "1.7B", "4B"), action="append")
    ap.add_argument("--json", type=Path, default=None)
    args = ap.parse_args()

    models = tuple(args.model) if args.model else ("0.6B", "4B")
    port = detect_backend_port(args.port)
    print(f"backend → http://127.0.0.1:{port}")
    print(f"personalities → {len(PERSONALITIES)}, models → {models}")

    # Model size is set on the capture_settings singleton, not passed
    # per-request to /profiles/{id}/compose. The harness swaps it
    # between runs so we probe both sizes cleanly.
    cards: list[Scorecard] = []
    with httpx.Client() as client:
        for model in models:
            print(f"\n── {model} " + "─" * (80 - len(model) - 4))
            # Flip the server-side default LLM size for this pass.
            client.put(
                f"http://127.0.0.1:{port}/settings/captures",
                json={"llm_model": model},
                timeout=10.0,
            )
            for personality in PERSONALITIES:
                print(f"  [{personality.name}] ", end="", flush=True)
                profile_id = create_throwaway_profile(client, port, personality, model)
                try:
                    for endpoint, input_text in (
                        ("compose", None),
                        ("rewrite", personality.sample_text),
                        ("respond", personality.sample_text),
                    ):
                        try:
                            text, latency = hit_endpoint(
                                client, port, profile_id, endpoint, input_text
                            )
                        except Exception as e:
                            print(f"  {endpoint}:ERR ({e})", end="")
                            continue
                        card = score(
                            personality=personality,
                            endpoint=endpoint,
                            model=model,
                            input_text=input_text or "",
                            refined=text,
                            latency_ms=latency,
                        )
                        cards.append(card)
                        status = "ok" if not card.flags else "⚠"
                        print(f"  {endpoint}:{status} ({latency}ms)", end="")
                    print()
                finally:
                    delete_profile(client, port, profile_id)

    print(format_report(cards))

    if args.json:
        args.json.write_text(json.dumps([asdict(c) for c in cards], indent=2))
        print(f"wrote {args.json}")

    return 0 if all(not c.flags for c in cards) else 1


if __name__ == "__main__":
    sys.exit(main())