"""
What a command will change, measured on this machine before it runs, so the approver sees a number, not a guess.

  git push --force      -> the commits on the remote that the push would remove (as of your last fetch)
  git reset --hard      -> the files with uncommitted changes it would throw away, and the commits it would leave
  git clean -f          -> the untracked files it would delete (git's own dry run)
  git checkout/restore  -> the changed files it would overwrite
  rm -r, Remove-Item, rmdir /s, del -> how many files and how much data the paths hold

  psql / mysql / sqlite3 DELETE, UPDATE, TRUNCATE, DROP TABLE -> how many rows, counted with the same WHERE
  terraform destroy / apply  -> how many resources go, and which of them hold data (databases, buckets, volumes)
  aws s3 rb / rm --recursive -> how many objects and how much data
  aws rds delete-db-*        -> the database's engine and size, and whether a final snapshot and backups are kept
  kubectl delete             -> the objects that match, and the cluster they're in

Used by the hooks (claude_hook.py, agent_hook.py), which run on the developer's machine: the gateway can't see
the repository, the database or the cloud account. Only reads: git commands here are local (no fetch, no network),
dry runs, or counts. Database counts run in a read-only session with a statement timeout; cloud and cluster
checks are list/describe calls with the same credentials the command itself would use. Everything is bounded in time
and size, and anything that fails is simply left out.
"""
from __future__ import annotations

import glob
import json
import os
import re
import shutil
import subprocess
import tempfile
import time
from pathlib import Path

import commands

BUDGET_SECONDS = 2.0        # for all local predictions on one command line
REMOTE_SECONDS = float(os.getenv("SQUIDBRAKE_EFFECTS_SECONDS", "10"))  # databases, clouds, clusters (only for those)
PLAN_SECONDS = float(os.getenv("SQUIDBRAKE_PLAN_SECONDS", "20"))       # a terraform plan, when it needs one
WALK_LIMIT = 20000          # files counted under the paths a delete names, at most
DELETE_PROGRAMS = {"rm", "remove-item", "ri", "rmdir", "rd", "del", "erase", "rimraf"}
WINDOWS_FLAG = re.compile(r"/[a-zA-Z]$")


def _git(cwd: str, *args: str, timeout: float = 2.0) -> str | None:
    try:
        r = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, timeout=timeout,
                           env={**os.environ, "GIT_OPTIONAL_LOCKS": "0", "GIT_TERMINAL_PROMPT": "0"})
    except (OSError, subprocess.SubprocessError):
        return None
    return r.stdout if r.returncode == 0 else None


def _names(lines: list[str], n: int = 3) -> str:
    shown = ", ".join(lines[:n])
    return shown + (f" and {len(lines) - n} more" if len(lines) > n else "")


def _size(n: float) -> str:
    for unit in ("bytes", "KB", "MB", "GB"):
        if n < 1024 or unit == "GB":
            return (f"{n:.0f} byte" + ("" if n == 1 else "s")) if unit == "bytes" else f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} GB"


# --------------------------------------------------------------------------- git

def _push(words: list[str], cwd: str) -> str | None:
    args = words[2:]
    force = any(a in ("-f", "--force") or a.startswith("--force-with-lease") or a.startswith("+") for a in args)
    if not force:
        return None
    plain = [a.lstrip("+") for a in args if not a.startswith("-")]
    remote = plain[0] if plain else "origin"
    branch = plain[1].split(":")[-1] if len(plain) > 1 else (_git(cwd, "rev-parse", "--abbrev-ref", "HEAD") or "").strip()
    local = plain[1].split(":")[0] if len(plain) > 1 and plain[1].split(":")[0] else "HEAD"
    if not branch or branch == "HEAD":
        return None
    target = f"{remote}/{branch}"
    if _git(cwd, "rev-parse", "--verify", "--quiet", f"refs/remotes/{target}") is None:
        return f"{target} isn't known on this machine yet, so nothing on it can be counted"
    lost = (_git(cwd, "log", "--format=%h %s", f"{local}..{target}") or "").splitlines()
    if not lost:
        return f"No commits on {target} would be lost (as of your last fetch)"
    return (f"Removes {len(lost)} commit{'s' if len(lost) != 1 else ''} from {target} that this branch doesn't have "
            f"(as of your last fetch): {_names([l.split(' ', 1)[-1] for l in lost])}")


def _changed_files(cwd: str, paths: list[str] | None = None) -> list[str]:
    out = _git(cwd, "status", "--porcelain", "--untracked-files=no", *(["--", *paths] if paths else []))
    return [l[3:] for l in (out or "").splitlines() if l.strip()]


def _reset(words: list[str], cwd: str) -> str | None:
    if "--hard" not in words:
        return None
    parts = []
    changed = _changed_files(cwd)
    if changed:
        parts.append(f"throws away uncommitted changes in {len(changed)} file{'s' if len(changed) != 1 else ''}: {_names(changed)}")
    ref = next((w for w in words[2:] if not w.startswith("-")), None)
    if ref:
        left = (_git(cwd, "log", "--format=%s", f"{ref}..HEAD") or "").splitlines()
        if left:
            parts.append(f"moves the branch off {len(left)} commit{'s' if len(left) != 1 else ''}: {_names(left)}")
    return ("It " + "; it ".join(parts)) if parts else "No uncommitted changes would be lost"


def _clean(words: list[str], cwd: str) -> str | None:
    flags = [w for w in words[2:] if w.startswith("-")]
    if not any("f" in f.lstrip("-") or f == "--force" for f in flags):
        return None
    keep = []  # the same flags as a dry run: -fdx -> -dx (plus -n), --force dropped
    for f in flags:
        if f.startswith("--"):
            if f not in ("--force", "--interactive", "--dry-run"):
                keep.append(f)
        elif rest := re.sub(r"[fni]", "", f[1:]):
            keep.append("-" + rest)
    out = _git(cwd, "clean", "-n", *keep, *[w for w in words[2:] if not w.startswith("-")])
    if out is None:
        return None
    files = [l.removeprefix("Would remove ").strip() for l in out.splitlines() if l.strip()]
    if not files:
        return "No untracked files would be deleted"
    return f"Deletes {len(files)} untracked file{'s' if len(files) != 1 else ''} git has never saved: {_names(files)}"


def _checkout(words: list[str], cwd: str) -> str | None:
    sub = words[1]
    if sub == "checkout" and "--" not in words and "." not in words:
        return None
    paths = [w for w in words[words.index("--") + 1:]] if "--" in words else [w for w in words[2:] if not w.startswith("-")]
    if sub == "restore" and ("--staged" in words and "--worktree" not in words):
        return None
    changed = _changed_files(cwd, paths or None)
    if not changed:
        return None
    return f"Overwrites uncommitted changes in {len(changed)} file{'s' if len(changed) != 1 else ''}: {_names(changed)}"


# --------------------------------------------------------------------------- deletes

def delete_targets(cmd: commands.Command, cwd: str) -> list[Path]:
    """The existing paths a delete command names (globs and ~ expanded)."""
    words, prog = cmd.words[1:], cmd.program
    if prog not in DELETE_PROGRAMS:
        return []
    out: list[Path] = []
    skip_next = False
    for w in words:
        if skip_next:
            skip_next = False
            continue
        if w.startswith("-"):
            if prog in ("remove-item", "ri") and w.lower() in ("-path", "-literalpath"):
                continue
            if prog in ("remove-item", "ri") and w.lower() in ("-include", "-exclude", "-filter"):
                skip_next = True
            continue
        if prog in ("rmdir", "rd", "del", "erase") and WINDOWS_FLAG.match(w):
            continue
        w = os.path.expandvars(os.path.expanduser(w.strip("'\"")))
        p = w if os.path.isabs(w) else os.path.join(cwd, w)
        for m in (glob.glob(p) if any(c in p for c in "*?[") else [p]):
            if os.path.lexists(m):
                out.append(Path(m))
    return out


def walk(paths: list[Path], deadline: float) -> tuple[int, int, bool]:
    """(files, bytes, complete) under the paths, stopping at WALK_LIMIT files or the deadline."""
    files = size = 0
    for p in paths:
        if p.is_file() or p.is_symlink():
            files += 1
            size += p.lstat().st_size
            continue
        for root, _dirs, names in os.walk(p):
            for n in names:
                files += 1
                try:
                    size += os.lstat(os.path.join(root, n)).st_size
                except OSError:
                    pass
                if files >= WALK_LIMIT or time.monotonic() > deadline:
                    return files, size, False
    return files, size, True


def _delete(cmd: commands.Command, cwd: str, deadline: float) -> str | None:
    targets = delete_targets(cmd, cwd)
    if not targets:
        return None
    files, size, complete = walk(targets, deadline)
    def shown(t: Path) -> str:  # inside the project: the short path the developer typed
        try:
            return os.path.relpath(t, cwd) if Path(t).resolve().is_relative_to(Path(cwd).resolve()) else str(t)
        except ValueError:      # another drive on Windows
            return str(t)
    where = _names([shown(t) for t in targets], 2)
    more = "+" if not complete else ""
    return f"Deletes {files:,}{more} file{'s' if files != 1 else ''} ({_size(size)}{more}) in {where}"


# --------------------------------------------------------------------------- running other tools (read-only calls)

def _left(deadline: float) -> float:
    return deadline - time.monotonic()


def _env(cmd: commands.Command, extra: dict | None = None) -> dict:
    """The environment the command would run with: ours, plus any FOO=bar it sets in front of itself."""
    env = dict(os.environ)
    for w in commands._words(cmd.raw or ""):
        m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$", w)
        if not m:
            break
        env[m.group(1)] = os.path.expandvars(m.group(2))
    return {**env, **(extra or {})}


def _run(args: list[str], timeout: float, env: dict | None = None, cwd: str | None = None) -> tuple[str | None, bool]:
    """(stdout, finished) of a read-only call. stdout is None if the program isn't here or failed; when time runs
    out, it's whatever was printed so far and finished is False."""
    exe = shutil.which(args[0])
    if not exe or timeout <= 0:
        return None, False
    try:
        r = subprocess.run([exe, *args[1:]], capture_output=True, text=True, timeout=timeout, env=env, cwd=cwd,
                           stdin=subprocess.DEVNULL)
    except subprocess.TimeoutExpired as e:
        out = e.output
        return (out.decode(errors="replace") if isinstance(out, bytes) else out), False
    except (OSError, subprocess.SubprocessError):
        return None, False
    return (r.stdout if r.returncode == 0 else None), True


def _plural(n: int, word: str) -> str:
    return f"{n:,} {word}{'' if n == 1 else 's'}"


# --------------------------------------------------------------------------- databases: count the rows first

SQL_PROGRAMS = {"psql", "mysql", "mariadb", "sqlite3"}
IDENT = r'(?:"[^"]+"|`[^`]+`|\[[^\]]+\]|[A-Za-z_][\w$]*)'
TABLE = rf"{IDENT}(?:\.{IDENT}){{0,2}}"
# Never run as part of a count, even read-only: functions that act, wait, or reach outside the database
SQL_SIDE_EFFECTS = re.compile(r"\b(pg_terminate_backend|pg_cancel_backend|pg_sleep\w*|pg_reload_conf|pg_read_\w+|"
                              r"pg_ls_dir|pg_advisory\w*|pg_notify|lo_\w+|dblink\w*|set_config|nextval|setval|copy|"
                              r"load_file|sleep|benchmark|outfile|dumpfile|load_extension|writefile|readfile|"
                              r"attach|pragma)\b", re.I)


def _statements(sql: str) -> list[str]:
    out, buf, quote = [], [], None
    for ch in sql:
        if quote:
            quote = None if ch == quote else quote
        elif ch in "'\"`":
            quote = ch
        elif ch == ";":
            out.append("".join(buf)); buf = []
            continue
        buf.append(ch)
    out.append("".join(buf))
    return [s.strip() for s in out if s.strip()]


def _top_level(sql: str, word: str) -> int:
    """Where `word` first appears as a keyword outside quotes and brackets, or -1."""
    depth, quote = 0, None
    for i, ch in enumerate(sql):
        if quote:
            quote = None if ch == quote else quote
        elif ch in "'\"`":
            quote = ch
        elif ch == "(":
            depth += 1
        elif ch == ")":
            depth -= 1
        elif depth == 0 and sql[i:i + len(word)].lower() == word and (i == 0 or not (sql[i - 1].isalnum() or sql[i - 1] == "_")) \
                and not (sql[i + len(word):i + len(word) + 1].isalnum() or sql[i + len(word):i + len(word) + 1] == "_"):
            return i
    return -1


def sql_targets(sql: str) -> list[tuple[str, str, str | None]] | None:
    """(verb, table, where) for each statement, or None unless every statement is one we can count safely:
    DELETE, UPDATE, TRUNCATE or DROP TABLE, with no side-effect functions in the WHERE."""
    stmts = _statements(sql)
    if not stmts or len(stmts) > 3:
        return None
    out: list[tuple[str, str, str | None]] = []
    for s in stmts:
        if (cut := _top_level(s, "returning")) >= 0:
            s = s[:cut].rstrip()
        if m := re.match(rf"(?is)delete\s+from\s+(?:only\s+)?({TABLE})(\s+(?:as\s+)?(?!where\b)\w+)?\s*(.*)$", s):
            table, alias, rest = m.group(1), m.group(2) or "", m.group(3).strip()
            if rest and not re.match(r"(?is)where\b", rest):
                return None                                     # USING, LIMIT, ...: not counted
            out.append(("delete", table + alias, rest[5:].strip() or None if rest else None))
        elif m := re.match(rf"(?is)update\s+(?:only\s+)?({TABLE})(\s+(?:as\s+)?(?!set\b)\w+)?\s+set\s+(.*)$", s):
            table, alias, rest = m.group(1), m.group(2) or "", m.group(3)
            if _top_level(rest, "from") >= 0:
                return None                                     # UPDATE ... FROM: not counted
            w = _top_level(rest, "where")
            out.append(("update", table + alias, rest[w + 5:].strip() or None if w >= 0 else None))
        elif m := re.match(rf"(?is)truncate\s+(?:table\s+)?(?:only\s+)?({TABLE}(?:\s*,\s*{TABLE})*)"
                           r"(?:\s+(?:cascade|restrict|restart\s+identity|continue\s+identity))*$", s):
            out += [("truncate", t.strip(), None) for t in m.group(1).split(",")]
        elif m := re.match(rf"(?is)drop\s+table\s+(?:if\s+exists\s+)?({TABLE}(?:\s*,\s*{TABLE})*)(?:\s+(?:cascade|restrict))?$", s):
            out += [("drop", t.strip(), None) for t in m.group(1).split(",")]
        else:
            return None
    if any(w and SQL_SIDE_EFFECTS.search(w) for _, _, w in out):
        return None
    return out[:5]


def _count_query(table: str, where: str | None) -> str:
    if where:
        return f"SELECT (SELECT count(*) FROM {table} WHERE {where}), (SELECT count(*) FROM {table})"
    return f"SELECT count(*) FROM {table}"


def _rows_message(verb: str, table: str, where: str | None, counts: list[int], label: str) -> str:
    name = table.split()[0]
    at = f" ({label})" if label else ""
    if where and len(counts) == 2:
        hit, total = counts
        every = hit == total and total > 0
        if verb == "delete":
            return (f"Deletes all {_plural(total, 'row')} in {name}{at}: the WHERE matches every row" if every
                    else f"Deletes {_plural(hit, 'row')} of {total:,} in {name}{at}")
        return (f"Changes all {_plural(total, 'row')} in {name}{at}: the WHERE matches every row" if every
                else f"Changes {_plural(hit, 'row')} of {total:,} in {name}{at}")
    n = counts[0]
    return {"delete": f"Deletes all {_plural(n, 'row')} in {name}{at}: there's no WHERE",
            "update": f"Changes all {_plural(n, 'row')} in {name}{at}: there's no WHERE",
            "truncate": f"Empties {name}{at}: {_plural(n, 'row')}",
            "drop": f"Drops table {name}{at} with its {_plural(n, 'row')}"}[verb]


def _sql_text(words: list[str], cwd: str, flags: tuple[str, ...], file_flags: tuple[str, ...]) -> str | None:
    """The SQL a client runs from -c/-e or a -f file (small local files only)."""
    for i, w in enumerate(words):
        nxt = words[i + 1] if i + 1 < len(words) else ""
        if w in flags or re.fullmatch(r"-[A-Za-z]*[" + "".join(f[1] for f in flags if len(f) == 2) + "]", w):
            return nxt
        for f in flags:
            if f.startswith("--") and w.startswith(f + "="):
                return w.split("=", 1)[1]
        if w in file_flags or any(f.startswith("--") and w.startswith(f + "=") for f in file_flags):
            path = Path(cwd, w.split("=", 1)[1] if "=" in w else nxt)
            try:
                return path.read_text(encoding="utf-8", errors="replace") if path.stat().st_size < 100_000 else None
            except OSError:
                return None
    return None


def _client_args(words: list[str], value_flags: set[str], drop_flags: set[str], drop_letters: str,
                 attached: tuple[str, ...] = ()) -> list[str]:
    """Only the connection arguments of a database client command: where to connect and as whom.
    drop_letters: short flags whose value is SQL or a file, also when combined (`psql -tAc "..."`)."""
    out, i = [], 0
    while i < len(words):
        w = words[i]
        if w in value_flags:
            out += words[i:i + 2]; i += 2; continue
        if any(w.startswith(f + "=") for f in value_flags if f.startswith("--")) or \
                any(w.startswith(f) and len(w) > 2 for f in value_flags | set(attached) if len(f) == 2):
            out.append(w); i += 1; continue
        if w in drop_flags or re.fullmatch(rf"-[A-Za-z]*[{drop_letters}]", w):
            i += 2; continue
        if not w.startswith("-"):
            out.append(w)
        i += 1
    return out


def _label(conn: list[str]) -> str:
    """Which database, for the approver: host/dbname from a URL or -h/-d flags."""
    for w in conn:
        if m := re.match(r"^\w+://(?:[^@/]*@)?([^/:?]+)(?::\d+)?/([^?]+)", w):
            return f"{m.group(2)} on {m.group(1)}"
    host = next((conn[i + 1] for i, w in enumerate(conn[:-1]) if w in ("-h", "--host")), "")
    db = next((conn[i + 1] for i, w in enumerate(conn[:-1]) if w in ("-d", "--dbname", "-D", "--database")), "")
    return " on ".join(x for x in (db, host) if x)


def _sql(cmd: commands.Command, cwd: str, deadline: float) -> list[str]:
    words = [os.path.expandvars(w) for w in cmd.words[1:]]
    prog = cmd.program
    if prog == "sqlite3":
        pos = [w for w in words if not w.startswith("-")]
        if len(pos) < 2:
            return []
        db, sql = Path(cwd, pos[0]), pos[1]
        targets = sql_targets(sql)
        if not targets or not db.is_file():
            return []
        out = []
        import sqlite3   # here, not at the top: every hook call loads this file, few touch a SQLite database
        con = sqlite3.connect(db.resolve().as_uri() + "?mode=ro", uri=True, timeout=1)
        try:
            con.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, 10000)
            for verb, table, where in targets:
                counts = list(con.execute(_count_query(table, where)).fetchone())
                out.append(_rows_message(verb, table, where, counts, db.name))
        except sqlite3.Error:
            pass
        finally:
            con.close()
        return out
    if prog == "psql":
        sql = _sql_text(words, cwd, ("-c", "--command"), ("-f", "--file"))
        conn = _client_args(words, {"-h", "--host", "-p", "--port", "-U", "--username", "-d", "--dbname"},
                            {"-c", "--command", "-f", "--file", "-o", "--output", "-L", "--log-file", "-v", "--set",
                             "--variable", "-P", "--pset"}, "cf")
        options = (os.environ.get("PGOPTIONS", "") + " -c default_transaction_read_only=on -c statement_timeout=5000").strip()
        env = _env(cmd, {"PGOPTIONS": options, "PGCONNECT_TIMEOUT": "3"})
        run = lambda q: _run(["psql", *conn, "-X", "-q", "-A", "-t", "-v", "ON_ERROR_STOP=1", "-c", q],
                             _left(deadline), env, cwd)
        split = "|"
    else:                                                       # mysql, mariadb
        sql = _sql_text(words, cwd, ("-e", "--execute"), ())
        conn = _client_args(words, {"-h", "--host", "-P", "--port", "-u", "--user", "-D", "--database", "-S", "--socket",
                                    "--password", "--defaults-file", "--defaults-extra-file"},
                            {"-e", "--execute"}, "e", attached=("-p",))   # -pSECRET only; a bare -p would prompt
        env = _env(cmd)
        # MAX_EXECUTION_TIME stops the count on the server too (MySQL); MariaDB reads the hint as a comment
        run = lambda q: _run([prog, *conn, "-N", "-B", "--connect-timeout=3", "-e", "SET SESSION TRANSACTION READ ONLY; "
                              + q.replace("SELECT ", "SELECT /*+ MAX_EXECUTION_TIME(5000) */ ", 1)], _left(deadline), env, cwd)
        split = "\t"
    targets = sql_targets(sql or "")
    if not targets:
        return []
    out = []
    for verb, table, where in targets:
        text, done = run(_count_query(table, where))
        line = next((l for l in (text or "").splitlines() if l.strip()), "") if done else ""
        try:
            counts = [int(x) for x in line.strip().split(split)]
        except ValueError:
            continue
        out.append(_rows_message(verb, table, where, counts, _label(conn)))
    return out


# --------------------------------------------------------------------------- terraform: what a plan would destroy

DATA_TYPES = re.compile(r"(db_instance|rds_cluster|docdb|neptune|redshift|dynamodb_table|elasticache|memorydb|"
                        r"opensearch|elasticsearch|s3_bucket$|storage_bucket$|storage_account$|ebs_volume|"
                        r"efs_file_system|fsx|backup_vault|sql_database|sql_server|database|bigquery_dataset|"
                        r"bigquery_table|bigtable|spanner|firestore|cosmosdb|compute_disk|managed_disk|kms_key|"
                        r"secretsmanager_secret|key_vault)", re.I)


def _tf_type(address: str) -> str:
    parts = re.sub(r"\[[^\]]*\]", "", address).split(".")
    return parts[-2] if len(parts) >= 2 else ""


def _holding_data(addresses: list[str]) -> list[str]:
    return [a for a in addresses if DATA_TYPES.search(_tf_type(a))]


def _tf_summary(verb: str, addresses: list[str], where: str) -> str:
    data = _holding_data(addresses)
    msg = f"{verb} {_plural(len(addresses), 'resource')}{where}"
    if data:
        return msg + f", including {len(data)} that hold{'s' if len(data) == 1 else ''} data: {_names(data)}"
    return msg + f": {_names(addresses)}"


def _terraform(cmd: commands.Command, cwd: str, deadline: float) -> str | None:
    words = [os.path.expandvars(w) for w in cmd.words]
    tf, chdir = words[0], [w for w in words[1:] if w.startswith("-chdir=")]
    rest = [w for w in words[1:] if not w.startswith("-chdir=")]
    sub = next((w for w in rest if not w.startswith("-")), None)
    if sub not in ("destroy", "apply"):
        return None
    args = rest[rest.index(sub) + 1:]
    plan_args, targets, positional, i = [], [], [], 0
    while i < len(args):
        a = args[i]
        if a in ("-var", "-var-file", "-target", "-replace") and i + 1 < len(args):
            plan_args += [f"{a}={args[i + 1]}"]; i += 2; continue
        if re.match(r"-(var|var-file|target|replace)=", a):
            plan_args.append(a)
        elif not a.startswith("-"):
            positional.append(a)
        i += 1
    targets = [a.split("=", 1)[1] for a in plan_args if a.startswith("-target=")]
    env = _env(cmd, {"TF_IN_AUTOMATION": "1", "TF_INPUT": "0"})
    ws, _ = _run([tf, *chdir, "workspace", "show"], min(3, _left(deadline)), env, cwd)
    ws = (ws or "").strip()
    where = f" in workspace {ws}" if ws and ws != "default" else ""

    if sub == "destroy" or "-destroy" in args:
        text, done = _run([tf, *chdir, "state", "list", *targets], _left(deadline), env, cwd)
        if text is None or not done:
            return None
        gone = [l.strip() for l in text.splitlines() if l.strip() and not re.match(r"(.*\.)?data\.", l.strip())]
        if not gone:
            return f"Destroys nothing{where}: the state has no matching resources"
        return _tf_summary("Destroys", gone, where)

    plan_file, tmp = (positional[0] if positional else None), None
    try:
        if not plan_file:                                       # plan against the saved state (no refresh, no lock)
            fd, tmp = tempfile.mkstemp(suffix=".tfplan")
            os.close(fd)
            _, done = _run([tf, *chdir, "plan", "-refresh=false", "-lock=false", "-input=false", "-no-color",
                            f"-out={tmp}", *plan_args], PLAN_SECONDS, env, cwd)
            if not done:
                return None
            plan_file = tmp
        text, done = _run([tf, *chdir, "show", "-json", plan_file], 10, env, cwd)
        if not text or not done:
            return None
        changes = json.loads(text).get("resource_changes") or []
    finally:
        if tmp:
            try:
                os.unlink(tmp)
            except OSError:
                pass
    acts = [(c.get("address", "?"), (c.get("change") or {}).get("actions") or []) for c in changes if c.get("mode") != "data"]
    gone = [a for a, x in acts if x == ["delete"]]
    replaced = [a for a, x in acts if "delete" in x and "create" in x]
    if not gone and not replaced:
        adds = sum(1 for _, x in acts if x == ["create"])
        updates = sum(1 for _, x in acts if x == ["update"])
        return f"Destroys nothing{where}: adds {adds:,} and changes {updates:,} resources (plan against the saved state)"
    parts = []
    if gone:
        parts.append(_tf_summary("Destroys", gone, where))
    if replaced:
        parts.append(_tf_summary("Replaces (destroys, then creates again)", replaced, "" if gone else where))
    return "; ".join(parts) + " (plan against the saved state)"


# --------------------------------------------------------------------------- AWS: what a delete takes with it

AWS_GLOBALS = ("--profile", "--region", "--endpoint-url")
AWS_SWITCHES = {"--recursive", "--force", "--dryrun", "--quiet", "--only-show-errors", "--skip-final-snapshot",
                "--delete-automated-backups", "--no-delete-automated-backups", "--debug", "--no-verify-ssl",
                "--no-paginate", "--no-sign-request", "--no-cli-pager"}


def _aws(cmd: commands.Command, cwd: str, deadline: float) -> str | None:
    words = [os.path.expandvars(w) for w in cmd.words[1:]]
    keep, pos, flags, i = [], [], {}, 0
    while i < len(words):
        w = words[i]
        if w in AWS_GLOBALS and i + 1 < len(words):
            keep += words[i:i + 2]; i += 2; continue
        if any(w.startswith(g + "=") for g in AWS_GLOBALS):
            keep.append(w); i += 1; continue
        if w.startswith("--"):
            name, _, value = w.partition("=")
            if name not in AWS_SWITCHES and not value and i + 1 < len(words) and not words[i + 1].startswith("-"):
                value = words[i + 1]; i += 1
            flags[name] = value
        else:
            pos.append(w)
        i += 1
    env = _env(cmd)
    if pos[:2] in (["s3", "rb"], ["s3", "rm"]) and len(pos) > 2 and pos[2].startswith("s3://"):
        if (pos[1] == "rb" and "--force" not in flags) or (pos[1] == "rm" and "--recursive" not in flags) \
                or "--dryrun" in flags:
            return None                                         # an empty bucket, one object, or a dry run
        text, done = _run(["aws", "s3", "ls", pos[2], "--recursive", "--summarize", *keep], _left(deadline), env, cwd)
        if text is None:
            return None
        if done and (m := re.search(r"Total Objects:\s*(\d+)\s*Total Size:\s*(\d+)", text)):
            files, size = int(m.group(1)), int(m.group(2))
            more = ""
        else:                                                   # still listing when time ran out: what it saw so far
            rows = [l.split(None, 3) for l in text.splitlines()]
            rows = [r for r in rows if len(r) == 4 and r[2].isdigit()]
            files, size, more = len(rows), sum(int(r[2]) for r in rows), "+"
        what = "the bucket and all" if pos[1] == "rb" else "all"
        return f"Deletes {what} {files:,}{more} object{'s' if files != 1 else ''} ({_size(size)}{more}) in {pos[2]}"
    if pos[:2] in (["rds", "delete-db-instance"], ["rds", "delete-db-cluster"]):
        cluster = pos[1] == "delete-db-cluster"
        name = flags.get("--db-cluster-identifier" if cluster else "--db-instance-identifier")
        if not name:
            return None
        parts = [f"Deletes RDS {'cluster' if cluster else 'database'} {name}"]
        text, done = _run(["aws", "rds", "describe-db-clusters" if cluster else "describe-db-instances",
                           "--db-cluster-identifier" if cluster else "--db-instance-identifier", name,
                           "--output", "json", *keep], _left(deadline), env, cwd)
        protected = False
        try:
            db = json.loads(text)["DBClusters" if cluster else "DBInstances"][0] if text and done else {}
            if db:
                size = f", {db['AllocatedStorage']} GB" if db.get("AllocatedStorage") else ""
                parts[0] += f" ({db.get('Engine', '?')}{size})"
                protected = bool(db.get("DeletionProtection"))
        except (ValueError, KeyError, IndexError, TypeError):
            pass
        if "--skip-final-snapshot" in flags:
            parts.append("no final snapshot is taken (--skip-final-snapshot)")
        elif flags.get("--final-db-snapshot-identifier"):
            parts.append(f"a final snapshot {flags['--final-db-snapshot-identifier']} is kept")
        if "--no-delete-automated-backups" not in flags:
            parts.append("its automated backups are removed with it (AWS's default)")
        if protected:
            parts.append("deletion protection is on, so AWS refuses until someone turns it off")
        return "; ".join(parts)
    return None


# --------------------------------------------------------------------------- Kubernetes: what a delete matches

KUBE_DELETE_ONLY = {"--grace-period", "--force", "--now", "--wait", "--cascade", "--ignore-not-found", "--timeout",
                    "--dry-run", "-i", "--interactive", "--all", "--raw"}
KUBE_VALUED = {"--grace-period", "--timeout", "--raw"}      # the others take a value only as --flag=value


def _kubectl(cmd: commands.Command, cwd: str, deadline: float) -> str | None:
    words = [os.path.expandvars(w) for w in cmd.words]
    rest = words[1:]
    at = next((i for i, w in enumerate(rest) if not w.startswith("-") and (i == 0 or not rest[i - 1] in
               ("-n", "--namespace", "--context", "--kubeconfig", "-l", "--selector", "-f", "--filename"))), None)
    if at is None or rest[at].lower() != "delete" or any(w.startswith("--dry-run") for w in rest):
        return None
    args, after, i = rest[:at] + ["get"], rest[at + 1:], 0
    while i < len(after):                                       # the same selection, without delete's own options
        w = after[i]
        if w.split("=", 1)[0] in KUBE_DELETE_ONLY:
            if "=" not in w and w in KUBE_VALUED and i + 1 < len(after) and not after[i + 1].startswith("-"):
                i += 1
        else:
            args.append(w)
        i += 1
    env = _env(cmd)
    text, done = _run([words[0], *args, "-o", "name", "--ignore-not-found"], _left(deadline), env, cwd)
    if text is None or not done:
        return None
    names = [l.strip() for l in text.splitlines() if l.strip()]
    if not names:
        return "Nothing matches it right now"
    get = lambda flag: next((rest[i + 1] for i, w in enumerate(rest[:-1]) if w == flag), None)
    ns = "all namespaces" if any(w in ("-A", "--all-namespaces") for w in rest) else \
        f"namespace {get('-n') or get('--namespace')}" if (get('-n') or get('--namespace')) else ""
    context = get("--context") or (_run([words[0], "config", "current-context"], min(2, _left(deadline)), env, cwd)[0] or "").strip()
    where = " in " + " on ".join(x for x in (ns, f"cluster {context}" if context else "") if x) if ns or context else ""
    inside = " and everything in them" if any(n.startswith("namespace/") for n in names) else ""
    return f"Deletes {_plural(len(names), 'Kubernetes object')}{where}{inside}: {_names(names)}"


# --------------------------------------------------------------------------- entry point

def predict(line: str, cwd: str | None = None) -> list[str]:
    """Plain-English lines saying what the command line would change. Empty when there's nothing to measure."""
    cwd = cwd or os.getcwd()
    start = time.monotonic()
    deadline, remote = start + BUDGET_SECONDS, start + REMOTE_SECONDS
    out: list[str] = []
    try:
        reading = commands.read(line)
    except Exception:
        return out
    for cmd in reading.commands:
        w = [x for x in cmd.words]
        local = cmd.program not in SQL_PROGRAMS | {"terraform", "tofu", "aws", "kubectl", "oc"}
        if time.monotonic() > (deadline if local else remote):
            continue
        try:
            if cmd.program == "git" and len(w) > 1:
                fn = {"push": _push, "reset": _reset, "clean": _clean, "checkout": _checkout, "restore": _checkout}.get(w[1])
                msg = fn(w, cwd) if fn else None
            elif cmd.program in SQL_PROGRAMS:
                msg = _sql(cmd, cwd, remote)
            elif cmd.program in ("terraform", "tofu"):
                msg = _terraform(cmd, cwd, remote)
            elif cmd.program == "aws":
                msg = _aws(cmd, cwd, remote)
            elif cmd.program in ("kubectl", "oc"):
                msg = _kubectl(cmd, cwd, remote)
            else:
                msg = _delete(cmd, cwd, deadline)
        except Exception:
            msg = None
        for m in ([msg] if isinstance(msg, str) else msg or []):
            if m and m[:400] not in out:
                out.append(m[:400])
    return out[:5]