cleanup: close #134 Low findings in inventory, snapshots and worktree checks

.getzilla/getzilla/resource_inventory.py

@@ -3,6 +3,7 @@
 Each `.getzilla/runtime` entry gets one class: `protected` (state, approvals, receipts, evidence,
 worktrees), `ephemeral` (output Getzilla can recreate) or `unknown`. Unknown is never disposable.
 A `..` part, a linked `.getzilla` or runtime folder, or a symlink below it makes the entry unknown.
+The inventory does not read a linked `.getzilla` or runtime folder at all and reports itself incomplete.
 Nothing in this module deletes. `disposable()` is a path check, not a lock: a deleting caller must
 re-check each part through held directory descriptors (`O_NOFOLLOW`, `dir_fd`) and must skip
 entries a live run still owns, such as `par/<stamp>-<pid>` while that pid runs.
@@ -103,29 +104,51 @@ def measure(path: Path, budget: int = MAX_ENTRIES) -> dict[str, Any]:
     return {'bytes': size, 'entries': entries, 'complete': complete}
 
 
-def repository_copies(root: Path, depth: int = MAX_COPY_DEPTH) -> list[dict[str, Any]]:
-    """Repository copies under the runtime folders agents write snapshots to. Links are not followed,
-    and a copy is not searched for nested copies. The copies stay `unknown`: nothing proves them disposable."""
+def _linked_runtime(root: Path) -> bool:
+    """A linked `.getzilla` or runtime folder points the inventory at another tree: it is not read."""
+    return os.path.islink(root / '.getzilla') or os.path.islink(root / RUNTIME)
+
+
+def repository_copies(root: Path, depth: int = MAX_COPY_DEPTH,
+                      budget: int = MAX_ENTRIES) -> tuple[list[dict[str, Any]], bool]:
+    """Repository copies under the runtime folders agents write snapshots to, and whether the search
+    and every size are complete. One entry budget covers the search and all sizes. Links are not
+    followed, and a copy is not searched for nested copies. The copies stay `unknown`: nothing proves
+    them disposable."""
     runtime = root / RUNTIME
     found: list[dict[str, Any]] = []
+    if _linked_runtime(root):
+        return found, False
+    complete = True
     for top in COPY_ROOTS:
-        if os.path.islink(runtime) or os.path.islink(runtime / top) or not (runtime / top).is_dir():
+        if os.path.islink(runtime / top):
+            complete = False
+            continue
+        if not (runtime / top).is_dir():
             continue
         pending = [(runtime / top, 0)]
         while pending:
             folder, level = pending.pop()
             try:
                 with os.scandir(folder) as listing:
                     entries = list(listing)
+                names = {entry.name for entry in entries}
+                children = [Path(entry.path) for entry in entries if entry.is_dir(follow_symlinks=False)]
             except OSError:
+                complete = False
                 continue
-            if COPY_MARKERS <= {entry.name for entry in entries}:
-                found.append({'path': folder.relative_to(root).as_posix(), **measure(folder)})
+            if COPY_MARKERS <= names:  # its own entries are counted once, by its size
+                size = measure(folder, budget)
+                budget -= size['entries']
+                complete = complete and size['complete']
+                found.append({'path': folder.relative_to(root).as_posix(), **size})
                 continue
+            budget -= len(entries)
+            if budget < 0:
+                return sorted(found, key=lambda item: item['path']), False
             if level < depth:
-                pending.extend((Path(entry.path), level + 1) for entry in entries
-                               if entry.is_dir(follow_symlinks=False))
-    return sorted(found, key=lambda item: item['path'])
+                pending.extend((child, level + 1) for child in children)
+    return sorted(found, key=lambda item: item['path']), complete
 
 
 def _disk(path: Path) -> dict[str, int] | None:
@@ -141,8 +164,9 @@ def inventory(root: Path, temp_dir: Path | None = None) -> dict[str, Any]:
     runtime = root / RUNTIME
     temp = Path(temp_dir or tempfile.gettempdir())
     resources: list[dict[str, Any]] = []
+    linked = _linked_runtime(root)
     try:
-        names = sorted(os.listdir(runtime))
+        names = [] if linked else sorted(os.listdir(runtime))
     except OSError:
         names = []
     for name in names:
@@ -159,9 +183,9 @@ def inventory(root: Path, temp_dir: Path | None = None) -> dict[str, Any]:
         group = totals.setdefault(item['class'], {'resources': 0, 'bytes': 0})
         group['resources'] += 1
         group['bytes'] += item['bytes']
-    copies = repository_copies(root)
+    copies, copies_complete = repository_copies(root)
     return {'schema_version': 'getzilla-resource-inventory-v1', 'runtime': str(runtime), 'temp': str(temp),
             'resources': resources, 'totals': totals, 'repository_copies': copies,
-            'complete': all(item['complete'] for item in resources),
+            'complete': copies_complete and all(item['complete'] for item in resources),
             'disk': {'runtime': _disk(runtime if runtime.exists() else root), 'temp': _disk(temp)},
             'deleted': 0}

.getzilla/getzilla/review_snapshot.py

@@ -4,7 +4,8 @@
 detached Git worktree under the main worktree's `.getzilla/runtime/snapshots` (mode 0700), so it
 shares the object store instead of copying it, and writes an owner marker next to it. `remove`
 deletes it again once the report is returned. Edits inside a snapshot are reviewer probes and are
-discarded on purpose; reports never live in a snapshot.
+discarded on purpose; reports never live in a snapshot. A snapshot holds committed content only:
+commit the candidate before review, because staged, unstaged and untracked changes are not copied.
 
 `remove` acts only on a direct child of that folder (no symlink on the way) whose marker names the
 commit the worktree is still detached at. Anything else is kept.
@@ -68,8 +69,10 @@ def remove(root: Path, path: str) -> dict[str, Any]:
     target = Path(os.path.abspath(path))
     if target.parent != Path(os.path.abspath(base)) or target.name.endswith('.json'):
         return {'path': path, 'result': 'kept', 'reason': 'not a direct child of .getzilla/runtime/snapshots'}
-    if target.is_symlink() or Path(os.path.realpath(target)).parent != Path(os.path.realpath(base)):
+    if target.is_symlink():
         return {'path': path, 'result': 'kept', 'reason': 'the snapshot path is a symlink'}  # remove would follow it
+    if Path(os.path.realpath(target)).parent != Path(os.path.realpath(base)):  # a Windows junction, for example
+        return {'path': path, 'result': 'kept', 'reason': 'the snapshot path resolves outside the snapshots folder'}
     marker = _marker(base, target.name)
     entry = next((item for item in list_worktrees(root)
                   if os.path.abspath(item['path']) == str(target)), None)

scripts/getzilla_status.py

@@ -8,9 +8,15 @@
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / '.getzilla'))
 
+import argparse
 import json
 import os
 
+argparse.ArgumentParser(
+    prog='getzilla status',
+    description='Print the active route, change package, agents, human gates and evidence gaps as JSON.',
+).parse_args()
+
 from getzilla.package_status import collect_worktree, inspect_package, receipt_inputs_unavailable
 from getzilla.receipts import validate_evidence
 from getzilla.human_gates import ARTIFACT_NOTICE, gate_statuses

tests/test_getzilla_cli.py

@@ -66,6 +66,13 @@ def test_arguments_pass_through_to_the_script(self) -> None:
         self.assertEqual(proc.returncode, 0, proc.stderr)
         self.assertIn('scaffold', proc.stdout)
 
+    def test_every_command_answers_help_with_usage(self) -> None:
+        for name in cli.COMMANDS:
+            with self.subTest(command=name):
+                proc = run(name, '--help')
+                self.assertEqual(proc.returncode, 0, proc.stderr)
+                self.assertIn('usage:', proc.stdout.lower())
+
     def test_project_scripts_take_precedence_inside_a_project(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             project = Path(tmp) / 'project'

tests/test_resource_inventory.py

@@ -7,11 +7,14 @@
 import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / '.getzilla'))
 
-from getzilla.resource_inventory import EPHEMERAL, PROTECTED, classify, disposable, inventory, measure  # noqa: E402
+from getzilla.resource_inventory import (  # noqa: E402
+    EPHEMERAL, PROTECTED, classify, disposable, inventory, measure, repository_copies,
+)
 
 
 def snapshot(base: Path) -> dict[str, bytes]:
@@ -95,6 +98,16 @@ def test_unreadable_path_parts_fail_closed(self) -> None:
         (self.runtime / 'par' / 'file').write_text('not a directory')
         self.assertEqual(classify(self.root, self.runtime / 'par' / 'file' / 'child'), 'unknown')
 
+    def test_a_missing_runtime_folder_fails_closed(self) -> None:
+        fresh = self.base / 'fresh'
+        fresh.mkdir()
+        self.assertEqual(classify(fresh, fresh / '.getzilla' / 'runtime' / 'par' / 'run-1'), 'unknown')
+        (fresh / '.getzilla').mkdir()
+        self.assertEqual(classify(fresh, fresh / '.getzilla' / 'runtime' / 'par' / 'run-1'), 'unknown')
+
+    def test_a_dot_dot_part_alone_is_unknown(self) -> None:
+        self.assertEqual(classify(self.root, self.runtime / 'par' / '..' / 'par' / 'run-1'), 'unknown')
+
     def test_measure_does_not_follow_links_and_discloses_the_budget(self) -> None:
         big = self.base / 'big'
         big.mkdir()
@@ -148,6 +161,65 @@ def test_repository_copies_are_found_counted_once_and_never_followed_through_lin
         self.assertTrue(all(item['bytes'] > 0 for item in report['repository_copies']))
         self.assertFalse(any(disposable(self.root, self.root / path) for path in paths))
 
+    def test_a_linked_getzilla_runtime_or_copy_folder_is_not_read_and_is_incomplete(self) -> None:
+        self.make_copy(self.runtime / 'scratch' / 'review-1')
+        linked = self.base / 'linked'
+        linked.mkdir()
+        self.link(linked / '.getzilla', self.root / '.getzilla')
+        report = inventory(linked, self.temp)
+        runtime_entries = [item for item in report['resources'] if item['location'] == 'runtime']
+        self.assertEqual((runtime_entries, report['repository_copies'], report['complete']), ([], [], False))
+        runtime_link = self.base / 'runtime-link'
+        (runtime_link / '.getzilla').mkdir(parents=True)
+        self.link(runtime_link / '.getzilla' / 'runtime', self.runtime)
+        self.assertEqual(repository_copies(runtime_link), ([], False))
+        outside = self.make_copy(self.base / 'outside-scratch' / 'copy')
+        self.link(self.runtime / 'archive', outside.parent)
+        copies, complete = repository_copies(self.root)
+        self.assertEqual([item['path'] for item in copies], ['.getzilla/runtime/scratch/review-1'])
+        self.assertFalse(complete)
+        self.assertFalse(inventory(self.root, self.temp)['complete'])
+
+    def test_one_entry_budget_covers_the_search_and_every_copy_size(self) -> None:
+        self.make_copy(self.runtime / 'scratch' / 'review-1')
+        self.make_copy(self.runtime / 'scratch' / 'review-2')
+        copies, complete = repository_copies(self.root)
+        self.assertEqual((len(copies), complete), (2, True))
+        copies, complete = repository_copies(self.root, budget=8)
+        self.assertFalse(complete)
+        self.assertTrue(all(item['entries'] <= 8 for item in copies))
+        self.assertFalse(repository_copies(self.root, budget=1)[1])
+
+    def test_a_search_that_uses_exactly_the_budget_is_complete(self) -> None:
+        (self.runtime / 'scratch' / 'empty').mkdir()
+        self.assertEqual(repository_copies(self.root, budget=1), ([], True))
+        self.assertEqual(repository_copies(self.root, budget=0), ([], False))
+
+    def test_the_budget_left_after_the_search_bounds_the_copy_size(self) -> None:
+        big = self.make_copy(self.runtime / 'scratch' / 'big')
+        for index in range(10):
+            (big / 'scripts' / f'file-{index}.py').write_text('x\n')
+        copies, complete = repository_copies(self.root, budget=100)
+        self.assertTrue(complete)
+        # The scratch listing has 1 entry; the copy counts its own 14 entries once, through its size.
+        self.assertEqual(copies[0]['entries'], 14)
+        self.assertTrue(repository_copies(self.root, budget=15)[1], 'the exact budget is enough')
+        copies, complete = repository_copies(self.root, budget=14)
+        self.assertEqual(([item['path'] for item in copies], complete), (['.getzilla/runtime/scratch/big'], False))
+        self.assertFalse(copies[0]['complete'])
+
+    def test_an_unreadable_copy_folder_marks_the_report_incomplete(self) -> None:
+        (self.runtime / 'scratch' / 'locked').mkdir()
+        real_scandir = os.scandir
+
+        def denied(path):
+            if Path(path).name == 'locked':
+                raise PermissionError('denied')
+            return real_scandir(path)
+
+        with patch('getzilla.resource_inventory.os.scandir', side_effect=denied):
+            self.assertEqual(repository_copies(self.root), ([], False))
+
     def test_cli_lists_repository_copies(self) -> None:
         (self.root / '.git').mkdir()
         self.make_copy(self.runtime / 'scratch' / 'review-9')

tests/test_review_snapshot.py

@@ -128,7 +128,68 @@ def test_an_unknown_commit_leaves_no_marker(self) -> None:
             review_snapshot.create(self.root, 'no-such-commit')
         self.assertFalse(self.snapshots.exists() and any(self.snapshots.iterdir()))
 
+    def test_an_option_shaped_revision_is_not_read_as_an_option(self) -> None:
+        for revision in ('--all', '--git-dir', '-q'):
+            with self.subTest(revision=revision), self.assertRaises(WorktreeError):
+                review_snapshot.create(self.root, revision)
+        self.assertFalse(self.snapshots.exists() and any(self.snapshots.iterdir()))
+
+    def test_an_annotated_tag_is_peeled_to_its_commit(self) -> None:
+        git(self.root, 'tag', '-a', '-m', 'release', 'v1', self.first)
+        path = review_snapshot.create(self.root, 'v1')
+        marker = json.loads((self.snapshots / f'{path.name}.json').read_text())
+        self.assertEqual(marker['commit'], self.first)
+        self.assertEqual(review_snapshot.remove(self.root, str(path))['result'], 'removed')
+
+    def test_a_failed_worktree_add_leaves_no_marker(self) -> None:
+        real_git = review_snapshot._git
+
+        def failing(root, *args):
+            if args[:2] == ('worktree', 'add'):
+                raise WorktreeError('worktree add failed')
+            return real_git(root, *args)
+
+        with patch('getzilla.review_snapshot._git', side_effect=failing), self.assertRaises(WorktreeError):
+            review_snapshot.create(self.root, self.first)
+        self.assertEqual(list(self.snapshots.iterdir()), [])
+
+    def test_a_snapshot_made_from_a_task_worktree_lives_under_the_main_worktree(self) -> None:
+        task = self.root / '.getzilla' / 'runtime' / 'worktrees' / 'task'
+        git(self.root, 'worktree', 'add', '-q', '-b', 'task', str(task))
+        path = review_snapshot.create(task, self.first)
+        self.assertEqual(path.parent, self.snapshots)
+        self.assertFalse((task / '.getzilla' / 'runtime' / 'snapshots').exists())
+        self.assertEqual(review_snapshot.remove(task, str(path))['result'], 'removed')
+
+    def test_a_marker_with_another_schema_is_not_an_owner_marker(self) -> None:
+        path = review_snapshot.create(self.root, self.first)
+        marker = self.snapshots / f'{path.name}.json'
+        marker.write_text(json.dumps({'schema_version': 'other', 'commit': self.first}))
+        result = review_snapshot.remove(self.root, str(path))
+        self.assertEqual((result['result'], result['reason']), ('kept', 'no owner marker'))
+        self.assertTrue(path.is_dir())
+
+    @unittest.skipIf(os.name == 'nt', 'POSIX modes')
+    def test_an_existing_snapshots_folder_is_made_private(self) -> None:
+        self.snapshots.mkdir(parents=True, mode=0o755)
+        os.chmod(self.snapshots, 0o755)
+        review_snapshot.create(self.root, self.first)
+        self.assertEqual(stat.S_IMODE(os.stat(self.snapshots).st_mode), 0o700)
+
+    def test_a_marker_next_to_an_unregistered_folder_is_kept(self) -> None:
+        self.snapshots.mkdir(parents=True)
+        plain = self.snapshots / 'plain'
+        plain.mkdir()
+        (plain / 'work.txt').write_text('unique')
+        marker = self.snapshots / 'plain.json'
+        marker.write_text(json.dumps({'schema_version': review_snapshot.SCHEMA, 'commit': self.first}))
+        result = review_snapshot.remove(self.root, str(plain))
+        self.assertEqual((result['result'], result['reason']), ('kept', 'not a registered snapshot worktree'))
+        self.assertEqual((plain / 'work.txt').read_text(), 'unique')
+        self.assertTrue(marker.is_file())
+
     def test_cli_creates_and_removes_a_snapshot(self) -> None:
+        (self.root / '.getzilla').mkdir(exist_ok=True)  # find_root must stop here, not at an outer project
         cli = [sys.executable, str(ROOT / 'scripts' / 'getzilla_cleanup.py'), 'snapshot']
         created = json.loads(subprocess.run([*cli, self.second, '--json'], cwd=self.root, capture_output=True,
                                             text=True, check=True).stdout)