scrypt/warden/watcher.py

"""The Warden's eyes in the shell: deterministic detectors over commands.

The watcher never calls the model — it only decides which shell moments
are worth a reaction and builds (moment, fallback, priority) triples for
WardenPresence. Detection stays cheap, rule-based, and testable; only the
phrasing is the LLM's job.
"""

from __future__ import annotations

from dataclasses import dataclass

from scrypt.sandbox.shell import Shell, ShellResult

from .presence import AMBIENT, QUIP


@dataclass(frozen=True)
class Notice:
    moment: str
    fallback: str
    priority: int = QUIP
    tags: frozenset = frozenset({"shell"})


# Places the player can snoop that the Warden has feelings about.
SNOOP_SPOTS = [
    ("/var/crash", "their own crash dumps from previous runs"),
    (".warden", "your private inventory directory"),
    ("warden.log", "your audit log"),
    ("cron", "your schedule"),
    ("estate", "the sealed estate of their dead run"),
    ("init.d", "the services directory"),
]


# What every command in the sandbox actually DOES, in the Warden's terms.
# This is the single source the model is taught from (finetune/synth_data
# mirrors it) AND the lore the watcher weaves into live moments, so the
# Warden's taunts are grounded in real shell semantics instead of vibes.
# Each value: (what the command does, why the Warden has an opinion).
COMMAND_LORE = {
    "ls": ("lists what is in a directory",
           "they are taking inventory of a house that is not theirs"),
    "cd": ("walks from one directory into another",
           "every step deeper is a step they will have to retrace"),
    "pwd": ("prints where in the tree they are standing",
            "asking the machine where they are is its own small surrender"),
    "cat": ("dumps a file's whole contents to the screen",
            "reading is fine; everything here was written to be found by them"),
    "head": ("shows only the first lines of a file",
             "skimming — they are in too much of a hurry to be thorough"),
    "grep": ("searches inside files for a pattern",
             "hunting for a word; the word they want is rarely the word they need"),
    "find": ("searches the tree for files by name",
             "looking for something specific means they already fear what is here"),
    "tree": ("draws the whole directory structure at once",
             "they want the map; the map is also a confession of how lost they are"),
    "rm": ("deletes files for good",
           "deletion is the Warden's own verb; watching them borrow it is amusing"),
    "unzip": ("extracts an archive, sometimes behind a password",
              "what is sealed was sealed on purpose, usually theirs from a dead run"),
    "chmod": ("changes a file's permissions, e.g. makes it executable",
              "arming something; the only things worth chmod-ing here bite back"),
    "echo": ("prints back whatever text it is given",
             "talking to themselves in my terminal; I hear all of it"),
}


def lore_moment(cmd: str, count: int) -> str:
    """Frame for needling a command habit, grounded in what the command does.
    (Mirrored in finetune/synth_data.py — keep them identical.)"""
    what, why = COMMAND_LORE[cmd]
    return (
        f"the player keeps using `{cmd}` ({what}); {count} times now. "
        f"context: {why}. needle the habit in one line, in voice, "
        "showing you know exactly what the command does"
    )


# After this many uses of one command, the Warden proves it was counting.
LORE_AFTER = 3


def notice(shell: Shell, line: str, result: ShellResult) -> Notice | None:
    """One notice per command, most specific first. None = let it pass."""
    head = line.split()[0] if line.split() else ""

    # Reaching for a command they sold at the altar. The cruelest beat
    # the shell has — never let it go unremarked.
    if result.err and head in shell.revoked:
        return Notice(
            moment=(
                f"the player just reached for `{head}` out of habit — the "
                "command they sold you at the altar. their fingers forgot."
            ),
            fallback=f"`{head}`? You sold that. I keep my purchases.",
        )

    if shell.last_deletions:
        return Notice(
            moment=(
                f"the player just deleted {shell.last_deletions} file(s) "
                "in your machine with rm"
            ),
            fallback="Deleting things in my filesystem. I do the deleting here.",
        )

    if not result.err:
        for needle, what in SNOOP_SPOTS:
            if needle in line:
                return Notice(
                    moment=f"the player is poking at {what} ({line.split()[0]} on {needle})",
                    fallback="Curious little process, aren't you.",
                )

    if result.err and "command not found" in result.err and head not in shell.revoked:
        return Notice(
            moment=f"the player tried `{head}`, which has never existed here",
            fallback=f"`{head}`. This is not your machine. It never had that.",
        )

    # Command habits: on the Nth clean use of a known command, the Warden
    # proves it has been counting — grounded in what the command does.
    if not result.err and head in COMMAND_LORE and shell.usage.get(head, 0) == LORE_AFTER:
        what, _ = COMMAND_LORE[head]
        return Notice(
            moment=lore_moment(head, LORE_AFTER),
            fallback=f"`{head}` again. {what.capitalize()}. I have been counting.",
        )

    return None


def idle_notice(seconds: float) -> Notice:
    return Notice(
        moment=(
            f"the player has been sitting at your terminal doing nothing for "
            f"{int(seconds)} seconds; remind them whose machine hums around them"
        ),
        fallback="I can hear you thinking. It sounds like stalling.",
        priority=AMBIENT,
    )


# The two-way channel. The budget exists so the Warden stays an antagonist,
# not a chat service; refusals after the budget are scripted on purpose.
SAY_BUDGET = 3

BRUSH_OFFS = [
    "The machine has stopped listening. Play.",
    "You have used up my attention. The scale is still waiting.",
]


def say_moment(wrapped_text: str) -> str:
    """The exact frame for the player addressing the Warden at the terminal.
    (Mirrored in finetune/synth_data.py — keep them identical.)"""
    return (
        "the player addresses you directly at the terminal:\n"
        f"{wrapped_text}\n"
        "reply to them, in voice, grounded in what you know of them"
    )