runtime: broker activation-ping handshake fix + cache-stable bundle-root predicate

plugins/agent-collab/runtime_bundle.py

@@ -325,6 +325,29 @@ def _validate_inspection(value: Any, record: Mapping[str, Any]) -> None:
             _raise("runtime bundle member inspection changed")
 
 
+def _bundle_root_mode_ok(mode: int) -> bool:
+    """Cache-stable bundle-root directory predicate.
+
+    Host plugin managers re-extract the installed bundle and normalize its directory
+    mode (0o755) on install AND on every autoUpdate/restart, so the build-store's exact
+    0o500 can never be preserved on the loaded copy. The real tamper guarantee is the
+    whole-bundle + per-member digest and signature verified below — NOT the root dir
+    mode — so the root predicate requires only owner read+execute, NO group/other write,
+    and NO setuid/setgid/sticky, accepting both 0o500 (build store) and host-normalized
+    0o755 while still rejecting 0o775 / 0o757 / 0o777 and special-bit modes. Accepted
+    residual (truthful): a same-UID owner CAN chmod/mutate the tree; that mutation is
+    detected at resolution by the digest/signature, and exact 0o500 was never a same-UID
+    TOCTOU boundary either (the owner could always chmod it). This matches the
+    operator-approved same-UID reliability posture.
+    """
+    perms = stat.S_IMODE(mode)
+    return (
+        (perms & 0o500) == 0o500  # owner read + execute present
+        and (perms & 0o022) == 0  # no group/other WRITE
+        and (perms & 0o7000) == 0  # no setuid / setgid / sticky
+    )
+
+
 def verify_bundle_tree(
     root: Path,
     records: Any,
@@ -356,7 +379,7 @@ def verify_bundle_tree(
             or stat.S_ISLNK(lexical_root.st_mode)
             or _stat_identity(root_before) != _stat_identity(lexical_root)
             or root_before.st_uid != os.geteuid()
-            or stat.S_IMODE(root_before.st_mode) != INSTALL_MODE
+            or not _bundle_root_mode_ok(root_before.st_mode)
         ):
             _raise("runtime bundle root is unsafe")
         try:

plugins/agent-collab/runtime_client.py

@@ -1567,11 +1567,34 @@ def _broker_job_loaded() -> bool:
     ).returncode == 0
 
 
+# The activation liveness knock must outlast the signed runtime's cold start.
+# A freshly published bundle (new digest directory, non-page-cached files)
+# cold-starts in 6-12s on first exec -- Gatekeeper/codesign validation plus
+# interpreter init for a ~20MB standalone bundle (unified-log evidence:
+# first-activation broker runs of 6302ms-12111ms; warm runs ~1.4s). A 5s
+# budget made the first activation after every fresh publish spuriously fail
+# ("activation ping failed") and fall through to the restore path, which
+# skips the ping and so silently masked the miscalibration.
+BROKER_COLD_START_TIMEOUT_SECONDS = 30.0
+
+
 def _broker_ping(socket_path: Path) -> bool:
     try:
         with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as peer:
-            peer.settimeout(5.0)
+            peer.settimeout(BROKER_COLD_START_TIMEOUT_SECONDS)
             peer.connect(str(socket_path))
+            # Hold the connection open through the broker's peer-identity
+            # observation, then half-close the write side so the broker's
+            # frame read returns end-of-stream promptly and it exits. A bare
+            # connect-and-close races socket activation: the peer can be gone
+            # before the broker reads LOCAL_PEERPID (ENOTCONN -> peer_error),
+            # and the broker would otherwise block in accept() for its full
+            # timeout instead of servicing this liveness knock.
+            peer.shutdown(socket.SHUT_WR)
+            try:
+                peer.recv(1)
+            except OSError:
+                pass
         return _wait_for_broker_exit()
     except OSError:
         return False
@@ -1581,10 +1604,15 @@ def _broker_process_idle() -> bool:
     result = _launchctl(["print", f"gui/{os.getuid()}/{BROKER_LABEL}"])
     if result.returncode != 0:
         return False
+    # launchctl prints a `state = active` line for each listening socket
+    # endpoint (always active while the job is bootstrapped). Those are not
+    # the job's lifecycle state; including them made the terminal-state check
+    # below never hold, so the activation ping never observed the broker exit.
     states = [
         line.split("=", 1)[1].strip().casefold()
         for line in result.stdout.splitlines()
         if line.strip().startswith("state =")
+        and line.split("=", 1)[1].strip().casefold() != "active"
     ]
     has_live_pid = any(
         line.strip().startswith("pid =")
@@ -1598,7 +1626,9 @@ def _broker_process_idle() -> bool:
 
 
 def _wait_for_broker_exit() -> bool:
-    deadline = time.monotonic() + 5.0
+    # Success returns as soon as the job reaches a terminal state; the
+    # cold-start budget only bounds the failure case (see the constant above).
+    deadline = time.monotonic() + BROKER_COLD_START_TIMEOUT_SECONDS
     while time.monotonic() < deadline:
         if _broker_process_idle():
             return True

tests/test_agent_collab_runtime_client.py

@@ -997,6 +997,12 @@ def test_broker_idle_requires_no_live_pid_and_an_explicit_stopped_state(self) ->
             ("state = running\npid = 123\n", False),
             ("pid = 123\n", False),
             ("", False),
+            # A stopped job still lists its listening socket endpoints as
+            # `state = active`; those must not veto the terminal job state.
+            ("state = not running\nstate = active\nstate = active\n", True),
+            # A running socket-activated job reports a live pid alongside the
+            # active socket endpoints.
+            ("state = running\npid = 123\nstate = active\n", False),
         )
         for output, expected in cases:
             with self.subTest(output=output), mock.patch.object(
@@ -1006,6 +1012,94 @@ def test_broker_idle_requires_no_live_pid_and_an_explicit_stopped_state(self) ->
             ):
                 self.assertEqual(self.client._broker_process_idle(), expected)
 
+    def test_broker_ping_half_closes_so_the_broker_observes_the_peer(self) -> None:
+        events: list[str] = []
+
+        class _FakePeer:
+            def __enter__(self_inner):
+                return self_inner
+
+            def __exit__(self_inner, *exc):
+                return False
+
+            def settimeout(self_inner, _value):
+                pass
+
+            def connect(self_inner, _address):
+                events.append("connect")
+
+            def shutdown(self_inner, how):
+                events.append(f"shutdown:{how}")
+
+            def recv(self_inner, _count):
+                events.append("recv")
+                return b""
+
+        with mock.patch.object(self.client.socket, "socket", return_value=_FakePeer()), \
+                mock.patch.object(self.client, "_wait_for_broker_exit", return_value=True):
+            self.assertTrue(self.client._broker_ping(Path("/tmp/broker.sock")))
+        # The peer connects, half-closes the write side (so the broker's frame
+        # read hits end-of-stream), and waits for the broker to close before
+        # polling for exit.
+        self.assertEqual(
+            events, ["connect", f"shutdown:{self.client.socket.SHUT_WR}", "recv"]
+        )
+
+    def test_broker_ping_hold_open_covers_the_runtime_cold_start(self) -> None:
+        # The activation ping's socket timeout is the hold-open window: the
+        # broker must observe a live peer before the client's recv gives up.
+        # A freshly published bundle cold-starts in 6-12s (first-exec
+        # signature validation + interpreter init on non-page-cached files),
+        # so a short hold recreates the ENOTCONN peer-gone race on every
+        # first activation after a fresh publish.
+        timeouts: list[float] = []
+
+        class _FakePeer:
+            def __enter__(self_inner):
+                return self_inner
+
+            def __exit__(self_inner, *exc):
+                return False
+
+            def settimeout(self_inner, value):
+                timeouts.append(value)
+
+            def connect(self_inner, _address):
+                pass
+
+            def shutdown(self_inner, _how):
+                pass
+
+            def recv(self_inner, _count):
+                return b""
+
+        with mock.patch.object(self.client.socket, "socket", return_value=_FakePeer()), \
+                mock.patch.object(self.client, "_wait_for_broker_exit", return_value=True):
+            self.assertTrue(self.client._broker_ping(Path("/tmp/broker.sock")))
+        self.assertEqual(timeouts, [self.client.BROKER_COLD_START_TIMEOUT_SECONDS])
+        self.assertGreaterEqual(self.client.BROKER_COLD_START_TIMEOUT_SECONDS, 15.0)
+
+    def test_wait_for_broker_exit_outlasts_the_runtime_cold_start(self) -> None:
+        # Simulated clock: the broker only reaches a terminal launchd state
+        # 12s after the poll starts (observed real cold-start upper bound).
+        # The exit wait must outlast that, or the primary activation path
+        # spuriously fails and the restore path masks it.
+        clock = [0.0]
+
+        def _monotonic() -> float:
+            return clock[0]
+
+        def _sleep(seconds: float) -> None:
+            clock[0] += seconds
+
+        def _idle() -> bool:
+            return clock[0] >= 12.0
+
+        with mock.patch.object(self.client.time, "monotonic", side_effect=_monotonic), \
+                mock.patch.object(self.client.time, "sleep", side_effect=_sleep), \
+                mock.patch.object(self.client, "_broker_process_idle", side_effect=_idle):
+            self.assertTrue(self.client._wait_for_broker_exit())
+
     def test_host_context_requires_the_exact_complete_codex_desktop_tuple(self) -> None:
         keys = self.client.CODEX_DESKTOP_TUPLE
         exact = dict(self.client.CODEX_DESKTOP_TUPLE.items())

tests/test_runtime_bundle.py

@@ -250,7 +250,9 @@ def test_wrong_root_or_member_mode_and_inspection_drift_fail_closed(self):
             root = Path(temporary) / "agent-collab-runtime.bundle"
             root.mkdir()
             records = self._create_bundle(root)
-            root.chmod(0o700)
+            # A group-writable root is a genuinely unsafe mode and must still fail closed
+            # (0o700 / host-normalized 0o755 are now ACCEPTED by the cache-stable predicate).
+            root.chmod(0o775)
             with self.assertRaises(rb.BundleContractError):
                 rb.verify_bundle_tree(root, records, inspector=self._inspector)
             root.chmod(0o500)