fix: #134 slice B — runner verdicts, report escaping, protected-write session

.getzilla/getzilla/python_test_runner.py

@@ -337,9 +337,11 @@ def run_named_tests(root: Path, targets: list[str], *, budget: int = 180) -> Pro
         for target in targets
     ):
         raise RunnerError('named tests require nonempty existing tests.test_module[.Class[.test_method]] targets')
-    with tempfile.TemporaryDirectory(prefix='grok-named-smoke-') as directory:
-        return execute([sys.executable, '-m', 'unittest', *targets], root,
-                       _environment(root, Path(directory) / '.coverage'), timeout=budget)
+    retained: dict[str, ProcessResult] = {}
+    with _run_directory('grok-named-smoke-', retained) as directory:
+        retained['tests'] = execute([sys.executable, '-m', 'unittest', *targets], root,
+                                    _environment(root, Path(directory) / '.coverage'), timeout=budget)
+    return retained['tests']
 
 
 def _tool_versions(required: list[str]) -> dict[str, str]:
@@ -617,10 +619,13 @@ def run_trust_tests(root: Path, workers: int) -> ProcessResult:
     command = (_pytest_command(workers, 'loadfile') + ['tests']) if workers else [
         sys.executable, '-m', 'unittest', 'discover', '-s', 'tests',
     ]
-    with tempfile.TemporaryDirectory(prefix='grok-trust-tests-') as directory:
+    retained: dict[str, ProcessResult] = {}
+    # Child temp files now live in this folder, so its cleanup must not replace the verdict (#134).
+    with _run_directory('grok-trust-tests-', retained) as directory:
         environment = _environment(suite, Path(directory) / '.coverage')
         environment['PYTHONPATH'] = os.pathsep.join(str(path) for path in (suite / 'src', suite / 'tests', suite))
-        return execute(command, suite, environment)
+        retained['tests'] = execute(command, suite, environment)
+    return retained['tests']
 
 
 def main() -> int:

scripts/getzilla_protected_write.py

@@ -3,13 +3,15 @@
 
 import argparse
 import json
+import os
 import sys
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / '.getzilla'))
 
 from getzilla.protected_write import ProtectedWriteError, apply_manifest
+from getzilla.state import bind_session
 from getzilla.util import find_root
 
 
@@ -24,6 +26,8 @@ def main() -> int:
     parser.add_argument('--manifest', required=True, type=Path)
     parser.add_argument('--dry-run', action='store_true')
     args = parser.parse_args()
+    # The caller's session, as in getzilla_approve.py, so a session-bound grant follows its later routes (#134).
+    bind_session(os.environ.get('CLAUDE_CODE_SESSION_ID') or os.environ.get('GETZILLA_SESSION_ID'))
     try:
         result = apply_manifest(find_root(ROOT), args.manifest, dry_run=args.dry_run)
     except ProtectedWriteError as exc:

scripts/getzilla_report.py

@@ -38,6 +38,11 @@ def _cell(value: object) -> str:
     return '-' if value is None else str(value)
 
 
+def _printable(value: object) -> str:
+    """Journal text for the terminal: control and format characters become `?` (#134)."""
+    return ''.join(char if char.isprintable() else '?' for char in str(value))
+
+
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(prog='getzilla report', description=__doc__.split('\n')[0])
     parser.add_argument('--since', default='7d', help='period: Nd, Nh or all (default 7d)')
@@ -56,7 +61,7 @@ def main(argv: list[str] | None = None) -> int:
     if summary['profiles']:
         print(f'{"profile":<10} {"routes":>6} {"median min to pass":>19} {"verify runs/route":>18} {"reviews":>8} {"failed outcomes":>20}')
         for name, group in sorted(summary['profiles'].items()):
-            print(f'{name:<10} {group["routes"]:>6} {_cell(group["median_minutes_to_pass"]):>19} '
+            print(f'{_printable(name):<10} {group["routes"]:>6} {_cell(group["median_minutes_to_pass"]):>19} '
                   f'{group["verify_runs_per_route"]:>18} {group["reviews"]:>8} {group["failed_check_or_review_outcomes"]:>20}')
     else:
         print('No routes recorded yet: metrics start with the next route and verification run.')
@@ -68,7 +73,7 @@ def main(argv: list[str] | None = None) -> int:
         print(f'Route observation gaps ({observed["routes_with_gaps"]} of {observed["routes"]} routes; unknown is not zero):')
         gapped = [route for route in summary['routes'] if route.get('gaps')]
         for route in gapped[:MAX_GAP_ROUTES]:
-            print(f'  {route["route_id"]}: ' + '; '.join(route['gaps']))
+            print(f'  {_printable(route["route_id"])}: ' + '; '.join(route['gaps']))
         if len(gapped) > MAX_GAP_ROUTES:
             print(f'  ... {len(gapped) - MAX_GAP_ROUTES} more; use --json for all.')
     hints = summary['counters'].get('path_hints', {})

tests/test_grant_lifecycle.py

@@ -193,6 +193,44 @@ def test_the_cli_records_the_caller_session(self) -> None:
             set_active_route(root, build_route(root, 'Next prompt', 's1').to_dict())
             self.assertTrue(valid(root))
 
+    def test_the_cli_falls_back_to_getzilla_session_id_and_prefers_claude_code(self) -> None:
+        approve = [sys.executable, str(ROOT / 'scripts' / 'getzilla_approve.py'), 'protected-path',
+                   '--reason', 'one file', '--action', 'protected-path-write', '--resource', TARGET]
+        for env_extra, expected in (({'GETZILLA_SESSION_ID': 's1'}, 's1'),
+                                    ({'GETZILLA_SESSION_ID': 's2', 'CLAUDE_CODE_SESSION_ID': 's1'}, 's1')):
+            with self.subTest(env=env_extra), github_project() as root:
+                env = {key: value for key, value in os.environ.items()
+                       if key not in {'CLAUDE_CODE_SESSION_ID', 'GETZILLA_SESSION_ID'}}
+                subprocess.run(approve, cwd=root, env={**env, **env_extra}, check=True, capture_output=True)
+                recorded = json.loads(approvals_path(root).read_text(encoding='utf-8'))[-1]['session_id']
+                self.assertEqual(recorded, expected)
+
+    def test_protected_write_cli_follows_the_caller_session(self) -> None:
+        import base64
+        import shutil
+        from getzilla.util import file_sha256
+        with github_project() as root:
+            (root / 'scripts').mkdir(exist_ok=True)
+            shutil.copy2(ROOT / 'scripts' / 'getzilla_protected_write.py', root / 'scripts')
+            target = root / 'AGENTS.md'
+            add_approval(root, 'protected-path', 'one file', 15, actions=['protected-path-write'],
+                         resources=['AGENTS.md'])
+            set_active_route(root, build_route(root, 'Next prompt', 's1').to_dict())
+            manifest = root.parent / 'protected-write.json'
+            manifest.write_text(json.dumps({'schema_version': 1, 'operations': [{
+                'path': 'AGENTS.md', 'expected_sha256': file_sha256(target),
+                'content_base64': base64.b64encode(target.read_bytes() + b'\n# edit\n').decode('ascii')}]}),
+                encoding='utf-8')
+            cli = [sys.executable, str(root / 'scripts' / 'getzilla_protected_write.py'), '--manifest', str(manifest),
+                   '--dry-run']
+            env = {key: value for key, value in os.environ.items()
+                   if key not in {'CLAUDE_CODE_SESSION_ID', 'GETZILLA_SESSION_ID'}}
+            unbound = subprocess.run(cli, cwd=root, env=env, capture_output=True, text=True)
+            self.assertEqual(unbound.returncode, 2, unbound.stdout)
+            bound = subprocess.run(cli, cwd=root, env={**env, 'CLAUDE_CODE_SESSION_ID': 's1'},
+                                   capture_output=True, text=True)
+            self.assertEqual(bound.returncode, 0, bound.stderr)
+
     def test_without_a_known_session_the_exact_route_is_still_required(self) -> None:
         with github_project() as root:
             approval = add_approval(root, 'protected-path', 'one file', 15,

tests/test_python_test_runner.py

@@ -134,6 +134,27 @@ def test_temp_files_a_killed_child_leaves_are_removed_with_the_run(self):
             self.assertNotEqual(leaked.parent, Path(tempfile.gettempdir()))
             self.assertFalse(leaked.exists())
 
+    def test_a_cleanup_error_keeps_the_named_and_trust_verdicts(self):
+        left: list[tempfile.TemporaryDirectory] = []
+        real_cleanup = tempfile.TemporaryDirectory.cleanup
+
+        def broken_cleanup(self):
+            left.append(self)
+            raise PermissionError('locked file')
+
+        # The real cleanup runs afterwards, so the test leaves no run folders behind.
+        self.addCleanup(lambda: [real_cleanup(directory) for directory in left])
+        with fixture() as root, patch.object(tempfile.TemporaryDirectory, 'cleanup', broken_cleanup):
+            named = python_test_runner.run_named_tests(root, ['tests.test_sample'], budget=30)
+            self.assertEqual(named.returncode, 0, named.stderr)
+            self.assertIn('PermissionError: locked file', named.cleanup_error)
+            (root / 'trust-ci' / 'tests').mkdir(parents=True)
+            (root / 'trust-ci' / 'tests' / 'test_trust.py').write_text(
+                'import unittest\nclass Trust(unittest.TestCase):\n    def test_ok(self):\n        pass\n')
+            trust = python_test_runner.run_trust_tests(root, 0)
+            self.assertEqual(trust.returncode, 0, trust.stderr)
+            self.assertIn('PermissionError: locked file', trust.cleanup_error)
+
     def test_existing_cli_requires_explicit_observation_mode_and_no_record(self):
         script = Path(__file__).resolve().parents[1] / 'scripts/getzilla_verify.py'
         for arguments in (

tests/test_route_metrics.py

@@ -243,6 +243,21 @@ def test_report_cli_bounds_the_route_gap_list(self) -> None:
         self.assertEqual(sum(line.startswith('  r') for line in text.splitlines()), 10)
         self.assertIn('2 more; use --json for all.', text)
 
+    def test_report_cli_never_prints_terminal_control_characters_from_the_journal(self) -> None:
+        stamp = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
+        (self.root / '.getzilla' / 'runtime').mkdir()
+        hostile = 'r1\x1b]0;owned\x07‮'
+        (self.root / METRICS).write_text('\n'.join(json.dumps(item) for item in [
+            {'at': stamp, 'event': 'route', 'route_id': hostile},
+            {'at': stamp, 'event': 'verify', 'route_id': hostile, 'status': 'pass', 'profile': 'std\x1b[2J'},
+        ]) + '\n', encoding='utf-8')
+        text = subprocess.run([sys.executable, str(ROOT / 'scripts' / 'getzilla_report.py'), '--since', '1d'],
+                              cwd=self.root, capture_output=True, text=True, check=True).stdout
+        self.assertNotIn('\x1b', text)
+        self.assertNotIn('\x07', text)
+        self.assertNotIn('‮', text)
+        self.assertIn('r1?]0;owned??:', text)
+
 
 if __name__ == '__main__':
     unittest.main()