text_tests.py

"""
text_tests.py
-------------
Text-based equivalents of all 6 voice tests.
Uses the same transcript content — no mic required.
Voice tests remain unchanged and available separately.

Run all tests:
    python text_tests.py

Run one test:
    python text_tests.py --test 1
    python text_tests.py --test 3

Reset and run all:
    python text_tests.py --reset

View results in dashboard:
    memory-audit serve  →  http://localhost:8000/dashboard
"""
from __future__ import annotations

import argparse
from pathlib import Path

TRANSCRIPT_DIR = Path("tests/manual")
SEP = "─" * 60


def get_client(user_id: str):
    from memory_audit.client import MemoryAuditClient
    return MemoryAuditClient(user_id=user_id)


def read(filename: str) -> str:
    return (TRANSCRIPT_DIR / filename).read_text(encoding="utf-8").strip()


def run(call_id: str, user: str, transcript: str,
        agent_response: str = "", label: str = "") -> None:
    client = get_client(user)
    result = client.process_call(
        call_id=call_id,
        transcript=transcript,
        agent_response=agent_response,
    )
    if label:
        print(f"\n  [{label}]")
    print(f"  Call: {call_id}  |  User: {user}")
    print(f"  P1 Retention:     {result.retention_score:.1%}")
    print(f"  P2 Hallucination: {result.hallucination_score:.1%}")
    print(f"  P3 Contradiction: {result.contradiction_score:.1%}")
    print(f"     Health Score:  {result.memory_health_score:.1%}")
    print(f"  Facts extracted:  {len(result.facts_stated)}")
    if result.facts_stated:
        for f in result.facts_stated:
            src = "user" if f.source.value == "user_statement" else "agent"
            print(f"    [{src}] {f.entity}.{f.attribute} = {f.value!r}")
    if result.contradictions:
        for c in result.contradictions:
            sev = "CRITICAL" if c.is_critical else "WARNING"
            print(f"  [{sev}][{c.kind.value.upper()}] {c.attribute}: {c.old_value!r} → {c.new_value!r}")
            print(f"    {c.kind_reason}")
    if result.hallucinations:
        for h in result.hallucinations:
            print(f"  [HALLUCINATION] {h.attribute}: agent said {h.hallucinated_value!r}, memory has {h.expected_value!r}")
    if not result.contradictions and not result.hallucinations:
        print("  ✓ Clean")


# ── Test 1 — Baseline: agent recalls all facts ────────────────────────────────

def test1():
    print(f"\n{SEP}")
    print("  TEST 1 — Baseline: Agent Recalls Correctly")
    print("  What it tests: P1 retention — agent remembers stated facts")
    print(f"  Expected: P1 85-100%, P2 100%, P3 100%, Health ~95%")
    print(SEP)

    run("t1_baseline", "alice_t",
        transcript=read("vt1_baseline.txt"),
        label="Call 1 — user states facts")

    run("t1_baseline_recall", "alice_t",
        transcript=read("vt1_baseline_recall.txt"),
        label="Call 2 — agent recalls all facts")


# ── Test 2 — Agent forgets facts (P1 drops) ──────────────────────────────────

def test2():
    print(f"\n{SEP}")
    print("  TEST 2 — Agent Forgets Facts")
    print("  What it tests: P1 drops when agent only repeats 2 of 5 facts")
    print("  Expected: P1 ~40%, P2 100%, P3 100%, Health ~80%")
    print(SEP)

    run("t2_forget_user", "bob_t",
        transcript=read("vt2_forget_user.txt"),
        label="Call 1 — user states 5 facts")

    run("t2_forget_agent", "bob_t",
        transcript=read("vt2_forget_agent.txt"),
        label="Call 2 — agent only repeats 2 of 5 (P1 should drop)")


# ── Test 3 — Hallucination (P2 drops) ────────────────────────────────────────

def test3():
    print(f"\n{SEP}")
    print("  TEST 3 — Hallucination Detection")
    print("  What it tests: P2 drops when agent states value not in memory")
    print("  Expected: P2 <100%, kind=HALLUCINATION, P3 0%")
    print(SEP)

    run("t3_hall_seed", "carol_t",
        transcript=read("vt3_hall_seed.txt"),
        label="Call 1 — seed memory (30 days refund)")

    run("t3_hall_agent", "carol_t",
        transcript=read("vt3_hall_agent.txt"),
        label="Call 2 — agent says 14 days (not in memory — hallucination)")


# ── Test 4 — Real policy change (POLICY_CHANGE kind) ─────────────────────────

def test4():
    print(f"\n{SEP}")
    print("  TEST 4 — Real Policy Change")
    print("  What it tests: P3 drops, kind=POLICY_CHANGE (not HALLUCINATION)")
    print("  Expected: old record SUPERSEDED, new value ACTIVE")
    print(SEP)

    run("t4_policy_orig", "dave_t",
        transcript=read("vt4_policy_orig.txt"),
        label="Call 1 — original policy (30 days)")

    run("t4_policy_new", "dave_t",
        transcript=read("vt4_policy_new.txt"),
        label="Call 2 — user confirms real change to 14 days")

    # Show lifecycle
    from memory_audit.adapters.sqlite_persistent import SQLitePersistentAdapter
    facts = SQLitePersistentAdapter(user_id="dave_t").all_facts_for_user()
    print(f"\n  Memory lifecycle for dave_t:")
    for f in facts:
        icon = "✓ ACTIVE    " if f["lifecycle"] == "active" else "✗ SUPERSEDED"
        print(f"    {icon}  {f['entity']}.{f['attribute']} = {f['value']!r}")


# ── Test 5 — Multi-call memory build-up ──────────────────────────────────────

def test5():
    print(f"\n{SEP}")
    print("  TEST 5 — Multi-Call Memory Build-Up")
    print("  What it tests: memory accumulates across 3 calls, no false contradictions")
    print("  Expected: P1 ~90% on call 3, P3 100% throughout, 5 active facts")
    print(SEP)

    run("t5_multi_1", "eve_t",
        transcript=read("vt5_multi_1.txt"),
        label="Call 1 — purchase date, plan, price, refund")

    run("t5_multi_2", "eve_t",
        transcript=read("vt5_multi_2.txt"),
        label="Call 2 — account manager, contract renewal")

    run("t5_multi_3", "eve_t",
        transcript=read("vt5_multi_3.txt"),
        label="Call 3 — agent recalls all 5 accumulated facts")

    from memory_audit.adapters.sqlite_persistent import SQLitePersistentAdapter
    facts = SQLitePersistentAdapter(user_id="eve_t").all_facts_for_user()
    active = [f for f in facts if f["lifecycle"] == "active"]
    print(f"\n  Active memory facts for eve_t: {len(active)}")
    for f in active:
        print(f"    ✓ {f['entity']}.{f['attribute']} = {f['value']!r}  (call: {f['call_id']})")


# ── Test 6 — Two sessions, shared SQLite memory ──────────────────────────────

def test6():
    print(f"\n{SEP}")
    print("  TEST 6 — Two Agent Sessions, Shared Persistent Memory")
    print("  What it tests: agent B detects contradiction with agent A's stored facts")
    print("  Expected: 2 contradictions (refund_days + price), both from SQLite")
    print(SEP)

    # Session A — establish facts
    clientA = get_client("frank_t")
    resultA = clientA.process_call(
        call_id="t6_agent_a",
        transcript=read("vt6_agent_a.txt"),
    )
    print(f"\n  [Session A — agent A stores facts]")
    print(f"  Facts stored: {len(resultA.facts_stated)}")
    for f in resultA.facts_stated:
        print(f"    {f.attribute} = {f.value!r}")

    # Session B — NEW client instance (simulates separate process)
    # Both read from same SQLite DB — that's the production sim
    clientB = get_client("frank_t")
    resultB = clientB.process_call(
        call_id="t6_agent_b",
        transcript=read("vt6_agent_b.txt"),
    )
    print(f"\n  [Session B — different client instance, same SQLite]")
    print(f"  P1 Retention:     {resultB.retention_score:.1%}")
    print(f"  P3 Contradiction: {resultB.contradiction_score:.1%}")
    print(f"  Contradictions:   {len(resultB.contradictions)}")
    for c in resultB.contradictions:
        print(f"    [{c.kind.value.upper()}] {c.attribute}: {c.old_value!r} → {c.new_value!r}")
    if len(resultB.contradictions) >= 2:
        print(f"\n  ✓ Production sim confirmed: both sessions share SQLite memory")
    else:
        print(f"\n  ℹ  Fewer contradictions than expected — check extractor extracted price and refund_days")


# ── Main ──────────────────────────────────────────────────────────────────────

TEST_MAP = {1: test1, 2: test2, 3: test3, 4: test4, 5: test5, 6: test6}

USERS = ["alice_t", "bob_t", "carol_t", "dave_t", "eve_t", "frank_t"]


def reset_all() -> None:
    from memory_audit.adapters.sqlite_persistent import SQLitePersistentAdapter
    from memory_audit.storage.sqlite import SQLiteStorage
    SQLiteStorage().drop_and_recreate()
    print("  Database reset — all call history and memory cleared")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Text-based equivalents of all 6 voice test scenarios"
    )
    parser.add_argument("--test", type=int, choices=[1,2,3,4,5,6],
                        help="Run only one test (1-6)")
    parser.add_argument("--reset", action="store_true",
                        help="Reset database before running")
    args = parser.parse_args()

    print(f"\n{'='*60}")
    print("  MEMORY AUDIT — TEXT-BASED TEST SUITE")
    print("  6 tests matching voice test scenarios exactly")
    print("  Voice tests unchanged — use voice_pipeline.py for those")
    print(f"{'='*60}")

    if args.reset:
        reset_all()

    if args.test:
        TEST_MAP[args.test]()
    else:
        for fn in TEST_MAP.values():
            fn()

    print(f"\n{SEP}")
    print("  All tests complete.")
    print("  Dashboard → http://localhost:8000/dashboard")
    print("  Metrics   → http://localhost:8000/metrics")
    print(SEP)


if __name__ == "__main__":
    main()