#!/usr/bin/env python3
"""check-read-size.py  -  decision core for the read-size PreToolUse gate.

Reads a Claude Code hook payload on stdin and prints ONE line:

    PASS
    OBSERVE\t<path>\t<lines>\t<why>
    BLOCK\t<path>\t<lines>\t<why>

It decides only. The wrapper (check-read-size.sh) owns telemetry, the message
the model sees, and the exit code. Splitting them is what lets the smoke gate
drive the decision without a hook harness, and it keeps the blocking path in
shell where the other two gates already live.

Why a gate at all: a phase that reads six 900-line files pays for 5,400 lines of
Swift at the phase's own rung, and the useful content is a handful of symbols.
`offload-ref.sh` already caught the other half of that bill - the build log - but
nothing looked at reads.

Why it must not simply block: in Claude Code, `Edit` requires that the SAME file
was read first. A gate that blocks reads during development stops the pipeline
from editing anything. So the development phase is exempt by default, and so is
any file already in the run's edit set. The gate exists for the phases that read
to UNDERSTAND (analysis, review), not for the one that reads to change.

Modes (prefs `global.bulkRead.mode`):
    off       nothing is inspected                      (default)
    observe   decide, log, never block  -  the baseline measurement
    enforce   block a read over the threshold

Fail-open everywhere: any error prints PASS. A gate that guesses wrong must cost
a log line, never a tool call.
"""

import json
import os
import re
import shlex
import sys

# Shipped defaults for the two prefs read below: `bulkRead.minLines` and
# `bulkRead.exemptPhases`. A pref that resolves to anything else - absent, zero,
# a string - takes these rather than silently turning the gate off.
DEFAULT_MIN_LINES = 350
DEFAULT_EXEMPT_PHASES = ["3"]

PREF_PATHS = [
    "~/.claude/multi-agent-preferences.json",
    "~/.config/multi-agent-pipeline/multi-agent-preferences.json",
    "~/.claude/preferences.json",
    "~/.config/multi-agent-pipeline/preferences.json",
]

# Commands that read a whole file to stdout. `sed` is here for `sed -n '1,900p'`,
# which is the spelling an agent reaches for when Read is unavailable.
READ_COMMANDS = {"cat", "head", "tail", "sed", "bat", "less", "more"}

# `1,900p` / `1,$p` / `900p` - a sed print script with a line range.
SED_RANGE = re.compile(r"^'?(\d+)(?:,(\d+|\$))?p'?$")


def emit(verdict, path="", lines=0, why=""):
    sys.stdout.write("%s\t%s\t%s\t%s\n" % (verdict, path, lines, why))
    sys.exit(0)


def load_prefs():
    for raw in PREF_PATHS:
        p = os.path.expanduser(raw)
        if not os.path.isfile(p):
            continue
        try:
            with open(p, "r", encoding="utf-8") as fh:
                return json.load(fh).get("global", {}).get("bulkRead", {}) or {}
        except Exception:
            return {}
    return {}


def positive_int(value, default):
    """A pref counts only as a positive integer; anything else takes the default.

    Same rule as offload-ref.sh: a zero or a string must not silently disable
    the threshold, because that reads as "the gate is off" when it is broken.
    """
    try:
        n = int(value)
    except (TypeError, ValueError):
        return default
    return n if n > 0 else default


def find_state():
    """Walk up from cwd for the run's agent-state.json.

    A hook runs in the worktree where the tool call happens, so the state file
    is at cwd or above it. The env override exists for the smoke gate.
    """
    override = os.environ.get("MULTI_AGENT_STATE")
    if override:
        return override if os.path.isfile(override) else None
    here = os.path.abspath(os.getcwd())
    while True:
        candidate = os.path.join(here, "agent-state.json")
        if os.path.isfile(candidate):
            return candidate
        parent = os.path.dirname(here)
        if parent == here:
            return None
        here = parent


def run_context():
    """(phase, edit_set) for the current run, both best-effort.

    Both come from agent-state.json's REAL shape, checked against
    pipeline/schemas/agent-state.schema.json: the phase is the top-level
    `currentPhase` integer, and the files a phase has touched are
    `phases["<n>"].files[]` - the same list semantic revert uses. An earlier
    draft of this read `run.phase` and `dev.editSet`, neither of which the
    schema declares, so the exemption could never have engaged from state and
    the gate would have blocked development. smoke-bulk-read.sh asserts the
    field names against the schema now, so that cannot come back quietly.
    """
    phase = os.environ.get("MULTI_AGENT_PHASE", "")
    edit_set = set()
    path = find_state()
    if not path:
        return phase, edit_set
    try:
        with open(path, "r", encoding="utf-8") as fh:
            state = json.load(fh)
    except Exception:
        return phase, edit_set
    if not phase:
        current = state.get("currentPhase")
        phase = str(current) if isinstance(current, int) else ""
    # Every phase's file list counts, not only the current one: a review that
    # re-reads what development just wrote is reading the run's own work.
    phases = state.get("phases")
    if isinstance(phases, dict):
        for entry in phases.values():
            if not isinstance(entry, dict):
                continue
            for f in entry.get("files") or []:
                if isinstance(f, str):
                    edit_set.add(os.path.basename(f))
    return phase, edit_set


def count_lines(path):
    """Line count, or -1 when the file cannot be measured.

    Binary-safe and streamed: the gate must not itself load the payload it
    exists to keep out of memory.
    """
    try:
        total = 0
        with open(path, "rb") as fh:
            while True:
                chunk = fh.read(1 << 20)
                if not chunk:
                    break
                total += chunk.count(b"\n")
        return total
    except Exception:
        return -1


def as_count(token):
    """A flag's numeric argument, or a large sentinel when it is not a number.

    Not-a-number must read as UNBOUNDED, not as 1: treating an unparseable
    count as a tiny one is how a gate lets through exactly the reads it exists
    to catch.
    """
    try:
        n = int(token)
    except (TypeError, ValueError):
        return 1 << 30
    return abs(n)


def looks_like_sed_script(token):
    return bool(SED_RANGE.match(token or ""))


def sed_span(script):
    """Lines a `sed` print script covers, or a large sentinel when unclear."""
    m = SED_RANGE.match(script or "")
    if not m:
        return 1 << 30
    start, end = m.group(1), m.group(2)
    try:
        if end in (None, "", "$"):
            return 1 << 30
        return abs(int(end) - int(start)) + 1
    except (TypeError, ValueError):
        return 1 << 30


def read_target(payload):
    """The file a tool call would read whole, or None.

    Returns None for every call that is already bounded - a Read with a small
    `limit`, a `head -n 50` - because those are not what this gate is for.
    """
    tool = payload.get("tool_name", "")
    args = payload.get("tool_input", {}) or {}

    if tool == "Read":
        path = args.get("file_path")
        if not path:
            return None
        limit = args.get("limit")
        if isinstance(limit, int) and limit > 0:
            return (path, limit)
        return (path, None)

    if tool != "Bash":
        return None

    command = args.get("command", "")
    if not command:
        return None
    # A compound command is not decided here: splitting on shell operators
    # correctly is the kind of parsing that goes wrong quietly, and a wrong
    # block is worse than a missed one.
    for operator in ("&&", "||", "|", ";", "$(", "`", ">", "<"):
        if operator in command:
            return None
    try:
        tokens = shlex.split(command)
    except ValueError:
        return None
    if not tokens or os.path.basename(tokens[0]) not in READ_COMMANDS:
        return None

    tool = os.path.basename(tokens[0])
    bound = None
    operands = []
    i = 1
    while i < len(tokens):
        token = tokens[i]

        if not token.startswith("-") or token == "-":
            if tool == "sed" and not operands and looks_like_sed_script(token):
                # `sed '1,900p' file` - the script, not the file.
                bound = max(bound or 0, sed_span(token))
            else:
                operands.append(token)
            i += 1
            continue

        # A following-tail stream is not a bulk read at all; blocking it would
        # be a false positive with no cheap alternative to offer.
        if tool == "tail" and token in ("-f", "-F", "--follow"):
            return None

        # Only head/tail take a separate count operand. `cat -n` NUMBERS LINES -
        # reading its next token as a count swallowed the filename, left no
        # operand, and let `cat -n <bigfile>` through the gate entirely.
        if tool in ("head", "tail") and token in ("-n", "-c"):
            nxt = tokens[i + 1] if i + 1 < len(tokens) else ""
            bound = max(bound or 0, as_count(nxt))
            i += 2
            continue

        # `sed -n` suppresses default output; the range lives in the script,
        # which arrives as the next operand and is handled above.
        if tool == "head" or tool == "tail":
            attached = token.lstrip("-")
            if attached.isdigit():
                bound = max(bound or 0, as_count(attached))
            elif attached[:1] in ("n", "c") and attached[1:].isdigit():
                bound = max(bound or 0, as_count(attached[1:]))
        i += 1

    if len(operands) != 1:
        return None
    return (operands[0], bound)


def main():
    try:
        payload = json.load(sys.stdin)
    except Exception:
        emit("PASS", why="unparseable payload")

    # `bulkRead.mode` decides whether this gate does anything at all.
    prefs = load_prefs()
    mode = str(prefs.get("mode", "off") or "off").lower()
    if mode not in ("observe", "enforce"):
        emit("PASS", why="mode=%s" % mode)

    target = read_target(payload)
    if target is None:
        emit("PASS", why="not a whole-file read")
    path, bound = target

    min_lines = positive_int(prefs.get("minLines"), DEFAULT_MIN_LINES)
    if bound is not None and bound < min_lines:
        emit("PASS", path, 0, "bounded read")

    if not os.path.isfile(path):
        emit("PASS", path, 0, "not a readable file")

    lines = count_lines(path)
    if lines < 0:
        emit("PASS", path, 0, "unmeasurable")
    if lines < min_lines:
        emit("PASS", path, lines, "under threshold")

    phase, edit_set = run_context()
    exempt = prefs.get("exemptPhases")
    if not isinstance(exempt, list) or not exempt:
        exempt = DEFAULT_EXEMPT_PHASES
    exempt = [str(x) for x in exempt]

    # The two exemptions that keep Edit working. Both are stated as PASS with a
    # reason rather than skipped silently, so the telemetry shows how often the
    # gate declined to act and why.
    if phase and phase in exempt:
        emit("PASS", path, lines, "phase %s exempt" % phase)
    if os.path.basename(path) in edit_set:
        emit("PASS", path, lines, "in edit set")

    emit("OBSERVE" if mode == "observe" else "BLOCK", path, lines, "over threshold")


if __name__ == "__main__":
    try:
        main()
    except Exception:
        emit("PASS", why="internal error")