fix(disk-hygiene): scan transcript head for guard-launch failures

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.17.3",
+  "version": "0.17.4",
   "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",

plugins/disk-hygiene/CHANGELOG.md

@@ -3,6 +3,13 @@
 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.17.4]
+
+### Fixed
+
+- **Guard-launch detector scans transcript head plus tail (#1514).** Guard failures that fall
+  outside the 2 MB tail window are no longer lost when a later turn appends a large record.
+
 ## [0.17.3]
 
 ### Fixed

plugins/disk-hygiene/skills/clean/scripts/guard_launch_monitor.py

@@ -162,21 +162,36 @@ def _write_marker(marker_paths: list[Path]) -> None:
             continue
 
 
+def _read_tail_from_handle(handle, *, size: int) -> str:
+    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 _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")
+        if size <= _MAX_TAIL_BYTES:
+            return handle.read().decode("utf-8", errors="replace")
+        # Guard failures can land in the head region when a later turn appends
+        # more than _MAX_TAIL_BYTES after them (#1514). Scan complete JSONL
+        # records from the head plus the capped tail window.
+        head_limit = size - _MAX_TAIL_BYTES
+        head_raw = handle.read(head_limit)
+        if head_raw and not head_raw.endswith(b"\n"):
+            head_raw = head_raw.rsplit(b"\n", 1)[0] + b"\n"
+        head_text = head_raw.decode("utf-8", errors="replace")
+        tail_text = _read_tail_from_handle(handle, size=size - head_limit)
+    return head_text + tail_text
 
 
 def _iter_guard_failures(transcript_text: str):

plugins/disk-hygiene/skills/clean/scripts/test_guard_launch_monitor.py

@@ -380,6 +380,17 @@ def test_tail_window_starting_on_a_record_boundary_keeps_that_record(self) -> No
         self.assertIn("exitCode: 9", message)
         self.assertIn("durationMs: 4242", message)
 
+    def test_head_region_failure_survives_later_oversized_append(self) -> None:
+        """#1514: a guard failure more than _MAX_TAIL_BYTES from EOF must still warn."""
+        failure = _record(exit_code=2, duration_ms=99, stderr="head failure")
+        pad = _filler_line(monitor._MAX_TAIL_BYTES + 100_000)
+        self.write_transcript([failure, pad])
+        self.assertGreater(self.transcript_path.stat().st_size, monitor._MAX_TAIL_BYTES)
+        message = self.run_monitor()
+        self.assertIsNotNone(message)
+        self.assertIn("exitCode: 2", message)
+        self.assertIn("head failure", message)
+
     def test_non_attachment_records_are_ignored(self) -> None:
         self.write_transcript(
             [