feat(harness): allocate canonical issue-scoped broker runtime

2026-10-09 16:11:52 +00:00
parent 3df85090ff
commit ac3cfdaeb9
13 changed files with 713 additions and 69 deletions
@@ -24,17 +24,36 @@ def wait_ready(proc, timeout=3):
    return json.loads(line)


+def ensure_cache(root, owner="alice", repo="ledger"):
+    repository = root / "repositories" / owner / repo
+    repository.mkdir(parents=True, exist_ok=True)
+    (root / "repositories" / owner).chmod(0o700)
+    repository.chmod(0o700)
+    return repository
+
+
+def prepare_fixture_runtime(root, run_id):
+    path = runtime_dir(root, run_id=run_id)
+    path.mkdir(parents=True, exist_ok=True)
+    for directory in (path.parent.parent.parent, path.parent.parent, path.parent, path):
+        directory.chmod(0o700)
+    return path
+
+
+def runtime_dir(root, owner="alice", repo="ledger", issue=7, run_id="run-1"):
+    return root / "repositories" / owner / repo / "worktrees" / f"issue-{issue}" / "runtime" / run_id
+
+
 class ManagedLockBrokerTests(unittest.TestCase):
    def start(self, root, owner="alice", repo="ledger", issue=7, ttl=5, role="gitea-impl-agent", mask=-1):
+        ensure_cache(root, owner, repo)
        locks = root / "repositories" / owner / repo / "locks"
        locks.mkdir(parents=True)
        os.chmod(locks, 0o700)
        (locks / "repository.lock").touch(mode=0o600)
        (locks / f"issue-{issue}.lock").touch(mode=0o600)
-        runtime = root / "runtime"
-        runtime.mkdir()
-        os.chmod(runtime, 0o700)
-        handle = runtime / "handle.json"
+        runtime = runtime_dir(root, owner, repo, issue, "run-1")
+        handle = runtime / "broker.handle"
        socket = runtime / "broker.sock"
        cmd = [sys.executable, str(BROKER), "serve", "--root", str(root), "--owner", owner,
               "--repository", repo, "--issue", str(issue), "--role", role, "--run-id", "run-1",
@@ -60,6 +79,97 @@ class ManagedLockBrokerTests(unittest.TestCase):
            with s.makefile("rb") as response_file:
                return json.loads(response_file.readline())

+    def test_constructor_rejects_lexical_root_and_handle_aliases(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d); ensure_cache(root)
+            locks = root / 'repositories/alice/ledger/locks'; locks.mkdir(mode=0o700)
+            values = dict(root=str(root), owner='alice', repository='ledger', issue=7,
+                          role='gitea-impl-agent', run_id='aliases', ttl_seconds=3,
+                          acquire_timeout=0, handle=None, socket=None)
+            path = runtime_dir(root, run_id='aliases')
+            for override in [dict(root=f'{root}/./'), dict(handle=f'{path}/./broker.handle')]:
+                with self.subTest(override=override), self.assertRaises(ValueError):
+                    broker.Broker(Namespace(**(values | override)))
+            self.assertFalse(path.exists())
+
+    def test_ready_retains_only_one_pinned_runtime_directory_descriptor(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d)
+            proc, handle, socket = self.start(root)
+            try:
+                targets = []
+                for fd in Path(f'/proc/{proc.pid}/fd').iterdir():
+                    try:
+                        if os.readlink(fd) == str(handle.parent):
+                            targets.append(fd.name)
+                    except FileNotFoundError:
+                        pass
+                self.assertEqual(len(targets), 1, f'leaked runtime descriptors: {targets}')
+            finally:
+                broker.request_handle(handle, 'release')
+                proc.wait(timeout=3)
+                proc.stdout.close(); proc.stderr.close()
+
+    def test_serve_defaults_to_canonical_names_and_rejects_path_override(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d)
+            ensure_cache(root)
+            locks = root / "repositories/alice/ledger/locks"
+            locks.mkdir(); locks.chmod(0o700)
+            (locks / "repository.lock").touch(mode=0o600)
+            (locks / "issue-7.lock").touch(mode=0o600)
+            base = [sys.executable, str(BROKER), "serve", "--root", str(root), "--owner", "alice",
+                    "--repository", "ledger", "--issue", "7", "--role", "gitea-impl-agent",
+                    "--run-id", "defaults", "--ttl-seconds", "3"]
+            proc = subprocess.Popen(base, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            try:
+                ready = wait_ready(proc)
+                expected = runtime_dir(root, run_id="defaults")
+                self.assertEqual(Path(ready["handle"]), expected / "broker.handle")
+                self.assertEqual(Path(ready["socket"]), expected / "broker.sock")
+                broker.request_handle(expected / "broker.handle", "release")
+                self.assertEqual(proc.wait(timeout=2), 0)
+            finally:
+                if proc.poll() is None: proc.kill(); proc.wait()
+                proc.stdout.close(); proc.stderr.close()
+            runtime = runtime_dir(root, run_id="override")
+            result = subprocess.run([*base[:], "--handle", str(root / "elsewhere"), "--run-id", "override"],
+                                    capture_output=True, text=True)
+            self.assertNotEqual(result.returncode, 0)
+            self.assertFalse(runtime.exists())
+
+    def test_issue_contention_does_not_create_runtime(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d)
+            ensure_cache(root)
+            locks = root / "repositories/alice/ledger/locks"
+            locks.mkdir(); locks.chmod(0o700)
+            (locks / "repository.lock").touch(mode=0o600)
+            issue_lock = locks / "issue-7.lock"
+            issue_lock.touch(mode=0o600)
+            with ManagedLock(issue_lock):
+                proc = subprocess.run([sys.executable, str(BROKER), "serve", "--root", str(root),
+                    "--owner", "alice", "--repository", "ledger", "--issue", "7",
+                    "--role", "gitea-impl-agent", "--run-id", "blocked", "--ttl-seconds", "2"],
+                    capture_output=True, text=True)
+            self.assertNotEqual(proc.returncode, 0)
+            self.assertFalse(runtime_dir(root, run_id="blocked").exists())
+
+    def test_long_socket_path_connects_via_held_directory_fd(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d) / ("r" * 48)
+            root.mkdir()
+            ensure_cache(root)
+            proc, handle, sock = self.start(root)
+            try:
+                self.assertGreater(len(str(sock).encode()), 107)
+                self.assertEqual(broker.request_handle(handle, "status")["status"], "held")
+                self.assertEqual(broker.request_handle(handle, "release")["status"], "released")
+                self.assertEqual(proc.wait(timeout=3), 0)
+            finally:
+                if proc.poll() is None: proc.kill(); proc.wait()
+                proc.stdout.close(); proc.stderr.close()
+
    def test_cross_process_contention_and_authenticated_release(self):
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
@@ -133,6 +243,7 @@ class ManagedLockBrokerTests(unittest.TestCase):
                proc, handle, sock = self.start(Path(d), mask=mask)
                try:
                    self.assertEqual(stat.S_IMODE(sock.lstat().st_mode), 0o600)
+                    self.assertEqual(stat.S_IMODE(handle.parent.stat().st_mode), 0o700)
                    self.assertTrue(stat.S_ISSOCK(sock.lstat().st_mode))
                    self.assertEqual(broker.request_handle(handle, "status")["status"], "held")
                    self.assertEqual(broker.request_handle(handle, "release")["status"], "released")
@@ -146,12 +257,13 @@ class ManagedLockBrokerTests(unittest.TestCase):

    def make_bound_instance(self, root):
        import socket
+        ensure_cache(root)
        locks = root / "repositories/alice/ledger/locks"
        locks.mkdir(parents=True); locks.chmod(0o700)
        (locks / "repository.lock").touch(mode=0o600)
-        runtime = root / "runtime"; runtime.mkdir(mode=0o700)
+        runtime = prepare_fixture_runtime(root, "cleanup-test")
        instance = broker.Broker(Namespace(root=str(root), owner="alice", repository="ledger", issue=7,
-            role="gitea-impl-agent", run_id="cleanup-test", handle=str(runtime / "handle.json"),
+            role="gitea-impl-agent", run_id="cleanup-test", handle=str(runtime / "broker.handle"),
            socket=str(runtime / "broker.sock"), ttl_seconds=5.0, acquire_timeout=0.0))
        instance.set.issue.acquire()
        broker._write_handle(instance.handle, instance.state, exclusive=True,
@@ -240,12 +352,16 @@ class ManagedLockBrokerTests(unittest.TestCase):
        with tempfile.TemporaryDirectory() as d:
            instance = self.make_bound_instance(Path(d))
            instance.cleanup("released")
-            args = Namespace(root=str(instance.root), owner="alice", repository="ledger", issue=7,
-                role="gitea-impl-agent", run_id="startup-permission", handle=str(instance.handle.parent / "new.handle"),
-                socket=str(instance.sock_path.parent / "new.sock"), ttl_seconds=5.0, acquire_timeout=0.0)
-            fresh = broker.Broker(args)
+            fresh = broker.Broker(Namespace(root=str(instance.root), owner="alice", repository="ledger", issue=7,
+                role="gitea-impl-agent", run_id="startup-permission", handle=str(runtime_dir(instance.root, run_id="startup-permission") / "broker.handle"),
+                socket=str(runtime_dir(instance.root, run_id="startup-permission") / "broker.sock"), ttl_seconds=5.0, acquire_timeout=0.0))
            out = io.StringIO()
-            with patch.object(broker.os, "chmod", side_effect=PermissionError("injected")), contextlib.redirect_stdout(out):
+            real_chmod = os.chmod
+            def fail_socket_chmod(path, mode, **kwargs):
+                if path == fresh.sock_path.name:
+                    raise PermissionError("injected")
+                return real_chmod(path, mode, **kwargs)
+            with patch.object(broker.os, "chmod", side_effect=fail_socket_chmod), contextlib.redirect_stdout(out):
                with self.assertRaises(PermissionError): fresh.run()
            self.assertEqual(out.getvalue(), "")
            self.assertFalse(fresh.sock_path.exists())
@@ -272,8 +388,8 @@ class ManagedLockBrokerTests(unittest.TestCase):
        with tempfile.TemporaryDirectory() as d:
            instance = self.make_bound_instance(Path(d)); instance.cleanup("released")
            fresh = broker.Broker(Namespace(root=str(instance.root), owner="alice", repository="ledger", issue=7,
-                role="gitea-impl-agent", run_id="handle-swap", handle=str(instance.handle.parent / "swap.handle"),
-                socket=str(instance.sock_path.parent / "swap.sock"), ttl_seconds=5.0, acquire_timeout=0.0))
+                role="gitea-impl-agent", run_id="handle-swap", handle=str(runtime_dir(instance.root, run_id="handle-swap") / "broker.handle"),
+                socket=str(runtime_dir(instance.root, run_id="handle-swap") / "broker.sock"), ttl_seconds=5.0, acquire_timeout=0.0))
            original_write = broker._write_handle
            foreign = b'{"foreign":"preserve"}\n'
            def replace_after_create(path, state, **kwargs):
@@ -303,13 +419,14 @@ class ManagedLockBrokerTests(unittest.TestCase):
        with tempfile.TemporaryDirectory() as d:
            instance = self.make_bound_instance(Path(d)); instance.cleanup("released")
            fresh = broker.Broker(Namespace(root=str(instance.root), owner="alice", repository="ledger", issue=7,
-                role="gitea-impl-agent", run_id="socket-swap", handle=str(instance.handle.parent / "swap.handle"),
-                socket=str(instance.sock_path.parent / "swap.sock"), ttl_seconds=5.0, acquire_timeout=0.0))
+                role="gitea-impl-agent", run_id="socket-swap", handle=str(runtime_dir(instance.root, run_id="socket-swap") / "broker.handle"),
+                socket=str(runtime_dir(instance.root, run_id="socket-swap") / "broker.sock"), ttl_seconds=5.0, acquire_timeout=0.0))
            real_chmod = os.chmod
            def replace_after_chmod(path, mode, **kwargs):
                real_chmod(path, mode, **kwargs)
-                fresh.sock_path.unlink()
-                fresh.sock_path.write_text("foreign socket replacement")
+                if path == fresh.sock_path.name:
+                    fresh.sock_path.unlink()
+                    fresh.sock_path.write_text("foreign socket replacement")
            out = io.StringIO()
            with patch.object(broker.os, "chmod", side_effect=replace_after_chmod), contextlib.redirect_stdout(out):
                with self.assertRaises(ValueError): fresh.run()
@@ -354,13 +471,14 @@ class ManagedLockBrokerTests(unittest.TestCase):
        for action in ("renew", "release"):
            with self.subTest(action=action), tempfile.TemporaryDirectory() as d:
                root = Path(d)
+                ensure_cache(root)
                locks = root / "repositories/alice/ledger/locks"
                locks.mkdir(parents=True); os.chmod(locks, 0o700)
                (locks / "repository.lock").touch(mode=0o600)
-                runtime = root / "runtime"; runtime.mkdir(mode=0o700)
+                runtime = prepare_fixture_runtime(root, "expiry-test")
                args = Namespace(root=str(root), owner="alice", repository="ledger", issue=7,
                                 role="gitea-impl-agent", run_id="expiry-test",
-                                 handle=str(runtime / "handle.json"), socket=str(runtime / "broker.sock"),
+                                 handle=str(runtime / "broker.handle"), socket=str(runtime / "broker.sock"),
                                 ttl_seconds=5.0, acquire_timeout=0.0)
                instance = broker.Broker(args)
                instance.set.issue.acquire()
@@ -378,11 +496,12 @@ class ManagedLockBrokerTests(unittest.TestCase):
    def test_startup_failure_releases_issue_and_terminalizes_partial_handle(self):
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
+            ensure_cache(root)
            locks = root / "repositories/alice/ledger/locks"
            locks.mkdir(parents=True); os.chmod(locks, 0o700)
            (locks / "repository.lock").touch(mode=0o600)
-            runtime = root / "runtime"; runtime.mkdir(mode=0o700)
-            handle = runtime / "handle.json"
+            runtime = runtime_dir(root, run_id="startup-failure")
+            handle = runtime / "broker.handle"
            args = Namespace(root=str(root), owner="alice", repository="ledger", issue=7,
                             role="gitea-impl-agent", run_id="startup-failure",
                             handle=str(handle), socket=str(runtime / "broker.sock"),
@@ -405,11 +524,12 @@ class ManagedLockBrokerTests(unittest.TestCase):
    def test_startup_existing_handle_is_never_claimed_or_modified(self):
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
+            ensure_cache(root)
            locks = root / "repositories/alice/ledger/locks"
            locks.mkdir(parents=True); os.chmod(locks, 0o700)
            (locks / "repository.lock").touch(mode=0o600)
-            runtime = root / "runtime"; runtime.mkdir(mode=0o700)
-            handle = runtime / "handle.json"
+            runtime = prepare_fixture_runtime(root, "existing-handle")
+            handle = runtime / "broker.handle"
            original = b'{"foreign":"do-not-touch"}\n'
            handle.write_bytes(original); handle.chmod(0o600)
            args = Namespace(root=str(root), owner="alice", repository="ledger", issue=7,
@@ -417,7 +537,7 @@ class ManagedLockBrokerTests(unittest.TestCase):
                             handle=str(handle), socket=str(runtime / "broker.sock"),
                             ttl_seconds=5.0, acquire_timeout=0.0)
            instance = broker.Broker(args)
-            with self.assertRaises(FileExistsError):
+            with self.assertRaises(ValueError):
                instance.run()
            self.assertEqual(handle.read_bytes(), original)
            self.assertTrue(probe_free(locks / "issue-7.lock"))
@@ -462,4 +582,4 @@ class ManagedLockBrokerTests(unittest.TestCase):


 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()
@@ -0,0 +1,189 @@
+import json
+import os
+import stat
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from harness.lib.managed_lock import ManagedLockSet
+from harness.lib import runtime_root as runtime
+
+
+class RuntimeRootTests(unittest.TestCase):
+    def cache(self, root):
+        repo = root / 'repositories' / 'owner' / 'repo'
+        for p in [root / 'repositories', root / 'repositories' / 'owner', repo, repo / 'locks']:
+            p.mkdir(mode=0o700)
+        return repo
+
+    def prepare(self, root, run='workflow-1'):
+        with ManagedLockSet.for_issue(root, 'owner', 'repo', 17) as locks:
+            with locks.repository_scope():
+                return runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, run, lock_set=locks)
+
+    def test_new_runtime_is_private_and_preserves_existing_cache(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); repo = self.cache(root)
+            os.chmod(root / 'repositories', 0o755)
+            before = repo.stat()
+            result = self.prepare(root)
+            try:
+                self.assertEqual(result.path, runtime.issue_runtime_path(root, 'owner', 'repo', 17, 'workflow-1'))
+                self.assertEqual(stat.S_IMODE(os.fstat(result.fd).st_mode), 0o700)
+                self.assertEqual(os.fstat(result.fd).st_ino, result.path.stat().st_ino)
+            finally:
+                os.close(result.fd)
+            self.assertEqual(repo.stat().st_ino, before.st_ino)
+            self.assertEqual(stat.S_IMODE((root / 'repositories').stat().st_mode), 0o755)
+            self.assertFalse((result.path.parents[1] / 'state.json').exists())
+
+    def test_missing_or_wrong_lock_set_cannot_create_directories(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); repo = self.cache(root)
+            locks = ManagedLockSet.for_issue(root, 'owner', 'repo', 17)
+            with self.assertRaises(ValueError):
+                runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, 'run', lock_set=locks)
+            self.assertFalse((repo / 'worktrees').exists())
+            with locks:
+                with self.assertRaises(ValueError):
+                    runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, 'run', lock_set=locks)
+                with locks.repository_scope():
+                    with self.assertRaises(ValueError):
+                        runtime.prepare_issue_runtime(root, 'owner', 'repo', 18, 'run', lock_set=locks)
+            self.assertFalse((repo / 'worktrees').exists())
+
+    def test_kernel_unlocked_wrappers_cannot_authorize_runtime_creation(self):
+        import fcntl
+        for resource in ('issue', 'repository'):
+            with self.subTest(resource=resource), tempfile.TemporaryDirectory() as tmp:
+                root = Path(tmp); repo = self.cache(root)
+                with ManagedLockSet.for_issue(root, 'owner', 'repo', 17) as locks:
+                    with locks.repository_scope():
+                        fcntl.flock(getattr(locks, resource).fd, fcntl.LOCK_UN)
+                        self.assertTrue(getattr(locks, resource).held)
+                        with self.assertRaises(ValueError):
+                            runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, 'run', lock_set=locks)
+                self.assertFalse((repo / 'worktrees').exists())
+
+    def test_run_collision_preserves_competing_run_and_parent_chain(self):
+        from unittest.mock import patch
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); self.cache(root)
+            path = runtime.issue_runtime_path(root, 'owner', 'repo', 17, 'collision')
+            mkdir = os.mkdir
+            def competing_run(name, mode=0o777, *, dir_fd=None):
+                mkdir(name, mode, dir_fd=dir_fd)
+                if name == 'collision':
+                    raise FileExistsError('competing allocator created run')
+            with patch.object(runtime.os, 'mkdir', side_effect=competing_run):
+                with self.assertRaises(ValueError): self.prepare(root, 'collision')
+            self.assertTrue(path.is_dir())
+            self.assertTrue(path.parent.is_dir())
+            self.assertTrue(path.parent.parent.is_dir())
+            self.assertEqual(list(path.iterdir()), [])
+
+    def test_stale_private_run_is_rejected_without_reuse(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); self.cache(root)
+            result = self.prepare(root); os.close(result.fd)
+            marker = result.path / 'marker'; marker.write_text('preserve')
+            before = result.path.stat()
+            with self.assertRaises(ValueError): self.prepare(root)
+            self.assertEqual(result.path.stat().st_ino, before.st_ino)
+            self.assertEqual(marker.read_text(), 'preserve')
+
+    def test_unsafe_parent_is_not_repaired(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); repo = self.cache(root)
+            p = repo / 'worktrees'; p.mkdir(mode=0o755)
+            with self.assertRaises(ValueError): self.prepare(root)
+            self.assertEqual(stat.S_IMODE(p.stat().st_mode), 0o755)
+            self.assertEqual(list(p.iterdir()), [])
+
+    def test_symlink_parent_cannot_redirect_creation(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); repo = self.cache(root)
+            other = root / 'elsewhere'; other.mkdir(mode=0o700)
+            (repo / 'worktrees').symlink_to(other)
+            with self.assertRaises((ValueError, OSError)): self.prepare(root)
+            self.assertEqual(list(other.iterdir()), [])
+
+    def test_invalid_identifiers_and_literal_traversal_are_rejected(self):
+        for root, owner, repository, issue, run in [
+            ('/tmp/../root', 'owner', 'repo', 17, 'run'),
+            ('/tmp/root', '../owner', 'repo', 17, 'run'),
+            ('/tmp/root', 'owner', 'repo', True, 'run'),
+            ('/tmp/root', 'owner', 'repo', 17, '..'),
+            ('/tmp/root', 'owner', 'repo', 17, 'x' * 129),
+        ]:
+            with self.subTest(root=root, owner=owner, issue=issue, run=run), self.assertRaises(ValueError):
+                runtime.issue_runtime_path(root, owner, repository, issue, run)
+        self.assertIn('o' * 255, str(runtime.issue_runtime_path('/tmp/root', 'o' * 255, 'repo', 17, 'run')))
+
+    def git(self, *args):
+        result = subprocess.run(['git', *map(str, args)], capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        return result.stdout
+
+    def test_runtime_only_container_supports_real_new_git_checkout_and_removal(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); repo = self.cache(root)
+            seed = root / 'seed'
+            self.git('init', seed)
+            (seed / 'file').write_text('fixture')
+            self.git('-C', seed, 'add', 'file')
+            self.git('-C', seed, '-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid', 'commit', '-m', 'fixture')
+            self.git('clone', '--bare', seed, repo / 'repository.git')
+            source = self.git('-C', seed, 'rev-parse', 'HEAD').strip()
+            run = self.prepare(root); os.close(run.fd)
+            issue = repo / 'worktrees' / 'issue-17'
+            self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'runtime-only')
+            self.git('--git-dir', repo / 'repository.git', 'worktree', 'add', '--detach', issue / 'checkout', source)
+            self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'invalid')
+            state = issue / 'state.json'; state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1'})); state.chmod(0o600)
+            self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'checkout-candidate')
+            self.git('--git-dir', repo / 'repository.git', 'worktree', 'remove', issue / 'checkout')
+            state.unlink()
+            self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'runtime-only')
+            self.assertTrue(run.path.exists())
+            (issue / 'unknown').write_text('preserve')
+            self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'invalid')
+
+    def test_inspection_cli_requires_live_repository_lease_and_uses_real_git(self):
+        import sys
+        from harness.lib import managed_lock_broker as broker
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); repo = self.cache(root)
+            self.git('init', '--bare', repo / 'repository.git')
+            base = Path(__file__).resolve().parents[2]
+            cmd = [sys.executable, '-B', str(base / 'lib' / 'managed_lock_broker.py'), 'serve',
+                   '--root', str(root), '--owner', 'owner', '--repository', 'repo', '--issue', '17',
+                   '--role', 'gitea-impl-agent', '--run-id', 'cli-workflow', '--ttl-seconds', '30']
+            process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+            try:
+                line = process.stdout.readline()
+                self.assertTrue(line, 'broker failed before ready')
+                ready = json.loads(line)
+                inspect = [sys.executable, '-B', str(base / 'lib' / 'runtime_root.py'), 'inspect-worktree',
+                           '--root', str(root), '--owner', 'owner', '--repository', 'repo', '--issue', '17',
+                           '--handle', ready['handle']]
+                result = subprocess.run(inspect, capture_output=True, text=True, timeout=10)
+                self.assertNotEqual(result.returncode, 0)
+                broker.request_handle(ready['handle'], 'acquire_repository')
+                result = subprocess.run(inspect, capture_output=True, text=True, timeout=10)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                self.assertEqual(json.loads(result.stdout)['container'], 'runtime-only')
+                broker.request_handle(ready['handle'], 'release_repository')
+                broker.request_handle(ready['handle'], 'release')
+                self.assertEqual(process.wait(timeout=10), 0)
+            finally:
+                if process.poll() is None:
+                    process.terminate(); process.wait(timeout=10)
+                process.stdout.close(); process.stderr.close()
+
+    def test_inspection_rejects_missing_cache_instead_of_creating_it(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            with self.assertRaises((ValueError, OSError)):
+                runtime.inspect_issue_worktree_admission(Path(tmp), 'owner', 'repo', 17)
+            self.assertEqual(list(Path(tmp).iterdir()), [])
@@ -17,6 +17,36 @@ CANONICAL = "<harness-root>/repositories/<owner>/<repo>/worktrees/issue-<number>


 class VerificationContractDocumentationTests(unittest.TestCase):
+    def test_runtime_policy_is_broker_owned_and_separate_from_verification(self):
+        text = CACHE.read_text(encoding="utf-8")
+        self.assertIn("runtime/<workflow-run-id>", text)
+        self.assertIn("broker.handle", text)
+        self.assertIn("broker.sock", text)
+        self.assertIn("ready JSON", text)
+        self.assertIn("runtime 생성은 승인이나 admission을 의미하지 않는다", text)
+        self.assertIn("active resource", text)
+        self.assertIn("unknown", text)
+        self.assertIn("성공 기록을 실패로 덮어쓰지 않는다", text)
+        for path in SKILLS.values():
+            with self.subTest(path=path.name):
+                role = path.read_text(encoding="utf-8")
+                gate = role.split("## Managed-lock broker gate", 1)[1].split("\n## ", 1)[0]
+                self.assertIn("broker-generated", gate)
+                self.assertIn("ready JSON", gate)
+                self.assertNotIn("pre-created", gate)
+                self.assertNotIn("pre-create", gate)
+
+    def test_role_runtime_startup_precedes_and_does_not_replace_admission(self):
+        roles = [path for path in (ROOT / 'harness/agents').rglob('SKILL.md')
+                 if 'broker-generated runtime' in path.read_text()]
+        self.assertEqual(len(roles), 6)
+        for path in roles:
+            with self.subTest(role=path.parent.name):
+                gate = path.read_text().split('## Managed-lock broker gate', 1)[1].split('\n## ', 1)[0]
+                self.assertIn('ready 이후 각 workflow의 진입 검증·승인 조건을 확인한다', gate)
+                self.assertIn('Runtime 생성은 approval이나 worktree admission을 대신하지 않는다', gate)
+                self.assertIn('ready JSON의 `handle`/`socket`', gate)
+
    def test_evidence_plan_disables_python_bytecode_in_clean_checkout(self):
        import json
        template = ROOT / "harness/agents/impl/skills/gitea-issue-rework/templates/evidence-plan.json"
@@ -184,6 +184,7 @@ class VerificationRootTests(unittest.TestCase):
        locks.mkdir(mode=0o700)
        (locks / "repository.lock").touch(mode=0o600)
        broker_script = Path(verification_root.broker.__file__)
+        self.handle = self.issue / "runtime/broker-run/broker.handle"
        socket_path = self.handle.parent / "broker.sock"
        command = [
            sys.executable,