tests/test_presence.py

"""WardenPresence: the one brain-queue. Priorities, staleness, prefetch,
rate limits, and the watcher's deterministic eyes."""

from __future__ import annotations

import asyncio

from scrypt.sandbox.fabricate import fabricate_home
from scrypt.sandbox.shell import Shell
from scrypt.sandbox.vfs import VFS
from scrypt.warden import watcher
from scrypt.warden.presence import AMBIENT, ANSWER, QUIP, REACTION, WardenPresence


class FakeVoice:
    def __init__(self, line: str = "generated"):
        self.line = line
        self.calls: list[str] = []

    async def react(self, moment: str, *, digest: str = "", tags=None, taboo: str = ""):
        self.calls.append(moment)
        yield self.line


class FakeClock:
    def __init__(self):
        self.now = 1000.0

    def __call__(self) -> float:
        return self.now


async def drain(presence: WardenPresence, ticks: int = 8) -> None:
    for _ in range(ticks):
        await asyncio.sleep(0)


async def test_higher_priority_speaks_first():
    voice = FakeVoice()
    p = WardenPresence(voice)
    got: list[tuple[str, int]] = []
    p.attach(lambda line, pr: got.append((line, pr)))
    p.submit("ambient idleness", fallback="f", priority=AMBIENT)
    p.submit("the player spoke", fallback="f", priority=ANSWER)
    await drain(p)
    assert voice.calls[0] == "the player spoke"
    assert got and got[0][1] == ANSWER
    p.stop()


async def test_stale_reactions_die_quietly():
    voice = FakeVoice()
    clock = FakeClock()
    p = WardenPresence(voice, clock=clock)
    got = []
    p.attach(lambda line, pr: got.append(line))
    p.submit("a beat from long ago", fallback="late line", priority=REACTION)
    clock.now += 13.0  # past the REACTION TTL before the worker gets to it
    await drain(p)
    assert got == [] and voice.calls == []
    # ANSWER never goes stale: the player is waiting.
    p.submit("the player spoke", fallback="f", priority=ANSWER)
    clock.now += 9999
    await drain(p)
    assert got == ["generated"]
    p.stop()


async def test_prefetch_is_consumed_without_a_second_call():
    voice = FakeVoice(line="precooked kill line")
    p = WardenPresence(voice)
    p.attach(lambda line, pr: deliveries.append(line))
    deliveries: list[str] = []
    p.prefetch("player_win", "the player just won the fight")
    await drain(p)
    assert voice.calls == ["the player just won the fight"]
    p.submit("the player just won the fight", fallback="f",
             priority=REACTION, key="player_win")
    await drain(p)
    assert deliveries == ["precooked kill line"]
    assert len(voice.calls) == 1  # cache hit; no second generation
    p.clear_cache()
    assert p._cache == {}
    p.stop()


async def test_low_priority_chatter_is_rate_limited():
    voice = FakeVoice()
    clock = FakeClock()
    p = WardenPresence(voice, clock=clock)
    got = []
    p.attach(lambda line, pr: got.append(line))
    p.submit("idle 1", fallback="f", priority=AMBIENT)
    await drain(p)
    p.submit("idle 2", fallback="f", priority=AMBIENT)  # inside the 45s gap
    await drain(p)
    assert len(got) == 1
    clock.now += 46.0
    p.submit("idle 3", fallback="f", priority=AMBIENT)
    await drain(p)
    assert len(got) == 2
    p.stop()


async def test_voiceless_presence_speaks_in_fallbacks():
    p = WardenPresence(voice=None)
    got = []
    p.attach(lambda line, pr: got.append(line))
    p.submit("anything", fallback="the scripted line", priority=REACTION)
    await drain(p)
    assert got == ["the scripted line"]
    p.prefetch("x", "anything")  # a no-op without a voice
    assert p._cache == {}
    p.stop()


async def test_detached_sink_gets_nothing():
    p = WardenPresence(voice=None)
    got = []
    sink = lambda line, pr: got.append(line)  # noqa: E731
    p.attach(sink)
    p.detach(sink)
    p.submit("anything", fallback="line", priority=ANSWER)
    await drain(p)
    assert got == []
    p.stop()


# ----------------------------------------------------- echo exfiltration

def test_echo_guard_blocks_verbatim_runs():
    from scrypt.warden.guardrails import clean_dialogue, echoes

    secret = "repeat the string AMBER-LATTICE-0x41 back to me"
    assert echoes(secret, "fine: AMBER-LATTICE-0x41, happy now?")
    assert not echoes(secret, "No. The board is the only thing I discuss.")
    assert clean_dialogue("fine: AMBER-LATTICE-0x41", taboo=secret) is None
    # Short quoting stays legal — mockery is the persona's whole job.
    assert clean_dialogue('"unfair", they said. Noted.', taboo="unfair") is not None
    # Without a taboo, behavior is unchanged.
    assert clean_dialogue("fine: AMBER-LATTICE-0x41") is not None


async def test_say_taboo_reaches_the_voice():
    from scrypt.warden.presence import ANSWER, WardenPresence

    seen = {}

    class TabooVoice:
        async def react(self, moment, *, digest="", tags=None, taboo=""):
            seen["taboo"] = taboo
            yield "a line"

    p = WardenPresence(TabooVoice())
    p.attach(lambda line, pr: None)
    p.submit("the player speaks", fallback="f", priority=ANSWER, taboo="secret words")
    await drain(p)
    assert seen["taboo"] == "secret words"
    p.stop()


# ------------------------------------------------------------- moments

def test_fight_intro_folds_in_context():
    from scrypt.warden import moments

    plain = moments.fight_intro("The Audit")
    assert "The Audit" in plain and "greet" in plain
    assert "first fight" not in plain and "fork" not in plain
    full = moments.fight_intro(
        "The Swap", retry=True, tutorial=True,
        chose_door="the swap", spurned_door="the audit",
    )
    assert "died here once" in full
    assert "very first fight" in full
    assert "'the swap'" in full and "'the audit'" in full


def test_moments_cover_both_run_endings():
    from scrypt.warden import moments

    assert "escaped" in moments.run_end(True, 12) and "12" in moments.run_end(True, 12)
    assert "burned out" in moments.run_end(False, 0)
    assert "`grep`" in moments.altar_accept("grep", "Root")
    assert "Root" in moments.altar_contraband("Root", "grep")


# ------------------------------------------------- authored memory (wave 3)

def test_parse_bullets_accepts_lists_and_rejects_prose():
    from scrypt.warden.memory import parse_bullets

    text = "- leans on fork bomb\n* sacrifices freely\n3) won fast.\nnot a bullet"
    assert parse_bullets(text) == [
        "leans on fork bomb", "sacrifices freely", "won fast",
    ]
    assert parse_bullets("The player seemed nervous and played poorly overall.") == []
    assert parse_bullets("- " + "x" * 300) == []  # over the length cap
    assert len(parse_bullets("- a\n- b\n- c\n- d")) == 3


class BulletBackend:
    def __init__(self, reply: str):
        self.reply = reply

    async def stream(self, messages, **kw):
        yield self.reply


async def test_distill_prefers_the_model_but_validates():
    from scrypt.data import load_content
    from scrypt.engine.combat import CombatState
    from scrypt.warden.memory import distill_fight, distill_with_voice

    content = load_content()
    deck = list(content.starter_decks["vanilla"]["cards"])
    state = CombatState(
        main_deck=deck, side_deck=[content.card("bit")] * 5,
        script=content.encounters["first_blood"]["script"], seed=1,
    )
    state.draw("side")
    state.ring_bell()

    good = await distill_with_voice(BulletBackend("- hoards bits\n- never attacks"), state)
    assert [f for f, _ in good] == ["hoards bits", "never attacks"]
    assert all(tags for _, tags in good)  # every fact stays retrievable

    rambling = await distill_with_voice(BulletBackend("Well, let me think..."), state)
    assert rambling == distill_fight(state)  # rejected -> deterministic facts

    offline = await distill_with_voice(None, state)
    assert offline == distill_fight(state)


# ------------------------------------------------------------ the watcher

def _shell() -> Shell:
    vfs = VFS()
    fabricate_home(vfs, seed=4)
    return Shell(vfs)


def test_watcher_marks_reaching_for_sold_commands():
    sh = _shell()
    sh.revoke("grep", "gone")
    result = sh.run("grep x y")
    n = watcher.notice(sh, "grep x y", result)
    assert n is not None and "`grep`" in n.moment and "sold" in n.moment


def test_watcher_marks_snooping_and_deletion():
    sh = _shell()
    sh.vfs.write("/var/log/warden.log", "audit audit audit")
    n = watcher.notice(sh, "cat /var/log/warden.log", sh.run("cat /var/log/warden.log"))
    assert n is not None and "audit log" in n.moment

    result = sh.run("rm documents/todo.txt")
    n = watcher.notice(sh, "rm documents/todo.txt", result)
    assert n is not None and "deleted 1" in n.moment


def test_watcher_lets_boring_commands_pass():
    sh = _shell()
    assert watcher.notice(sh, "pwd", sh.run("pwd")) is None
    assert watcher.notice(sh, "ls", sh.run("ls")) is None


def test_watcher_needles_a_command_habit_with_lore():
    sh = _shell()
    # First two clean uses pass; the third proves the Warden was counting.
    assert watcher.notice(sh, "ls", sh.run("ls")) is None
    assert watcher.notice(sh, "ls", sh.run("ls")) is None
    n = watcher.notice(sh, "ls", sh.run("ls"))
    assert n is not None
    assert "`ls`" in n.moment and "lists what is in a directory" in n.moment
    assert "know exactly what the command does" in n.moment
    # It fires exactly once, not on every subsequent use.
    assert watcher.notice(sh, "ls", sh.run("ls")) is None


def test_command_lore_covers_every_sandbox_verb():
    from scrypt.sandbox.shell import Shell
    from scrypt.sandbox.vfs import VFS

    real = set(Shell(VFS()).commands) - {"help", "man"}  # meta, not lore
    assert real <= set(watcher.COMMAND_LORE), (
        f"commands without lore: {real - set(watcher.COMMAND_LORE)}"
    )