diff --git a/.getzilla/getzilla/resource_inventory.py b/.getzilla/getzilla/resource_inventory.py index 0a4b6789..816518c3 100644 --- a/.getzilla/getzilla/resource_inventory.py +++ b/.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/-` while that pid runs. @@ -103,13 +104,27 @@ 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: @@ -117,15 +132,23 @@ def repository_copies(root: Path, depth: int = MAX_COPY_DEPTH) -> list[dict[str, 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} diff --git a/.getzilla/getzilla/review_snapshot.py b/.getzilla/getzilla/review_snapshot.py index adcd27a4..cb69d35b 100644 --- a/.getzilla/getzilla/review_snapshot.py +++ b/.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) diff --git a/scripts/getzilla_status.py b/scripts/getzilla_status.py index a00db01f..8ca40f41 100644 --- a/scripts/getzilla_status.py +++ b/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 diff --git a/tests/test_getzilla_cli.py b/tests/test_getzilla_cli.py index 92483967..596a88ee 100644 --- a/tests/test_getzilla_cli.py +++ b/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' diff --git a/tests/test_resource_inventory.py b/tests/test_resource_inventory.py index c73862a7..e7232dc4 100644 --- a/tests/test_resource_inventory.py +++ b/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') diff --git a/tests/test_review_snapshot.py b/tests/test_review_snapshot.py index 502836a7..33c585c8 100644 --- a/tests/test_review_snapshot.py +++ b/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) diff --git a/tests/test_worktree_safety.py b/tests/test_worktree_safety.py index 2de0bb9c..5350fc38 100644 --- a/tests/test_worktree_safety.py +++ b/tests/test_worktree_safety.py @@ -5,6 +5,7 @@ import subprocess import sys import tempfile +import time import unittest from pathlib import Path from unittest.mock import patch @@ -167,6 +168,8 @@ def test_commits_only_a_worktree_ref_or_the_head_reflog_reaches_keep_it(self) -> commit(kept_ref, 'app.py', 'value = 5\n', 'only in refs/worktree') git(kept_ref, 'update-ref', 'refs/worktree/keep', 'HEAD') git(kept_ref, 'checkout', '-q', '--detach', 'origin/main') + # Only refs/worktree reaches that commit now; the reflog worktree is added after the expiry. + git(kept_ref, 'reflog', 'expire', '--expire=now', '--all') reflog = self.add('reflog', '--detach') commit(reflog, 'app.py', 'value = 6\n', 'only in the reflog') git(reflog, 'checkout', '-q', '--detach', 'origin/main') @@ -175,6 +178,21 @@ def test_commits_only_a_worktree_ref_or_the_head_reflog_reaches_keep_it(self) -> self.assertIn('commits not on any remote-tracking ref', report['reasons'], path) self.assertEqual(remove(self.root, str(path))['result'], 'kept') + def test_a_failed_head_reflog_read_keeps_the_worktree(self) -> None: + from getzilla import worktree_safety + path = self.add('reflog-error', '-b', 'reflog-error') + real_git = worktree_safety._git + + def failing(worktree, *args): + if args[:2] == ('reflog', 'show'): + raise worktree_safety.WorktreeError('reflog unreadable') + return real_git(worktree, *args) + + with patch('getzilla.worktree_safety._git', side_effect=failing): + report = self.report(path) + self.assertFalse(report['removable']) + self.assertIn('check failed: reflog unreadable', report['reasons']) + def test_an_operation_in_progress_keeps_the_worktree(self) -> None: path = self.add('merging', '-b', 'merging') git_dir = Path(git(path, 'rev-parse', '--absolute-git-dir')) @@ -266,8 +284,8 @@ def denied(target, *args, **kwargs): hidden.add(str(os.getpid())) # this test process started before the worktree below older = worktree_safety._started_at(Path('/proc') / str(os.getpid())) self.assertIsNotNone(older) - self.assertFalse(worktree_safety._in_use(path, older + 60)) - self.assertFalse(worktree_safety._in_use(path, older + 10), 'a 2 s margin, not more') + self.assertIs(worktree_safety._in_use(path, older + 60), False) + self.assertIs(worktree_safety._in_use(path, older + 10), False, 'a 2 s margin, not more') with patch('getzilla.worktree_safety._started_at', return_value=None): self.assertIsNone(worktree_safety._in_use(path, older + 60), 'an unknown start fails closed') @@ -295,6 +313,11 @@ def test_the_creation_time_survives_move_and_repair_and_a_clock_step(self) -> No os.utime(git_dir / 'commondir', (future, future)) os.utime(git_dir / 'gitdir', (future, future)) self.assertIsNone(worktree_safety._created_at(path)) + for ahead, trusted in ((5, False), (1, True)): # a 2 s tolerance for clock ticks, not more + stamp = time.time() + ahead + os.utime(git_dir / 'commondir', (stamp, stamp)) + os.utime(git_dir / 'gitdir', (stamp, stamp)) + self.assertEqual(worktree_safety._created_at(path) is not None, trusted, ahead) def test_a_worktree_a_process_works_in_is_kept(self) -> None: path = self.add('busy', '-b', 'busy')