diff --git a/plugins/disk-hygiene/.claude-plugin/plugin.json b/plugins/disk-hygiene/.claude-plugin/plugin.json index 98da8eeab..25ef977e9 100644 --- a/plugins/disk-hygiene/.claude-plugin/plugin.json +++ b/plugins/disk-hygiene/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json", "name": "disk-hygiene", - "version": "0.9.4", + "version": "0.9.5", "description": "Context-aware disk hygiene for arbitrary directory trees: inventories orphaned and temporary artifacts, classifies evidence into review tiers, and offers exact-path cleanup only after a fresh safety preview and explicit per-tier approval. The target is read-only by default; OS-managed paths, links and mount points, VCS-tracked content, changed entries, and live-handle uncertainty fail closed.", "author": { "name": "Melodic Software", diff --git a/plugins/disk-hygiene/CHANGELOG.md b/plugins/disk-hygiene/CHANGELOG.md index 80eb3d750..58f37b259 100644 --- a/plugins/disk-hygiene/CHANGELOG.md +++ b/plugins/disk-hygiene/CHANGELOG.md @@ -3,6 +3,40 @@ All notable changes to the `disk-hygiene` plugin are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); this plugin uses semantic versioning. +## [0.9.5] + +### Fixed + +- **A silent `destructive_guard.py` launch/runtime failure is now surfaced instead of looking + identical to an approval (#1416).** A repo-operator investigation of the original #1416 report + found both cited launch-refusal root causes already fixed and merged (#1242/0.9.0 here, + repo-hygiene's own guard by #1006); what remained live was that "the guard denied nothing because + it approved" and "the guard denied nothing because it never ran, or ran and died" were + indistinguishable from outside the harness. A new detector, + `skills/clean/scripts/guard_launch_monitor.py`, registers as a second, independent hook in + `hooks/hooks.json` — on `Stop`, not `PreToolUse`/`PostToolUse`, to avoid repeating the per-tool-call + cost class documented in + `docs/adr/0004-rightsize-instruction-surfaces-by-incumbent-first-arbitration.md`'s D-12 — and + scans the session transcript's tail for `hook_non_blocking_error` records naming + `destructive_guard.py`. On a match it emits one `systemMessage` per session (never a block, never a + `permissionDecision`) naming the guard, the failure count, and the most recent failure's exit code, + duration, and truncated stderr. It is a separate, stdlib-only process — deliberately not wired + through the guard's own code, since a guard that cannot launch cannot report that it did not + launch — and fails silently closed on any read/parse error so it can never itself become the + reason a turn is blocked. It covers only `destructive_guard.py`'s own command string: repo-hygiene's + guard is out of scope (verified working separately), there is no retroactive scan of prior + sessions, and — because both hooks are wired with the same literal `python3` command — the + interpreter-resolution fail-open documented in the README (the WindowsApps alias stub, or a + missing/broken `python3`) takes the detector down with the guard, so that one vector stays + unreported until the detector gets a launcher independent of the guarded interpreter (#1504). The + bounded tail read discards its first line only when the retained window actually starts mid-record: + when `size - _MAX_TAIL_BYTES` lands exactly on a record's first byte, an unconditional discard threw + away a whole record — which can be the session's only guard failure, silencing the very report the + detector exists to make. The once-per-session marker is written only after the warning has actually + left the process (`print` then `flush`, then mark): marking first meant a closed pipe or a kill + between the two silenced every later `Stop` in the session while the broad never-fail-loudly handler + exited quietly — reinstating the silence the detector exists to break. + ## [0.9.4] ### Fixed diff --git a/plugins/disk-hygiene/README.md b/plugins/disk-hygiene/README.md index 09a951ac1..4ff4b4326 100644 --- a/plugins/disk-hygiene/README.md +++ b/plugins/disk-hygiene/README.md @@ -64,6 +64,18 @@ at preview. Backups remain the recovery boundary for user data. commands are denied after a clean run ends, start a new session and see that issue. PreToolUse hooks also fire inside subagents, so fanned-out workers run under the same guards. The plugin never downloads a runtime. +- **A silent engine-gate launch/runtime failure is now surfaced (since 0.9.5, #1416).** A `Stop`-event + detector (`skills/clean/scripts/guard_launch_monitor.py`, a second hook entry in `hooks/hooks.json`, + independent of the engine-gate guard itself) scans the session transcript for + `hook_non_blocking_error` records naming the engine gate's own command string and warns once per + session with the failure count and the most recent failure's exit code, duration, and stderr — so a + guard that never ran or died mid-run no longer looks identical to a guard that ran and approved. + This covers only the `destructive_guard.py` command string in the current session's transcript: it + does not cover repo-hygiene's own guard (a separate plugin, verified working independently), and it + never retroactively scans a prior session's transcript. It is also wired with the same literal + `python3` command as the guard it watches, so the interpreter-resolution failure below — the + WindowsApps alias stub, or a missing/broken `python3` — takes the detector down with the guard and + goes unreported (#1504). - Git is optional for ordinary trees. If a target contains or sits inside a Git worktree, Git becomes required so tracked content can be proven safe; otherwise cleanup for that subtree is blocked. - Windows has the full **audit** lane (Python 3.11's `lstat` reparse metadata plus Win32 APIs diff --git a/plugins/disk-hygiene/hooks/hooks.json b/plugins/disk-hygiene/hooks/hooks.json index 9b4396bcb..b9be079fe 100644 --- a/plugins/disk-hygiene/hooks/hooks.json +++ b/plugins/disk-hygiene/hooks/hooks.json @@ -19,6 +19,23 @@ } ] } + ], + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "python3", + "args": [ + "${CLAUDE_PLUGIN_ROOT}/skills/clean/scripts/guard_launch_monitor.py", + "--data-root", + "${CLAUDE_PLUGIN_DATA}" + ], + "timeout": 20, + "statusMessage": "Checking for silent destructive-guard failures..." + } + ] + } ] } } diff --git a/plugins/disk-hygiene/skills/clean/reference/safety-model.md b/plugins/disk-hygiene/skills/clean/reference/safety-model.md index 8c924ec1f..428d79d53 100644 --- a/plugins/disk-hygiene/skills/clean/reference/safety-model.md +++ b/plugins/disk-hygiene/skills/clean/reference/safety-model.md @@ -202,6 +202,31 @@ Even when the switch resolves enabled, the PowerShell lane is a raised bar, not mutation spelling passes it, so the engine's own containment, revalidation, and platform gates remain the deletion authority. +**Guard launch/runtime failures are now surfaced, not silently indistinguishable from approval (since +0.9.5, #1416).** A `PreToolUse` hook that fails to launch, or launches and then exits non-zero, denies +nothing — Claude Code treats a non-blocking hook result as approval, so "the guard denied nothing because +it approved" and "the guard denied nothing because it never ran, or ran and silently died" looked +identical from outside the harness. `skills/clean/scripts/guard_launch_monitor.py` closes that gap with a +second, independent hook registered on `Stop` in `hooks/hooks.json` (deliberately not `PreToolUse`, to +avoid taxing every guarded tool call the way `docs/adr/0004-...`'s D-12 did): it scans the session +transcript's tail for `hook_non_blocking_error` records whose command string names +`destructive_guard.py`, and if it finds any, emits a `systemMessage` — never a block, never a +`permissionDecision` — naming the guard, the total failure count, and the most recent failure's exit +code, duration, and truncated stderr, at most once per session. It is a separate, stdlib-only process +that imports nothing from the guard: a guard that cannot launch cannot report that it did not launch, so +the detector cannot depend on the guard's own code path, and it fails silently closed (exit 0, no output) +on any transcript-read or parse error so it can never itself become the reason a turn is blocked. What +this does **not** cover: repo-hygiene ships its own, structurally different guard, verified working +independently and out of scope here; the detector's command-substring filter matches only +`destructive_guard.py` invocations, so a renamed or unrelated guard script is invisible to it the same +way it is invisible to the engine gate's own coverage marker (see above); it never retroactively +scans a prior session's transcript — only the transcript named by the current `Stop` event's own +`transcript_path`; and it is wired with the same literal `python3` command as the guard it watches, +so the interpreter-resolution fail-open the plugin README documents (the WindowsApps +`python3.exe` alias stub, or a missing/broken `python3`) takes the detector down with the guard and +leaves that one vector unreported. Closing that requires a launcher whose availability does not +depend on the same lookup, which is tracked separately (#1504). + A depth-limited scan records every directory it declined to enter in `truncated_paths`. Truncated directories have no captured descendant set, so the preview blocks them (and anything beneath them) as `truncated-not-inventoried`; they are coverage gaps, never candidates. diff --git a/plugins/disk-hygiene/skills/clean/scripts/guard_launch_monitor.py b/plugins/disk-hygiene/skills/clean/scripts/guard_launch_monitor.py new file mode 100755 index 000000000..329237293 --- /dev/null +++ b/plugins/disk-hygiene/skills/clean/scripts/guard_launch_monitor.py @@ -0,0 +1,278 @@ +#!/usr/bin/env python3 +"""Surface silent ``destructive_guard.py`` launch/runtime failures (#1416). + +A `PreToolUse` hook that fails to launch, or launches and then exits +non-zero, denies nothing — Claude Code treats a non-blocking hook result as +approval and lets the guarded Bash/PowerShell command proceed ungated (the +0.6.3 fail-open shape; see ``reference/safety-model.md``). From outside the +harness "the guard denied nothing because it approved" and "the guard denied +nothing because it never ran, or ran and silently died" look identical. This +module closes that observability gap for one specific, narrow signal: the +engine-gate guard invocation (``destructive_guard.py``) recorded as a +``hook_non_blocking_error`` attachment in the session transcript. + +This is a *detector*, not a guard, and it must never behave like one: + +- It never emits ``permissionDecision`` and never blocks a tool call — it is + registered on the ``Stop`` event (see below), which has no blocking + vocabulary for this purpose to begin with; the only output it ever emits is + ``{"systemMessage": "..."}``. +- On any failure to read or parse the transcript it exits 0 with no stdout. + A detector that fails loudly for its own sake is exactly the kind of + second-order fragility this issue exists to avoid: it must never itself + become the reason a turn is blocked or a tool call denied. +- It is a separate process from the guard, deliberately: a guard that cannot + launch cannot report that it did not launch, so the detector cannot be + wired through the guard's own code path. It imports nothing from + ``destructive_guard.py`` or ``lib/`` — stdlib only — so its own failure + surface stays near zero. + +Why ``Stop``, not ``PreToolUse``/``PostToolUse`` +================================================= +This repo has already paid for a hook that runs on every matching tool call: +``docs/adr/0004-rightsize-instruction-surfaces-by-incumbent-first-arbitration.md`` +documents D-12, a guardrails ``PreToolUse`` hook that cost 12-19s p50 on every +single Bash call across roughly 1,464 runs in six days. A new hook wired to +``PreToolUse``/``PostToolUse`` with a Bash/PowerShell matcher would repeat +that risk class on every guarded command, forever, to catch a failure that - +by construction - is already sitting in the transcript file by the time the +guarded command's hook step in the docs' three-cadence classification +(once-per-session / once-per-turn / once-per-tool-call) completes. ``Stop`` +fires once per turn: the guard, if it ran, ran synchronously before the +guarded command executed, so its failure record (if any) is already appended +to the transcript well before the turn ends. Once-per-turn cadence catches +the failure just as promptly as once-per-tool-call would, at a small +fraction of the invocation count. + +Bounded-cost transcript read +============================= +Transcript files grow for the life of a session and are written +asynchronously (may lag the in-memory conversation — see the hooks docs). +Reading the whole file on every ``Stop`` would scale invocation cost with +session length, the same shape of problem D-12 created a different way. This +module instead seeks to the last ``_MAX_TAIL_BYTES`` of the file, discards +the (likely-truncated) first partial line, and parses what remains — cost is +O(cap), not O(session length), regardless of how large the transcript gets. +Once a warning has fired for a session, the once-per-session marker (see +below) short-circuits *before* the read, so the amortized per-turn cost for +the rest of a long session is a single stat() of a small marker file. + +Once-per-session marker +======================== +Keyed by the hook's own ``session_id`` input field (never anything found +*inside* transcript records — a transcript can carry attachment records +whose own ``session_id`` field differs from the session that produced the +file; only the hook's own stdin input names the current session reliably). +Primary location is a marker file under ``${CLAUDE_PLUGIN_DATA}`` (passed +via ``--data-root``); this is a plugin-level ``hooks/hooks.json`` +registration, so unlike a skill-frontmatter hook, ``${CLAUDE_PLUGIN_DATA}`` +substitutes directly here — no need for ``destructive_guard.py``'s +``--plugin-root`` derivation dance (a literal unsubstituted +``${CLAUDE_PLUGIN_DATA}`` token is still treated as absent, matching that +module's placeholder idiom). Fallback is a subdirectory under +``tempfile.gettempdir()``. The bookkeeping degrades toward *re-warning*, +never toward silence: if both the primary and fallback marker writes fail, +the warning is still emitted this run rather than suppressed — over-warning +is the safe failure direction for a module that exists specifically to kill +a silent-suppression defect class. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from pathlib import Path + +_MAX_TAIL_BYTES = 2_000_000 + +_GUARD_COMMAND_SUBSTRING = "destructive_guard.py" +_GUARD_DISPLAY_NAME = "destructive_guard.py" + +_DATA_ROOT_FLAG = "--data-root" +_CLAUDE_PLUGIN_DATA_ENV = "CLAUDE_PLUGIN_DATA" +_DATA_ROOT_PLACEHOLDER = f"${{{_CLAUDE_PLUGIN_DATA_ENV}}}" + +_MARKER_DIRNAME = "guard-launch-monitor" +_STDERR_TRUNCATE_CHARS = 300 + + +def _argv_flag_value(argv: list[str], flag: str) -> str | None: + for index, item in enumerate(argv): + if item == flag and index + 1 < len(argv): + return argv[index + 1] + if item.startswith(flag + "="): + return item.split("=", 1)[1] + return None + + +def _resolve_data_root(argv: list[str]) -> str | None: + """Resolve the plugin's persistent data root, or None if unavailable. + + A literal unsubstituted ``${CLAUDE_PLUGIN_DATA}`` placeholder is treated + as absent, matching ``destructive_guard.py``'s placeholder idiom (see + module docstring) — this module does not import that idiom, only mirrors + it, per the issue's stdlib-only requirement. + """ + direct = _argv_flag_value(argv, _DATA_ROOT_FLAG) + if direct and direct != _DATA_ROOT_PLACEHOLDER: + return direct + env_value = os.environ.get(_CLAUDE_PLUGIN_DATA_ENV) + if env_value and env_value != _DATA_ROOT_PLACEHOLDER: + return env_value + return None + + +def _marker_path_candidates(data_root: str | None, session_id: str) -> list[Path]: + safe_session = "".join( + ch if (ch.isalnum() or ch in "-_") else "_" for ch in session_id + ) or "unknown-session" + candidates: list[Path] = [] + if data_root: + candidates.append( + Path(data_root) / _MARKER_DIRNAME / f"{safe_session}.warned" + ) + candidates.append( + Path(tempfile.gettempdir()) + / "disk-hygiene-guard-launch-monitor" + / f"{safe_session}.warned" + ) + return candidates + + +def _already_warned(marker_paths: list[Path]) -> bool: + for path in marker_paths: + try: + if path.is_file(): + return True + except OSError: + continue + return False + + +def _write_marker(marker_paths: list[Path]) -> None: + """Best-effort write. Failure here must never suppress a real finding.""" + for path in marker_paths: + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("1", encoding="utf-8") + return + except OSError: + continue + + +def _read_tail(transcript_path: str) -> str: + path = Path(transcript_path) + size = path.stat().st_size + with path.open("rb") as handle: + if size > _MAX_TAIL_BYTES: + offset = size - _MAX_TAIL_BYTES + # A newline immediately before the window means the window already + # starts on a record boundary; discarding then would throw away a + # whole record, which can be the only guard failure in the tail. + handle.seek(offset - 1) + starts_mid_record = handle.read(1) != b"\n" + if starts_mid_record: + handle.readline() # discard the truncated partial first line + raw = handle.read() + return raw.decode("utf-8", errors="replace") + + +def _iter_guard_failures(transcript_text: str): + for line in transcript_text.splitlines(): + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue + if not isinstance(record, dict): + continue + if record.get("type") != "attachment": + continue + attachment = record.get("attachment") + if not isinstance(attachment, dict): + continue + if attachment.get("type") != "hook_non_blocking_error": + continue + command = attachment.get("command") + if not isinstance(command, str) or _GUARD_COMMAND_SUBSTRING not in command: + continue + yield record, attachment + + +def _format_stderr(stderr: object) -> str: + if not isinstance(stderr, str) or not stderr.strip(): + return "(no stderr output)" + text = stderr.strip() + if len(text) > _STDERR_TRUNCATE_CHARS: + return text[:_STDERR_TRUNCATE_CHARS] + "... (truncated)" + return text + + +def _build_message(failures: list[tuple[dict, dict]]) -> str: + count = len(failures) + # Records are scanned in file order (chronological); the last match is + # the most recent failure. + _, latest = failures[-1] + exit_code = latest.get("exitCode") + duration_ms = latest.get("durationMs") + stderr_text = _format_stderr(latest.get("stderr")) + plural = "s" if count != 1 else "" + return ( + f"disk-hygiene: {_GUARD_DISPLAY_NAME} failed to run or exited " + f"non-zero {count} time{plural} this session and its failure(s) were " + "not visible as a denial. Most recent failure: " + f"exitCode: {exit_code}, durationMs: {duration_ms}, stderr: {stderr_text} " + "This means destructive-action review may not have been enforced for " + "the guarded command(s) in question. This detector covers only " + f"{_GUARD_DISPLAY_NAME}'s own command string in this session's " + "transcript; it does not cover repo-hygiene's guard (verified " + "working separately) and does not retroactively scan past sessions." + ) + + +def _run(hook_input: dict, data_root: str | None) -> tuple[str, list[Path]] | None: + """Return the warning and where to record it, or ``None`` if there is nothing to say. + + Recording is the caller's job, deliberately: the marker suppresses every + later ``Stop`` in the session, so it must not be written until the warning + has actually left the process. + """ + transcript_path = hook_input.get("transcript_path") + session_id = hook_input.get("session_id") or "unknown-session" + marker_paths = _marker_path_candidates(data_root, str(session_id)) + if _already_warned(marker_paths): + return None + if not transcript_path or not isinstance(transcript_path, str): + return None + transcript_text = _read_tail(transcript_path) + failures = list(_iter_guard_failures(transcript_text)) + if not failures: + return None + return _build_message(failures), marker_paths + + +def main(argv: list[str] | None = None) -> int: + argv = sys.argv[1:] if argv is None else argv + try: + data_root = _resolve_data_root(argv) + raw_input = sys.stdin.read() + hook_input = json.loads(raw_input) if raw_input.strip() else {} + if not isinstance(hook_input, dict): + return 0 + found = _run(hook_input, data_root) + if found: + message, marker_paths = found + print(json.dumps({"systemMessage": message})) + sys.stdout.flush() + _write_marker(marker_paths) + return 0 + except BaseException: # noqa: BLE001 - detector must never fail loudly + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/plugins/disk-hygiene/skills/clean/scripts/guard_launch_monitor.test.sh b/plugins/disk-hygiene/skills/clean/scripts/guard_launch_monitor.test.sh new file mode 100755 index 000000000..06f01a904 --- /dev/null +++ b/plugins/disk-hygiene/skills/clean/scripts/guard_launch_monitor.test.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# Cross-platform contract wrapper for the guard-launch-monitor test suite. +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# The Python floor has one origin: MIN_PYTHON in the clean engine. Parse it +# rather than restating the number here. +ENGINE="$SCRIPT_DIR/hygiene.py" +FLOOR="$(sed -n 's/^MIN_PYTHON = (\([0-9]*\), \([0-9]*\)).*/\1.\2/p' "$ENGINE")" +if [[ -z "$FLOOR" ]]; then + echo "FAIL: could not parse MIN_PYTHON from $ENGINE" >&2 + exit 1 +fi + +if command -v python >/dev/null 2>&1; then + PYTHON=python +elif command -v python3 >/dev/null 2>&1; then + PYTHON=python3 +else + echo "SKIP: Python ${FLOOR}+ not found" >&2 + exit 0 +fi + +"$PYTHON" -c "import sys; floor = tuple(int(part) for part in '$FLOOR'.split('.')); raise SystemExit(0 if sys.version_info >= floor else 1)" || { + echo "SKIP: Python ${FLOOR}+ required" >&2 + exit 0 +} +"$PYTHON" -m unittest -v "$SCRIPT_DIR/test_guard_launch_monitor.py" diff --git a/plugins/disk-hygiene/skills/clean/scripts/test_guard_launch_monitor.py b/plugins/disk-hygiene/skills/clean/scripts/test_guard_launch_monitor.py new file mode 100755 index 000000000..fe73fad2d --- /dev/null +++ b/plugins/disk-hygiene/skills/clean/scripts/test_guard_launch_monitor.py @@ -0,0 +1,424 @@ +#!/usr/bin/env python3 +"""Behavioral tests for the guard-launch-visibility detector (#1416). + +Fixtures below are synthetic, minimal JSONL shaped like real +``hook_non_blocking_error`` transcript attachment records (schema confirmed +against real local transcripts this session — see the module docstring of +``guard_launch_monitor.py``) but contain only the fields the detector +actually reads. No real transcript file, session id, or path is committed +here. +""" + +from __future__ import annotations + +import importlib.util +import io +import json +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest import mock + +SCRIPT_DIR = Path(__file__).resolve().parent + + +def load_module(name: str, filename: str): + spec = importlib.util.spec_from_file_location(name, SCRIPT_DIR / filename) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +monitor = load_module("guard_launch_monitor", "guard_launch_monitor.py") + + +_GUARD_COMMAND = ( + "python3 ${CLAUDE_PLUGIN_ROOT}/skills/clean/scripts/destructive_guard.py " + "--mode engine-gate --plugin-root ${CLAUDE_PLUGIN_ROOT} " + "--authorized-data-root ${CLAUDE_PLUGIN_DATA} " + "--disk-hygiene-enabled ${user_config.disk_hygiene_enabled}" +) + +_OTHER_HOOK_COMMAND = "python3 /some/other/plugin/scripts/other_hook.py --mode check" + + +def _record( + *, + hook_event="PreToolUse", + command=_GUARD_COMMAND, + stderr="", + stdout="", + exit_code=1, + duration_ms=11, + record_session_id="unrelated-record-session", + attachment_type="hook_non_blocking_error", + record_type="attachment", +) -> str: + payload = { + "parentUuid": "54fcf5e2-dc5f-4e29-accf-7a8489af9906", + "isSidechain": False, + "attachment": { + "type": attachment_type, + "hookName": f"{hook_event}:Bash", + "toolUseID": "toolu_01Xj2c8BgDPM7tC75ZyvJMJ4", + "hookEvent": hook_event, + "stderr": stderr, + "stdout": stdout, + "exitCode": exit_code, + "command": command, + "durationMs": duration_ms, + }, + "type": record_type, + "uuid": "c8e67437-ea97-444a-821a-e6e062e36f32", + "timestamp": "2026-07-23T15:41:16.178Z", + "session_id": record_session_id, + "userType": "external", + "entrypoint": "cli", + "cwd": "C:\\Projects\\melodic\\claude-code-plugins", + "sessionId": "10bcff28-b64c-490d-8848-de5d6c76a43a", + "version": "2.1.218", + "gitBranch": "feat/example", + } + return json.dumps(payload) + + +def _filler_line(nbytes: int) -> str: + """An ignorable ASCII JSONL record occupying exactly ``nbytes`` with its newline.""" + prefix = '{"type": "other", "noise": "' + suffix = '"}' + padding = nbytes - 1 - len(prefix) - len(suffix) + assert padding >= 0, "nbytes too small to hold a record" + return prefix + "y" * padding + suffix + + +class GuardLaunchMonitorTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.data_root = Path(self.tmp.name) / "data-root" + self.transcript_path = Path(self.tmp.name) / "transcript.jsonl" + + # Isolate every test from real machine state: the fallback marker + # location is tempfile.gettempdir() and the data-root resolution + # falls back to the real CLAUDE_PLUGIN_DATA env var when no + # --data-root is supplied (test_unsubstituted_data_root_placeholder_ + # treated_as_absent exercises exactly that "no usable data root" + # path). Without this isolation those tests would write a + # once-per-session marker into this machine's real temp dir (or, + # worse, its real plugin data directory if CLAUDE_PLUGIN_DATA + # happens to be exported) and never clean it up, making a second + # run of the suite on the same machine spuriously fail. + fake_tempdir = Path(self.tmp.name) / "fake-system-tempdir" + fake_tempdir.mkdir() + gettempdir_patcher = mock.patch.object( + monitor.tempfile, "gettempdir", return_value=str(fake_tempdir) + ) + gettempdir_patcher.start() + self.addCleanup(gettempdir_patcher.stop) + + env_patcher = mock.patch.dict( + "os.environ", {"CLAUDE_PLUGIN_DATA": ""}, clear=False + ) + env_patcher.start() + self.addCleanup(env_patcher.stop) + + def write_transcript(self, lines: list[str]) -> None: + self.transcript_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + def run_monitor( + self, session_id: str = "session-1", data_root: Path | None = None + ) -> str | None: + hook_input = { + "session_id": session_id, + "transcript_path": str(self.transcript_path), + } + argv = [] + root = self.data_root if data_root is None else data_root + if root is not None: + argv = ["--data-root", str(root)] + stdout = io.StringIO() + with mock.patch.object(monitor.sys, "stdin", io.StringIO(json.dumps(hook_input))): + with redirect_stdout(stdout): + exit_code = monitor.main(argv) + self.assertEqual(0, exit_code) + text = stdout.getvalue().strip() + if not text: + return None + parsed = json.loads(text) + return parsed.get("systemMessage") + + # -- criterion 3: the #1423 discriminating shape ----------------------- + + def test_1423_shape_states_exit_code_and_duration_explicitly(self) -> None: + self.write_transcript( + [ + _record( + stderr="", + exit_code=1, + duration_ms=17054, + ) + ] + ) + message = self.run_monitor() + self.assertIsNotNone(message) + self.assertIn("destructive_guard.py", message) + self.assertIn("exitCode: 1", message) + self.assertIn("durationMs: 17054", message) + self.assertNotIn("no stderr", message.lower().replace("(no stderr output)", "")) + + def test_empty_stderr_renders_explicit_placeholder(self) -> None: + self.write_transcript([_record(stderr="", exit_code=1, duration_ms=17054)]) + message = self.run_monitor() + self.assertIn("(no stderr output)", message) + + # -- criterion 4: launch-refusal shape ----------------------------------- + + def test_launch_refusal_shape_warns(self) -> None: + self.write_transcript( + [ + _record( + stderr=( + 'Failed to run: Plugin option "disk_hygiene_enabled" ' + "isn't set. Open /plugin manage to configure it, or " + "check that the plugin's userConfig schema declares " + '"disk_hygiene_enabled".' + ), + exit_code=1, + duration_ms=11, + ) + ] + ) + message = self.run_monitor() + self.assertIsNotNone(message) + self.assertIn("Failed to run", message) + self.assertIn("disk_hygiene_enabled", message) + self.assertIn("exitCode: 1", message) + self.assertIn("durationMs: 11", message) + + # -- criterion 5: no false positives ------------------------------------- + + def test_clean_session_with_other_hook_failure_emits_nothing(self) -> None: + self.write_transcript( + [ + _record( + command=_OTHER_HOOK_COMMAND, + stderr="some other hook failed", + exit_code=1, + duration_ms=42, + ) + ] + ) + message = self.run_monitor() + self.assertIsNone(message) + + def test_fully_clean_session_emits_nothing(self) -> None: + self.transcript_path.write_text("", encoding="utf-8") + message = self.run_monitor() + self.assertIsNone(message) + + # -- criterion 7: unreadable transcript degrades silently --------------- + + def test_unreadable_transcript_path_exits_quietly(self) -> None: + missing = Path(self.tmp.name) / "does-not-exist.jsonl" + hook_input = { + "session_id": "session-x", + "transcript_path": str(missing), + } + stdout = io.StringIO() + with mock.patch.object( + monitor.sys, "stdin", io.StringIO(json.dumps(hook_input)) + ): + with redirect_stdout(stdout): + exit_code = monitor.main(["--data-root", str(self.data_root)]) + self.assertEqual(0, exit_code) + self.assertEqual("", stdout.getvalue().strip()) + + def test_missing_transcript_path_field_exits_quietly(self) -> None: + hook_input = {"session_id": "session-y"} + stdout = io.StringIO() + with mock.patch.object( + monitor.sys, "stdin", io.StringIO(json.dumps(hook_input)) + ): + with redirect_stdout(stdout): + exit_code = monitor.main(["--data-root", str(self.data_root)]) + self.assertEqual(0, exit_code) + self.assertEqual("", stdout.getvalue().strip()) + + def test_malformed_stdin_exits_quietly(self) -> None: + stdout = io.StringIO() + with mock.patch.object(monitor.sys, "stdin", io.StringIO("{not json")): + with redirect_stdout(stdout): + exit_code = monitor.main(["--data-root", str(self.data_root)]) + self.assertEqual(0, exit_code) + self.assertEqual("", stdout.getvalue().strip()) + + # -- criterion 8: once-per-session cap ----------------------------------- + + def test_second_invocation_same_session_is_suppressed(self) -> None: + self.write_transcript([_record(exit_code=1, duration_ms=17054)]) + first = self.run_monitor(session_id="same-session") + self.assertIsNotNone(first) + second = self.run_monitor(session_id="same-session") + self.assertIsNone(second) + + def test_different_session_ids_each_warn_once(self) -> None: + self.write_transcript([_record(exit_code=1, duration_ms=17054)]) + first = self.run_monitor(session_id="session-a") + second = self.run_monitor(session_id="session-b") + self.assertIsNotNone(first) + self.assertIsNotNone(second) + + def test_marker_write_failure_still_emits_warning_this_run(self) -> None: + self.write_transcript([_record(exit_code=1, duration_ms=17054)]) + with mock.patch.object( + monitor.Path, "write_text", side_effect=OSError("read-only fs") + ): + message = self.run_monitor(session_id="session-unwritable") + self.assertIsNotNone(message) + self.assertIn("exitCode: 1", message) + + def test_undelivered_warning_is_not_marked_as_warned(self) -> None: + """A delivery that never reached the runner must stay repeatable. + + Marking before delivery would let one broken pipe suppress every later + `Stop` in the session — the exact silence this detector exists to break. + """ + self.write_transcript([_record(exit_code=1, duration_ms=17054)]) + hook_input = { + "session_id": "session-broken-pipe", + "transcript_path": str(self.transcript_path), + } + argv = ["--data-root", str(self.data_root)] + + class _BrokenStdout(io.StringIO): + def flush(self) -> None: + raise BrokenPipeError("hook runner closed the pipe") + + with mock.patch.object(monitor.sys, "stdin", io.StringIO(json.dumps(hook_input))): + with redirect_stdout(_BrokenStdout()): + self.assertEqual(0, monitor.main(argv)) + + retry = self.run_monitor(session_id="session-broken-pipe") + self.assertIsNotNone(retry) + self.assertIn("exitCode: 1", retry) + + # -- data-root placeholder handling -------------------------------------- + + def test_unsubstituted_data_root_placeholder_treated_as_absent(self) -> None: + self.write_transcript([_record(exit_code=1, duration_ms=17054)]) + hook_input = { + "session_id": "session-placeholder", + "transcript_path": str(self.transcript_path), + } + stdout = io.StringIO() + with mock.patch.object( + monitor.sys, "stdin", io.StringIO(json.dumps(hook_input)) + ): + with redirect_stdout(stdout): + exit_code = monitor.main( + ["--data-root", "${CLAUDE_PLUGIN_DATA}"] + ) + self.assertEqual(0, exit_code) + text = stdout.getvalue().strip() + self.assertTrue(text) + parsed = json.loads(text) + self.assertIn("exitCode: 1", parsed["systemMessage"]) + + # -- multiple failures: count and most-recent selection ------------------ + + def test_multiple_failures_report_count_and_most_recent(self) -> None: + self.write_transcript( + [ + _record(exit_code=1, duration_ms=100, stderr="first failure"), + _record(exit_code=2, duration_ms=200, stderr="second failure"), + ] + ) + message = self.run_monitor() + self.assertIn("2 time", message) + self.assertIn("exitCode: 2", message) + self.assertIn("durationMs: 200", message) + self.assertIn("second failure", message) + + # -- bounded tail read ---------------------------------------------------- + + def test_tail_read_still_finds_failure_near_end_of_large_transcript(self) -> None: + filler = json.dumps({"type": "other", "noise": "x" * 200}) + lines = [filler for _ in range(20000)] + lines.append(_record(exit_code=1, duration_ms=17054)) + self.write_transcript(lines) + self.assertGreater(self.transcript_path.stat().st_size, monitor._MAX_TAIL_BYTES) + message = self.run_monitor() + self.assertIsNotNone(message) + self.assertIn("exitCode: 1", message) + self.assertIn("durationMs: 17054", message) + + def test_tail_window_starting_on_a_record_boundary_keeps_that_record(self) -> None: + """The retained window can begin exactly at a record's first byte. + + Discarding unconditionally would drop a whole record there, and that + record can be the session's only guard failure. + """ + failure = _record(exit_code=9, duration_ms=4242, stderr="boundary failure") + pad = _filler_line(monitor._MAX_TAIL_BYTES - len(failure) - 1) + head = [_filler_line(4096) for _ in range(4)] + # Written as bytes with explicit LF: the byte offsets are the fixture's + # whole point, and text mode would insert a platform line ending. + body = "\n".join(head + [failure, pad]) + "\n" + self.transcript_path.write_bytes(body.encode("utf-8")) + + size = self.transcript_path.stat().st_size + self.assertEqual( + size - monitor._MAX_TAIL_BYTES, + sum(len(line) + 1 for line in head), + "fixture must place the window boundary exactly at the failure record", + ) + + message = self.run_monitor() + self.assertIsNotNone(message) + self.assertIn("exitCode: 9", message) + self.assertIn("durationMs: 4242", message) + + def test_non_attachment_records_are_ignored(self) -> None: + self.write_transcript( + [ + json.dumps({"type": "user", "message": "hello"}), + json.dumps({"type": "assistant", "message": "hi"}), + ] + ) + message = self.run_monitor() + self.assertIsNone(message) + + def test_malformed_json_line_does_not_abort_scan(self) -> None: + self.write_transcript( + [ + "{not valid json at all", + _record(exit_code=1, duration_ms=17054), + ] + ) + message = self.run_monitor() + self.assertIsNotNone(message) + self.assertIn("exitCode: 1", message) + + def test_no_permission_decision_or_block_ever_emitted(self) -> None: + self.write_transcript([_record(exit_code=1, duration_ms=17054)]) + hook_input = { + "session_id": "session-safety", + "transcript_path": str(self.transcript_path), + } + stdout = io.StringIO() + with mock.patch.object( + monitor.sys, "stdin", io.StringIO(json.dumps(hook_input)) + ): + with redirect_stdout(stdout): + exit_code = monitor.main(["--data-root", str(self.data_root)]) + self.assertEqual(0, exit_code) + parsed = json.loads(stdout.getvalue().strip()) + self.assertNotIn("permissionDecision", parsed) + self.assertNotIn("decision", parsed) + self.assertEqual({"systemMessage"}, set(parsed.keys())) + + +if __name__ == "__main__": + unittest.main()