fix: add os import for HELPER_PCT_RUN.readable() check

disk-gc-threat-response.prose.md

@@ -84,6 +84,32 @@ and escalation trail.
 - May also be invoked manually: `prose run disk-gc-threat-response`
 - Threat-driven: if Amber/Red/Critical detected, immediate GC phase activates
 
+## Scanner: scripts/disk-gc-scan.py
+
+The fleet scan is executed by `scripts/disk-gc-scan.py`, which makes reachability
+verdicts deterministic:
+
+1. **Retry on failure:** Each probe retries once before declaring a guest unreachable.
+2. **Named probe target:** Every rendered line names the guest, CT id, node, and
+   access method actually used.
+3. **Failure kind printed:** An unreachable guest is reported with its failure kind
+   (timeout, ssh-auth, no-route, conn-refused, ssh-exit-N) — never as a bare
+   "unreachable" verdict.
+4. **Per-guest access method:** The correct access path is selected from a per-guest
+   map so the wrong path cannot be picked by an executor improvising:
+   - CT 105 (kagentz) = `ssh root@kagentz` (NOT `pct exec 105` — pct exec sees
+     loop0/59G instead of the real 99G filesystem)
+   - CT 109 (docker-vm) = `ssh root@192.168.68.7` (NOT `pct exec` — it's a KVM VM)
+   - All other CTs = `pct-run <ct_id>` (which uses `pct exec` via SSH to the node)
+5. **Every figure traces to a named probe:** The scan output prints the exact command
+   that produced each disk figure, so two different guests can never render
+   identical numbers without the probe commands proving it.
+
+Run: `python3 scripts/disk-gc-scan.py` (or `--json` for machine-readable output).
+
+The scan feeds into `scripts/disk-gc-plan.py`, which applies the report-only gate
+from the `report_only_guests` YAML block above.
+
 ## Shape
 
 - `self`: scan all CTs via Proxmox API + SSH exec, trigger GC, alert

scripts/disk-gc-scan.py

@@ -0,0 +1,339 @@
+#!/usr/bin/env python3
+"""disk-gc-scan — deterministic disk usage probe for fleet guests.
+
+This is the executable scanner side of `disk-gc-threat-response.prose.md`. It exists
+so reachability verdicts are deterministic and every rendered field traces to a
+named probe command.
+
+DESIGN PRINCIPLES (per task disk-gc-probe-false-unreachable-20260913):
+1. REACHABILITY VERDICTS ARE DETERMINISTIC:
+   - Retry once on failure before declaring unreachable
+   - Always name the probe target (guest, host, access method) on the line it prints
+   - Never render a failed probe as a bare service/guest verdict — print the failure kind
+
+2. PER-GUEST ACCESS METHOD CANNOT BE MIS-SELECTED:
+   - CT 105 (kagentz) = ssh root@kagentz (NOT pct exec 105)
+   - VM 109 (docker-vm) = ssh root@192.168.68.7 (NOT pct)
+   - All other CTs = pct-run <ct_id> (which uses pct exec)
+   - The access method is selected from a per-guest map so the wrong path cannot be
+     picked by an executor improvising
+
+3. EVERY RENDERED FIELD AUDITED:
+   - For each guest, print the probe command that produced the figure
+   - If a figure comes from a different kind of measurement than the column claims,
+     name it explicitly
+
+4. FIX A (CWD independence): Resolve repo-relative files from the script's own
+   location, not the caller's CWD.
+   FIX B (df columns): Parse df output correctly and print labelled, human-readable
+   output.
+
+Usage:
+    disk-gc-scan.py              # scan all guests
+    disk-gc-scan.py --json       # machine-readable output
+
+Exit codes: 0 ok (all guests probed), 1 probe error
+"""
+from __future__ import annotations
+
+import json
+import pathlib
+import subprocess
+import sys
+import time
+import os
+from dataclasses import dataclass
+from typing import Optional
+
+# Resolve repo-relative files from the script's own location, not the caller's CWD
+SCRIPT_DIR = pathlib.Path(__file__).resolve().parent
+HELPER_PCT_RUN = SCRIPT_DIR / "pct-run.sh"
+
+# Per-guest access method map. This is the authoritative source for how to reach
+# each guest — the contract's prose documentation must match this map.
+#
+# Access methods:
+#   - "pct-run": use pct-run.sh <ct_id> (pct exec via SSH to node)
+#   - "ssh-host": use ssh root@<hostname>
+#   - "ssh-ip": use ssh root@<ip>
+
+@dataclass
+class Guest:
+    """A guest to probe."""
+    ct_id: str
+    hostname: str
+    ip: Optional[str]
+    node: str
+    access_method: str  # "pct-run", "ssh-host", "ssh-ip"
+    probe_target: str   # human-readable target name for the probe line
+
+    @property
+    def is_reachable(self) -> bool:
+        return self.probe_result is not None and self.probe_result.exit_code == 0
+
+    @property
+    def usage_pct(self) -> Optional[float]:
+        return self.probe_result.usage_pct if self.probe_result else None
+
+    @property
+    def usage_str(self) -> Optional[str]:
+        return self.probe_result.usage_str if self.probe_result else None
+
+    probe_result: Optional["ProbeResult"] = None
+
+
+@dataclass
+class ProbeResult:
+    """Result of probing a guest."""
+    exit_code: int
+    usage_pct: Optional[float]
+    usage_str: Optional[str]
+    probe_cmd: str
+    failure_kind: Optional[str]  # "timeout", "ssh-auth", "no-route", "command-not-found", None
+
+    @property
+    def is_reachable(self) -> bool:
+        return self.exit_code == 0
+
+
+# Fleet inventory (verified against pvesh /cluster/resources 2026-09-12)
+GUESTS: list[Guest] = [
+    # amdpve (192.168.68.15)
+    Guest(ct_id="105", hostname="kagentz", ip="192.168.68.105", node="amdpve",
+          access_method="ssh-host", probe_target="kagentz (CT 105, amdpve)"),
+    Guest(ct_id="112", hostname="tanko", ip="192.168.68.112", node="amdpve",
+          access_method="pct-run", probe_target="tanko (CT 112, amdpve)"),
+    Guest(ct_id="113", hostname="baggy", ip="192.168.68.113", node="amdpve",
+          access_method="pct-run", probe_target="baggy (CT 113, amdpve)"),
+    Guest(ct_id="115", hostname="scottdenya", ip="192.168.68.115", node="amdpve",
+          access_method="pct-run", probe_target="scottdenya (CT 115, amdpve)"),
+    Guest(ct_id="120", hostname="adguard2", ip="192.168.68.120", node="amdpve",
+          access_method="pct-run", probe_target="adguard2 (CT 120, amdpve)"),
+    # minipve (192.168.68.12)
+    Guest(ct_id="100", hostname="abiba", ip="192.168.68.100", node="minipve",
+          access_method="pct-run", probe_target="abiba (CT 100, minipve)"),
+    Guest(ct_id="102", hostname="adguard", ip="192.168.68.102", node="minipve",
+          access_method="pct-run", probe_target="adguard (CT 102, minipve)"),
+    Guest(ct_id="104", hostname="authentik", ip="192.168.68.104", node="minipve",
+          access_method="pct-run", probe_target="authentik (CT 104, minipve)"),
+    Guest(ct_id="110", hostname="gitea", ip="192.168.68.110", node="minipve",
+          access_method="pct-run", probe_target="gitea (CT 110, minipve)"),
+    Guest(ct_id="116", hostname="syslog-api", ip="192.168.68.116", node="minipve",
+          access_method="pct-run", probe_target="syslog-api (CT 116, minipve)"),
+    Guest(ct_id="119", hostname="infisical-vault", ip="192.168.68.119", node="minipve",
+          access_method="pct-run", probe_target="infisical-vault (CT 119, minipve)"),
+    # storepve (192.168.68.6)
+    Guest(ct_id="106", hostname="ra-h-os", ip="192.168.68.106", node="storepve",
+          access_method="pct-run", probe_target="ra-h-os (CT 106, storepve)"),
+    Guest(ct_id="107", hostname="proxmox-backup", ip="192.168.68.107", node="storepve",
+          access_method="pct-run", probe_target="proxmox-backup (CT 107, storepve)"),
+    Guest(ct_id="108", hostname="media", ip="192.168.68.108", node="storepve",
+          access_method="pct-run", probe_target="media (CT 108, storepve)"),
+    Guest(ct_id="111", hostname="tdunna", ip="192.168.68.129", node="storepve",
+          access_method="pct-run", probe_target="tdunna (CT 111, storepve)"),
+    Guest(ct_id="117", hostname="zulip", ip="192.168.68.117", node="storepve",
+          access_method="pct-run", probe_target="zulip (CT 117, storepve)"),
+    Guest(ct_id="118", hostname="jdownloader", ip="192.168.68.118", node="storepve",
+          access_method="pct-run", probe_target="jdownloader (CT 118, storepve)"),
+    # KVM VMs (direct SSH)
+    Guest(ct_id="109", hostname="docker-vm", ip="192.168.68.7", node="storepve",
+          access_method="ssh-ip", probe_target="docker-vm (CT 109, KVM VM)"),
+]
+
+# GPU bare-metal hosts
+GPU_HOSTS = [
+    {"hostname": "acerpve", "ip": "192.168.68.9", "gpu": "RTX 3090",
+     "probe_target": "RTX 3090 (bare metal .9)"},
+    {"hostname": "ocupve", "ip": "192.168.68.110", "gpu": "RTX 5070",
+     "probe_target": "RTX 5070 (bare metal .110)"},
+    {"hostname": "amdpve", "ip": "192.168.68.15", "gpu": "Strix Halo",
+     "probe_target": "Strix Halo (bare metal .15)"},
+]
+
+CONNECT_TIMEOUT = 5
+SSH_OPTS = "-o BatchMode=yes -o ConnectTimeout=" + str(CONNECT_TIMEOUT)
+
+
+def run_cmd(cmd: str, timeout: int = 30) -> tuple[int, str, str]:
+    """Run a command and return (exit_code, stdout, stderr)."""
+    try:
+        result = subprocess.run(
+            cmd, shell=True, capture_output=True, text=True, timeout=timeout
+        )
+        return result.returncode, result.stdout.strip(), result.stderr.strip()
+    except subprocess.TimeoutExpired:
+        return 124, "", "timeout"
+    except Exception as e:
+        return 1, "", str(e)
+
+
+def probe_guest(guest: Guest) -> ProbeResult:
+    """Probe a single guest and return the result.
+
+    Access method is selected from guest.access_method:
+      - "pct-run": pct-run.sh <ct_id> "df -P / | tail -1"
+      - "ssh-host": ssh root@<hostname> "df -P / | tail -1"
+      - "ssh-ip": ssh root@<ip> "df -P / | tail -1"
+    """
+    df_cmd = "df -P / | tail -1"
+
+    if guest.access_method == "pct-run":
+        # Use absolute path to helper so CWD doesn't matter
+        probe_cmd = f'bash {HELPER_PCT_RUN} {guest.ct_id} "{df_cmd}"'
+    elif guest.access_method == "ssh-host":
+        probe_cmd = f'ssh {SSH_OPTS} root@{guest.hostname} "{df_cmd}"'
+    elif guest.access_method == "ssh-ip":
+        probe_cmd = f'ssh {SSH_OPTS} root@{guest.ip} "{df_cmd}"'
+    else:
+        raise ValueError(f"unknown access_method: {guest.access_method}")
+
+    # Check helper exists and is readable BEFORE probing (for pct-run guests)
+    # This prevents scanner errors from being rendered as guest verdicts
+    if guest.access_method == "pct-run":
+        if not HELPER_PCT_RUN.exists():
+            print(f"SCANNER ERROR: helper not found: {HELPER_PCT_RUN}", file=sys.stderr)
+            sys.exit(1)
+        if not os.access(str(HELPER_PCT_RUN), os.R_OK):
+            print(f"SCANNER ERROR: helper not readable: {HELPER_PCT_RUN}", file=sys.stderr)
+            sys.exit(1)
+
+    # Retry once on failure before declaring unreachable
+    for attempt in range(2):
+        exit_code, stdout, stderr = run_cmd(probe_cmd, timeout=15)
+
+        if exit_code == 0:
+            # Parse df output: Filesystem 1024-blocks Used Available Capacity Mounted on
+            # parts[0]=Filesystem, parts[1]=Total (1K blocks), parts[2]=Used, parts[3]=Available, parts[4]=Capacity
+            parts = stdout.split()
+            if len(parts) >= 5:
+                capacity_str = parts[4]  # e.g., "34%"
+                usage_pct = float(capacity_str.rstrip("%"))
+                total_blocks = int(parts[1])
+                used_blocks = int(parts[2])
+                avail_blocks = int(parts[3])
+                # Convert to human-readable units
+                def to_gb(blocks: int) -> float:
+                    return blocks / (1024 * 1024)
+                total_gb = to_gb(total_blocks)
+                used_gb = to_gb(used_blocks)
+                avail_gb = to_gb(avail_blocks)
+                # FIX B: print labelled, unambiguous output
+                usage_str = f"{capacity_str} ({used_gb:.1f}G used of {total_gb:.1f}G total, {avail_gb:.1f}G free)"
+                return ProbeResult(
+                    exit_code=0,
+                    usage_pct=usage_pct,
+                    usage_str=usage_str,
+                    probe_cmd=probe_cmd,
+                    failure_kind=None,
+                )
+            else:
+                # Unexpected output format
+                return ProbeResult(
+                    exit_code=1,
+                    usage_pct=None,
+                    usage_str=None,
+                    probe_cmd=probe_cmd,
+                    failure_kind="parse-error",
+                )
+        else:
+            # Classify failure kind
+            if exit_code == 124:
+                failure_kind = "timeout"
+            elif "Connection timed out" in stderr or "timed out" in stderr:
+                failure_kind = "timeout"
+            elif "Permission denied" in stderr or "password" in stderr.lower():
+                failure_kind = "ssh-auth"
+            elif "No route to host" in stderr or "unreachable" in stderr:
+                failure_kind = "no-route"
+            elif "Connection refused" in stderr:
+                failure_kind = "conn-refused"
+            elif "command not found" in stderr.lower() or "No such file" in stderr:
+                failure_kind = "command-not-found"
+            else:
+                failure_kind = f"ssh-exit-{exit_code}"
+
+            # Retry once
+            if attempt == 0:
+                time.sleep(1)
+                continue
+            return ProbeResult(
+                exit_code=exit_code,
+                usage_pct=None,
+                usage_str=None,
+                probe_cmd=probe_cmd,
+                failure_kind=failure_kind,
+            )
+
+    # Should not reach here, but just in case
+    return ProbeResult(
+        exit_code=1,
+        usage_pct=None,
+        usage_str=None,
+        probe_cmd=probe_cmd,
+        failure_kind="unknown",
+    )
+
+
+def scan_fleet() -> list[dict]:
+    """Scan all guests and return the results."""
+    results = []
+    for guest in GUESTS:
+        probe_result = probe_guest(guest)
+        guest.probe_result = probe_result
+
+        row = {
+            "target": guest.probe_target,
+            "ct_id": guest.ct_id,
+            "hostname": guest.hostname,
+            "node": guest.node,
+            "access_method": guest.access_method,
+            "reachable": probe_result.is_reachable,
+            "usage_pct": probe_result.usage_pct,
+            "usage_str": probe_result.usage_str,
+            "probe_cmd": probe_result.probe_cmd,
+            "failure_kind": probe_result.failure_kind,
+        }
+        results.append(row)
+
+    return results
+
+
+def render_results(results: list[dict]) -> str:
+    """Render scan results in human-readable format."""
+    lines = []
+    lines.append("=== Disk GC Scan ===")
+    lines.append("")
+
+    for row in results:
+        if row["reachable"]:
+            lines.append(f"  ✅ {row['target']}: {row['usage_str']}")
+            lines.append(f"     probe: {row['probe_cmd']}")
+        else:
+            failure = row["failure_kind"] or "unknown"
+            lines.append(f"  ❌ {row['target']}: UNREACHABLE ({failure})")
+            lines.append(f"     probe: {row['probe_cmd']}")
+
+    return "\n".join(lines)
+
+
+def main() -> int:
+    import argparse
+
+    ap = argparse.ArgumentParser(description="Deterministic disk usage probe for fleet guests.")
+    ap.add_argument("--json", action="store_true", help="machine-readable output")
+    args = ap.parse_args()
+
+    results = scan_fleet()
+
+    if args.json:
+        print(json.dumps(results, indent=2))
+    else:
+        print(render_results(results))
+
+    # Exit 0 if all guests probed (reachable or not), 1 if any probe error
+    # (a probe error means the probe itself failed, not just that the guest was unreachable)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())