[verified] Align broker fixtures with descriptor-relative socket paths

harness/tests/README.md

@@ -21,6 +21,14 @@ trees under `/opt/data/gitea-agent/harness/tests/{common,spec,impl,review}`.
 This document is owned by the canonical harness tests and must be updated in
 place when their supported discovery contract or command paths change.
 
+Broker fixture contract: in-process broker fixtures bind and connect through
+`/proc/self/fd/{parent_fd}/{socket_name}`, matching the production broker. Long
+UTF-8 roots and socket paths are supported; the fixture must not shorten them or
+reject them merely because the conventional Unix-domain path encoding exceeds
+107 bytes. Fixture setup failures clean up only resources they own, preserve
+foreign/replaced paths, close every client and directory descriptor, and retain
+unauthenticated-token rejection behavior.
+
 It validates all `test*.py` files recursively under exactly these required
 roles: `common`, `spec`, `impl`, and `review`. `harness/tests`, every role, and
 every nested test package must have an `__init__.py`; every role and every

harness/tests/common/test_managed_lock_broker.py

@@ -70,14 +70,19 @@ class ManagedLockBrokerTests(unittest.TestCase):
         if token is not None:
             data["token"] = token
         import socket
-        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
+        parent_fd, name = broker._open_parent(Path(data["socket"]))
-            s.connect(data["socket"])
+        try:
-            payload = {"action": action, "token": data["token"] if token is None else token}
+            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
-            if ttl is not None:
+                s.settimeout(3)
-                payload["ttl_seconds"] = ttl
+                s.connect(f"/proc/self/fd/{parent_fd}/{name}")
-            s.sendall((json.dumps(payload) + "\n").encode())
+                payload = {"action": action, "token": data["token"] if token is None else token}
-            with s.makefile("rb") as response_file:
+                if ttl is not None:
-                return json.loads(response_file.readline())
+                    payload["ttl_seconds"] = ttl
+                s.sendall((json.dumps(payload) + "\n").encode())
+                with s.makefile("rb") as response_file:
+                    return json.loads(response_file.readline())
+        finally:
+            os.close(parent_fd)
 
     def test_constructor_rejects_lexical_root_and_handle_aliases(self):
         with tempfile.TemporaryDirectory() as d:
@@ -255,7 +260,8 @@ class ManagedLockBrokerTests(unittest.TestCase):
                         proc.terminate(); proc.wait(timeout=3)
                     proc.stdout.close(); proc.stderr.close()
 
-    def make_bound_instance(self, root):
+    def make_bound_instance(self, root, *, fail_stage=None, captured=None, stages=None, failure=None):
+        """Build a real fixture, cleaning up and re-raising setup failures."""
         import socket
         ensure_cache(root)
         locks = root / "repositories/alice/ledger/locks"
@@ -265,16 +271,158 @@ class ManagedLockBrokerTests(unittest.TestCase):
         instance = broker.Broker(Namespace(root=str(root), owner="alice", repository="ledger", issue=7,
             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()
+        if captured is not None:
-        broker._write_handle(instance.handle, instance.state, exclusive=True,
+            captured.append(instance)
-                             on_created=instance._record_handle_identity)
+        observed = stages if stages is not None else {}
-        instance.handle_created = True; instance.started = True
+        try:
-        instance.socket_parent_fd = broker._open_directory(runtime)
+            instance.set.issue.acquire()
-        instance.server = socket.socket(socket.AF_UNIX)
+            observed["issue_acquired"] = instance.set.issue.held
-        instance.server.bind(str(instance.sock_path))
+            if fail_stage == "after-issue":
-        st = instance.sock_path.lstat()
+                raise RuntimeError("sentinel after issue acquire")
-        instance.socket_identity = (st.st_dev, st.st_ino)
+            broker._write_handle(instance.handle, instance.state, exclusive=True,
-        return instance
+                                 on_created=instance._record_handle_identity)
+            instance.handle_created = True; instance.started = True
+            observed["handle_created"] = instance.handle_created and instance.handle_identity is not None
+            if fail_stage == "after-handle":
+                raise RuntimeError("sentinel after handle creation")
+            instance.socket_parent_fd = broker._open_directory(runtime)
+            observed["parent_fd_assigned"] = instance.socket_parent_fd is not None
+            if fail_stage == "after-parent-fd":
+                raise RuntimeError("sentinel after parent fd ownership")
+            instance.server = socket.socket(socket.AF_UNIX)
+            if fail_stage == "socket-constructor":
+                raise RuntimeError("sentinel socket constructor failure")
+            instance.server.bind(f"/proc/self/fd/{instance.socket_parent_fd}/{instance.sock_path.name}")
+            st = os.stat(instance.sock_path.name, dir_fd=instance.socket_parent_fd, follow_symlinks=False)
+            instance.socket_identity = (st.st_dev, st.st_ino)
+            observed["socket_bound"] = instance.socket_identity is not None
+            if fail_stage == "post-bind":
+                raise RuntimeError("sentinel after socket identity")
+            return instance
+        except BaseException as original:
+            if failure is not None:
+                failure.append(original)
+            try:
+                instance.cleanup("released")
+            except BaseException as cleanup_error:
+                original.add_note(f"fixture cleanup failed: {type(cleanup_error).__name__}")
+            raise
+
+    def assert_failed_fixture_cleanup(self, instance, stages, failure, original):
+        self.assertIsNotNone(instance)
+        self.assertIs(failure[0], original)
+        self.assertTrue(stages["issue_acquired"])
+        self.assertTrue(probe_free(instance.set.issue.path))
+        self.assertTrue(probe_free(instance.set.repository.path))
+        self.assertIsNone(instance.socket_parent_fd)
+        self.assertIsNone(instance.server)
+        self.assertTrue(instance.handle_created)
+        terminal = json.loads(instance.handle.read_text())
+        self.assertNotIn("token", terminal)
+        self.assertIn(terminal["status"], ("released", "cleanup-incomplete"))
+
+    def test_fixture_partial_handle_write_failure_releases_owned_lease(self):
+        captured = []
+        writes = []
+        original_write = broker._write_all
+        sentinel = OSError("sentinel partial fixture handle write")
+
+        def partial_write(fd, data):
+            if not writes:
+                writes.append(os.write(fd, data[:1]))
+                raise sentinel
+            return original_write(fd, data)
+
+        with tempfile.TemporaryDirectory() as d:
+            try:
+                with patch.object(broker, "_write_all", side_effect=partial_write):
+                    with self.assertRaises(OSError) as raised:
+                        self.make_bound_instance(Path(d), captured=captured)
+                self.assertIs(raised.exception, sentinel)
+                self.assertEqual(writes, [1], "partial write was not exercised")
+                instance = captured[0]
+                self.assertTrue(instance.handle_created)
+                self.assertTrue(probe_free(instance.set.issue.path), "fixture leaked its acquired lease")
+                self.assertIsNone(instance.socket_parent_fd)
+                self.assertIsNone(instance.server)
+                terminal = json.loads(instance.handle.read_text())
+                self.assertNotIn("token", terminal)
+                self.assertEqual(terminal["status"], "released")
+            finally:
+                if captured:
+                    captured[0].cleanup("released")
+
+    def test_fixture_failure_after_handle_creation_preserves_original_and_cleans(self):
+        captured, stages, failure = [], {}, []
+        instance = None
+        with tempfile.TemporaryDirectory() as d:
+            try:
+                with self.assertRaisesRegex(RuntimeError, "sentinel after handle creation") as raised:
+                    self.make_bound_instance(Path(d), fail_stage="after-handle",
+                                             captured=captured, stages=stages, failure=failure)
+                instance = captured[0]
+                self.assert_failed_fixture_cleanup(instance, stages, failure, raised.exception)
+                self.assertTrue(stages["handle_created"])
+                self.assertFalse(stages.get("parent_fd_assigned", False))
+            finally:
+                if instance is not None:
+                    instance.cleanup("released")
+
+    def test_fixture_socket_constructor_failure_closes_owned_parent_fd(self):
+        captured, stages, failure = [], {}, []
+        instance = None
+        with tempfile.TemporaryDirectory() as d:
+            try:
+                with patch("socket.socket", side_effect=RuntimeError("sentinel socket constructor failure")):
+                    with self.assertRaisesRegex(RuntimeError, "sentinel socket constructor failure") as raised:
+                        self.make_bound_instance(Path(d), captured=captured, stages=stages, failure=failure)
+                instance = captured[0]
+                self.assert_failed_fixture_cleanup(instance, stages, failure, raised.exception)
+                self.assertTrue(stages["parent_fd_assigned"])
+                self.assertFalse(stages.get("socket_bound", False))
+            finally:
+                if instance is not None:
+                    instance.cleanup("released")
+
+    def test_fixture_post_bind_failure_removes_only_captured_socket(self):
+        captured, stages, failure = [], {}, []
+        instance = None
+        with tempfile.TemporaryDirectory() as d:
+            try:
+                with self.assertRaisesRegex(RuntimeError, "sentinel after socket identity") as raised:
+                    self.make_bound_instance(Path(d), fail_stage="post-bind",
+                                             captured=captured, stages=stages, failure=failure)
+                instance = captured[0]
+                self.assert_failed_fixture_cleanup(instance, stages, failure, raised.exception)
+                self.assertTrue(stages["parent_fd_assigned"])
+                self.assertTrue(stages["socket_bound"])
+                self.assertFalse(instance.sock_path.exists())
+            finally:
+                if instance is not None:
+                    instance.cleanup("released")
+
+    def test_fixture_cleanup_diagnostic_does_not_mask_setup_failure(self):
+        captured, stages, failure = [], {}, []
+        instance = None
+        real_cleanup = broker.Broker.cleanup
+
+        def cleanup_then_raise(status):
+            real_cleanup(captured[0], status)
+            raise OSError("sentinel cleanup diagnostic")
+
+        with tempfile.TemporaryDirectory() as d:
+            try:
+                with patch.object(broker.Broker, "cleanup", side_effect=cleanup_then_raise):
+                    with self.assertRaisesRegex(RuntimeError, "sentinel after handle creation") as raised:
+                        self.make_bound_instance(Path(d), fail_stage="after-handle",
+                                                 captured=captured, stages=stages, failure=failure)
+                instance = captured[0]
+                self.assertIs(failure[0], raised.exception)
+                self.assertIn("fixture cleanup failed: OSError", getattr(raised.exception, "__notes__", []))
+                self.assertTrue(probe_free(instance.set.issue.path))
+            finally:
+                if instance is not None:
+                    real_cleanup(instance, "released")
 
     def test_release_reports_socket_failure_separately_and_stops(self):
         with tempfile.TemporaryDirectory() as d:
@@ -326,12 +474,17 @@ class ManagedLockBrokerTests(unittest.TestCase):
             target = root / "replacement-target"; target.write_text("untouched")
             try:
                 data = json.loads(handle.read_text())
-                with socket.socket(socket.AF_UNIX) as client:
+                parent_fd, socket_name = broker._open_parent(sock_path)
-                    client.settimeout(3); client.connect(str(sock_path))
+                try:
-                    sock_path.unlink(); sock_path.symlink_to(target)
+                    with socket.socket(socket.AF_UNIX) as client:
-                    client.sendall((json.dumps({"action": "release", "token": data["token"]}) + "\n").encode())
+                        client.settimeout(3)
-                    with client.makefile("rb") as stream:
+                        client.connect(f"/proc/self/fd/{parent_fd}/{socket_name}")
-                        result = json.loads(stream.readline())
+                        sock_path.unlink(); sock_path.symlink_to(target)
+                        client.sendall((json.dumps({"action": "release", "token": data["token"]}) + "\n").encode())
+                        with client.makefile("rb") as stream:
+                            result = json.loads(stream.readline())
+                finally:
+                    os.close(parent_fd)
                 self.assertEqual(result["error"], "cleanup-incomplete")
                 self.assertTrue(result["cleanup"]["lease_released"])
                 self.assertEqual(proc.wait(timeout=3), 2)
@@ -580,6 +733,105 @@ class ManagedLockBrokerTests(unittest.TestCase):
                 self.assertNotEqual(p.returncode, 0)
                 self.assertNotIn("token", p.stderr.lower())
 
+    def test_fixture_binds_and_cleans_long_utf8_socket_path(self):
+        """The in-process fixture must exercise the broker's FD-relative path."""
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d) / ("é" * 28)
+            root.mkdir()
+            instance = None
+            try:
+                instance = self.make_bound_instance(root)
+                self.assertGreater(len(os.fsencode(instance.sock_path)), 107)
+                self.assertTrue(instance.server)
+                self.assertIsNotNone(instance.socket_parent_fd)
+            finally:
+                if instance is not None:
+                    instance.cleanup("released")
+                    self.assertIsNone(instance.socket_parent_fd)
+                    self.assertIsNone(instance.server)
+                self.assertTrue(probe_free(root / "repositories/alice/ledger/locks/issue-7.lock"))
+                self.assertFalse((runtime_dir(root, run_id="cleanup-test") / "broker.sock").exists())
+
+    def test_long_utf8_helper_request_preserves_unauthorized_semantics(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d) / ("é" * 28)
+            root.mkdir()
+            proc, handle, _ = self.start(root)
+            try:
+                self.assertEqual(self.request(handle, "status")["status"], "held")
+                self.assertEqual(
+                    self.request(handle, "status", token=secrets.token_hex(32))["error"],
+                    "unauthorized",
+                )
+                self.assertEqual(self.request(handle, "release")["status"], "released")
+            finally:
+                if proc.poll() is None:
+                    proc.kill(); proc.wait(timeout=3)
+                proc.stdout.close(); proc.stderr.close()
+
+    def test_startup_failures_release_owned_resources_without_masking_original(self):
+        """Each setup boundary must leave no owned FD, socket, lock, or token."""
+        import contextlib
+        import io
+
+        failures = ("metadata", "runtime-fd", "handle-write", "socket-bind")
+        for failure in failures:
+            with self.subTest(failure=failure), tempfile.TemporaryDirectory() as d:
+                root = Path(d)
+                ensure_cache(root)
+                locks = root / "repositories/alice/ledger/locks"
+                locks.mkdir(parents=True); locks.chmod(0o700)
+                (locks / "repository.lock").touch(mode=0o600)
+                (locks / "issue-7.lock").touch(mode=0o600)
+                runtime = runtime_dir(root, run_id=f"failure-{failure}")
+                args = Namespace(root=str(root), owner="alice", repository="ledger", issue=7,
+                                 role="gitea-impl-agent", run_id=f"failure-{failure}",
+                                 handle=str(runtime / "broker.handle"),
+                                 socket=str(runtime / "broker.sock"), ttl_seconds=5.0,
+                                 acquire_timeout=0.0)
+                instance = broker.Broker(args)
+                original_write_all = broker._write_all
+                original_chmod = broker.os.chmod
+                calls = {"write": 0}
+
+                def partial_write(fd, data):
+                    calls["write"] += 1
+                    if calls["write"] == 1:
+                        os.write(fd, data[:1])
+                        raise OSError("injected handle write")
+                    return original_write_all(fd, data)
+
+                def late_chmod(path, mode, **kwargs):
+                    original_chmod(path, mode, **kwargs)
+                    if path == "broker.sock":
+                        raise OSError("injected socket setup")
+
+                with contextlib.ExitStack() as stack:
+                    if failure == "metadata":
+                        stack.enter_context(patch.object(instance.set.issue, "write_metadata",
+                                                         side_effect=OSError("injected metadata")))
+                    elif failure == "runtime-fd":
+                        stack.enter_context(patch.object(broker.os, "dup",
+                                                         side_effect=OSError("injected dup")))
+                    elif failure == "handle-write":
+                        stack.enter_context(patch.object(broker, "_write_all",
+                                                         side_effect=partial_write))
+                    else:
+                        stack.enter_context(patch.object(broker.os, "chmod",
+                                                         side_effect=late_chmod))
+                    with contextlib.redirect_stdout(io.StringIO()):
+                        with self.assertRaises(OSError):
+                            instance.run()
+                self.assertTrue(probe_free(locks / "issue-7.lock"))
+                self.assertTrue(probe_free(locks / "repository.lock"))
+                self.assertIsNone(instance.socket_parent_fd)
+                self.assertIsNone(instance.server)
+                if failure in ("handle-write", "socket-bind"):
+                    terminal = json.loads(instance.handle.read_text())
+                    self.assertNotIn("token", terminal)
+                    self.assertIn(terminal["status"], ("released", "cleanup-incomplete"))
+                self.assertFalse((runtime / "broker.sock").exists())
+
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()