diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 371f9b51c7..5b755f8f60 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -63,11 +63,7 @@ jobs:
             stack/.venv/bin/python -m pytest tests/test_stack_native_compaction.py tests/test_stack_persistent_compaction.py --no-cov -q
 
   test:
-    runs-on: ${{ matrix.os }}
-    strategy:
-      fail-fast: false
-      matrix:
-        os: [ubuntu-latest, macos-latest]
+    runs-on: ubuntu-latest
 
     steps:
       - name: Checkout
@@ -104,18 +100,10 @@ jobs:
         shell: bash
         run: npm ci --ignore-scripts --omit=dev --no-audit --no-fund
 
-      - name: Run tests with coverage (85% enforced on Linux)
-        if: matrix.os == 'ubuntu-latest'
+      - name: Run tests with coverage (85% enforced)
         run: |
           pytest -n auto --maxprocesses=8 --dist=worksteal --cov=agent_comms --cov-report=xml --cov-report=term-missing -v
 
-      # Sealed /var/tmp, /proc and SO_PEERCRED fixtures run only on Linux.
-      # Keep macOS test execution, but do not enforce Linux coverage there.
-      - name: Run tests (coverage reported, Linux-only fixtures skipped)
-        if: matrix.os != 'ubuntu-latest'
-        run: |
-          pytest -n auto --maxprocesses=8 --dist=worksteal --cov=agent_comms --cov-fail-under=0 --cov-report=xml --cov-report=term-missing -v
-
   build:
     runs-on: ubuntu-latest
     needs: [test]
diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py
index 0df850a317..c265842654 100644
--- a/src/agent_comms/child_process.py
+++ b/src/agent_comms/child_process.py
@@ -1856,13 +1856,14 @@ def launching(
         stdin: int | BinaryIO | TextIO = subprocess.DEVNULL,
         output: int | BinaryIO | TextIO = subprocess.DEVNULL,
         stderr: int | BinaryIO | TextIO = subprocess.STDOUT,
+        pass_fds: tuple[int, ...] = (),
         runtime: LaunchRuntime = LaunchRuntime.running(),
         lifetime: type[ChildLifetime] = DiesWithLauncherLifetime,
     ) -> Generator[SpawnedProcess]:
         """Yield the child before it executes; leaving the body releases it, raising reaps it."""
         if not command:
             raise ValueError("A child command is required")
-        with Platform.current().launch(command, (), runtime, lifetime) as launch:
+        with Platform.current().launch(command, pass_fds, runtime, lifetime) as launch:
             child = launch.spawn(cwd=cwd, env=env, stdin=stdin, stdout=output,
                                  stderr=stderr)
             try:
diff --git a/src/agent_comms/goal_attempts.py b/src/agent_comms/goal_attempts.py
index 927078767a..b6cfb69392 100644
--- a/src/agent_comms/goal_attempts.py
+++ b/src/agent_comms/goal_attempts.py
@@ -751,23 +751,26 @@ def _sync(self) -> None:
         AtomicFile(self.path).sync()
 
     @contextmanager
-    def writing(self) -> Iterator[sqlite3.Connection]:
-        with closing(self._connect()) as conn, self.transaction(conn):
+    def writing(self, committed: Callable[[], None] = lambda: None) -> Iterator[sqlite3.Connection]:
+        with closing(self._connect()) as conn, self.transaction(conn, committed):
             yield conn
 
     def _change(
         self, operation: Callable[[sqlite3.Connection], _T], verify: Callable[[_T], bool]
     ) -> _T:
-        """A failed commit, sync, or readback is never a launch authorization."""
-        try:
-            with self.writing() as conn:
-                result = operation(conn)
+        """A failed commit, sync, or readback is never a launch authorization, and is never
+        announced: the fsync and readback run before the change is published."""
+        def attest() -> None:
             self._sync()
             try:
                 if not verify(result):
                     raise StorageUncertainError("Goal attempt readback does not attest the write.")
             except sqlite3.Error as error:
                 raise StorageUncertainError("Goal attempt readback failed.") from error
+
+        try:
+            with self.writing(attest) as conn:
+                result = operation(conn)
         except errors.GoalAttemptError:
             raise
         except (sqlite3.Error, OSError) as error:
diff --git a/src/agent_comms/pi_commands.py b/src/agent_comms/pi_commands.py
index 1858f669d7..623057a6b9 100644
--- a/src/agent_comms/pi_commands.py
+++ b/src/agent_comms/pi_commands.py
@@ -249,7 +249,9 @@ def on_response(
             yield session.context_info()
 
 
-CATALOG_TIMEOUT = 10  # seconds for one metadata child to start and answer
+# Seconds for one metadata child to start and answer. An idle Pi boot answers in 6-7 s, so
+# 10 s timed out `fork --model` on a loaded host (CI, a machine running tests).
+CATALOG_TIMEOUT = 30
 
 
 class CatalogQuery(NativeQuery):
diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py
index ebbd1bab67..b99898725b 100644
--- a/src/agent_comms/typed_table.py
+++ b/src/agent_comms/typed_table.py
@@ -2199,14 +2199,18 @@ def store(self) -> WrittenFile:
         """The file whose kind permits writes and whose committed row changes are published."""
 
     @contextmanager
-    def transaction(self, connection: sqlite3.Connection) -> Iterator[sqlite3.Connection]:
-        """Publish through the physical store after its row transaction commits. Maintenance
-        is asked under the write lock, so a hold taken before this transaction refuses it and
-        an installer's drain of the write lock waits out every transaction that passed."""
+    def transaction(
+        self, connection: sqlite3.Connection, committed: Callable[[], None] = lambda: None,
+    ) -> Iterator[sqlite3.Connection]:
+        """Publish through the physical store after its row transaction commits and
+        ``committed`` returns; a raise there announces nothing. Maintenance is asked under
+        the write lock, so a hold taken before this transaction refuses it and an
+        installer's drain of the write lock waits out every transaction that passed."""
         with self.publishing_changes(connection, self.store.path):
             with self.write_transaction(connection):
                 self.store.require_maintenance_open()
                 yield connection
+            committed()
 
     @abstractmethod
     def writing(self) -> AbstractContextManager[sqlite3.Connection]:
diff --git a/tests/guards/test_child_launch_ownership.py b/tests/guards/test_child_launch_ownership.py
index e8c4743839..c05ff8db45 100644
--- a/tests/guards/test_child_launch_ownership.py
+++ b/tests/guards/test_child_launch_ownership.py
@@ -13,7 +13,7 @@
 pytestmark = pytest.mark.refactor_guard
 
 ROOT = Path(__file__).resolve().parents[2]
-RAW_LAUNCHES = 33
+RAW_LAUNCHES = 32
 
 
 class RawLaunch(ast.NodeVisitor):
diff --git a/tests/test_carry.py b/tests/test_carry.py
index f0b7df3df1..8d7ffd86ff 100644
--- a/tests/test_carry.py
+++ b/tests/test_carry.py
@@ -278,24 +278,21 @@ def test_the_install_start_stops_owners_on_a_ledger_the_outgoing_runtime_cannot_
     """Install 14: f23ca8968 refused this ledger at every opening, so its stop answered
     nothing. The install's start holds the root, and the target stops the owners under
     that hold, carrying the ledger, which stays held until the install releases it."""
-    from agent_comms.child_process import ProcessIdentity
+    from agent_comms.child_process import SpawnedProcess
 
     path = ledger_one_failure_short(comms.root)
     maintenance = MaintenanceBarrier(RegistryStore.at(comms.root).path)
-    with subprocess.Popen(("sleep", "600")) as owner:
-        try:
-            comms.registry.declare(replace(Thread("unit", frozenset(), str(comms.root)),
-                                           process_identity=ProcessIdentity.capture(owner.pid)),
-                                   RunningThreadStatus())
-            with ExistingBusInstall.owners_stopped(comms.root, stack, stack) as owners:
-                assert owners.stopped == ["unit"]
-                assert stack.running_owners(comms.root) == []
-                assert maintenance.read().phase is DrainingPhase
-                owners.retired()
-                owners.release()
-                assert maintenance.read().phase is ReadyPhase
-        finally:
-            owner.kill()
+    with SpawnedProcess.launch(("sleep", "600")) as owner:
+        comms.registry.declare(replace(Thread("unit", frozenset(), str(comms.root)),
+                                       process_identity=owner.identity),
+                               RunningThreadStatus())
+        with ExistingBusInstall.owners_stopped(comms.root, stack, stack) as owners:
+            assert owners.stopped == ["unit"]
+            assert stack.running_owners(comms.root) == []
+            assert maintenance.read().phase is DrainingPhase
+            owners.retired()
+            owners.release()
+            assert maintenance.read().phase is ReadyPhase
     GoalAttemptStore.at(path.parent)
 
 
diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py
index 7cc61e40e4..a42ebe5a54 100644
--- a/tests/test_end_to_end.py
+++ b/tests/test_end_to_end.py
@@ -15,6 +15,10 @@
 from agent_comms.read_basis import NamedViewer
 from native_backend_fixture import native_backend_fixture
 
+# Each test boots native Pi and chains CLI children; `fork` alone boots Pi twice (its
+# --model catalog through GetAvailableModels, then the thread), so 60 s is not a budget.
+pytestmark = pytest.mark.timeout(180)
+
 
 @pytest.fixture
 async def native(tmp_path, native_runtime_package):
@@ -27,7 +31,7 @@ async def cli(programs, root, *arguments):
     # wire are one runtime: a root refuses a second one (root_runtime.py:56).
     result = await BoundedRun.run(
         (sys.executable, str(programs.script(main)), '--root', str(root), *arguments),
-        timeout=20, env=dict(os.environ),
+        timeout=60, env=dict(os.environ),
     )
     assert result.outcome.successful, (result.stdout + result.stderr).decode()
     return json.loads(result.stdout)
diff --git a/tests/test_held_sends.py b/tests/test_held_sends.py
index b599cf4dde..111522dcc3 100644
--- a/tests/test_held_sends.py
+++ b/tests/test_held_sends.py
@@ -1,4 +1,5 @@
-"""Real send entrypoints wait through a private bus hold and publish once."""
+"""CLI send entrypoints wait through a private bus hold and publish once; native tool
+publications refuse while held (test_connector.test_publishing_tool_held_mid_call_refuses_at_once)."""
 
 
 import asyncio
@@ -17,8 +18,6 @@
 from agent_comms.errors import ToolRefusedError
 from agent_comms.maintenance_barrier import MaintenanceOperator
 from agent_comms.store_files import StoreLockContention, _store_lock
-from agent_comms.tool_caller import ThreadCaller
-from agent_comms.tools import invoke_tool
 from test_locked_store import event, process
 from agent_comms.threads import Thread
 
@@ -100,19 +99,3 @@ async def test_cli_child_waits_then_publishes_once(held_bus, arguments):
     assert child.returncode == 0, (stdout.decode(), stderr.decode())
     assert_published_once(comms)
     print(f'{arguments[0]} child: held without publication; exit 0; published exactly once; no refusal')
-
-
-async def test_native_thread_caller_waits_then_publishes_once(held_bus):
-    comms = held_bus
-    before = comms.bus.bus.revision()
-    pending = asyncio.create_task(asyncio.to_thread(
-        invoke_tool, comms, 'comms_send',
-        {'to': 'recipient', 'body': 'held publication'},
-        caller=ThreadCaller('sender')))
-    with pytest.raises(TimeoutError):
-        await asyncio.wait_for(asyncio.shield(pending), .25)
-    assert comms.bus.bus.revision() == before
-    comms.owners.maintenance.clear(comms.owners.maintenance.read())
-    await asyncio.wait_for(pending, 15)
-    assert_published_once(comms)
-    print('native ThreadCaller: held without publication; published exactly once; no refusal')
diff --git a/tests/test_install_rehearsal.py b/tests/test_install_rehearsal.py
index 63fb67f327..ecd4423519 100644
--- a/tests/test_install_rehearsal.py
+++ b/tests/test_install_rehearsal.py
@@ -65,6 +65,11 @@ def declare_storage(root: Path, scratch: Path) -> None:
         StorageRule(scratch_base=scratch, floor_gib=0))))
 
 
+# Every test here pays installed_home's private runtime build, so the first one a worker
+# runs would spend the 60 s default building it.
+pytestmark = pytest.mark.timeout(280)
+
+
 @pytest.fixture(scope="module")
 def installed_home(tmp_path_factory) -> dict[str, str]:
     """A home holding what an installed machine has: the installer's uv, and as the
@@ -165,7 +170,6 @@ def drop_one_index(root: Path) -> Path:
     raise AssertionError("No declared store has an index to drop")
 
 
-@pytest.mark.timeout(280)
 @pytest.mark.parametrize("repos, drifted", (
     ((), False), (("--repos",), True),
 ), ids=("kept-repos-current-stores", "given-repos-drifted-store"))
@@ -201,7 +205,6 @@ def test_install_rehearses_without_changing_the_bus(installed_comms, installed_h
     assert bus_files(comms.root) == before
 
 
-@pytest.mark.timeout(280)
 def test_rehearsal_refuses_changed_native_inputs(installed_comms, installed_home, install_checkout):
     """The target's own installer refuses a stale pin before rehearsing the bus."""
     checkout = install_checkout
@@ -225,7 +228,6 @@ def test_rehearsal_refuses_changed_native_inputs(installed_comms, installed_home
     assert "Native build inputs changed; rebuild with --update-pins" in installed.stderr
 
 
-@pytest.mark.timeout(280)
 def test_failed_live_migration_restores_data_checkout_and_runtime(
         installed_comms, installed_home, install_checkout, tmp_path):
     comms, checkout = installed_comms, install_checkout
@@ -279,7 +281,6 @@ def carry(self, *, apply):
     assert comms.registry.require("worker").name == "worker"
 
 
-@pytest.mark.timeout(280)
 def test_failed_publication_restores_absent_route_without_store_changes(
         installed_comms, installed_home, install_checkout, tmp_path):
     checkout = install_checkout
@@ -329,7 +330,6 @@ def test_failed_publication_restores_absent_route_without_store_changes(
     MaintenanceBarrier(RegistryStore.at(installed_comms.root).path).read().phase.new_sends.require_open()
 
 
-@pytest.mark.timeout(280)
 @pytest.mark.parametrize("fresh, live_kind", ((False, CarriedRun), (True, CreatedRun)),
                          ids=("retained-bus", "fresh-bus"))
 def test_installed_entry_reports_completion_after_target_source_retirement(
diff --git a/tests/test_install_session_bridge.py b/tests/test_install_session_bridge.py
index 9b165e9b98..828364d44c 100644
--- a/tests/test_install_session_bridge.py
+++ b/tests/test_install_session_bridge.py
@@ -14,7 +14,7 @@
 
 import pytest
 
-from agent_comms.child_process import ProcessIdentity
+from agent_comms.child_process import ProcessIdentity, SpawnedProcess
 from agent_comms.cli_bridges.monitor import (
     ClaimedReplacement, InspectSession, PauseSession, SessionInformation,
 )
@@ -361,19 +361,19 @@ def test_a_bridge_that_dies_during_pause_counts_as_stopped(tmp_path, bridge_home
     current.symlink_to(tmp_path / 'runtime', target_is_directory=True)
     monkeypatch.setenv('HOME', str(bridge_home))
     claim.socket_path.parent.mkdir(parents=True, exist_ok=True)
-    dying = subprocess.Popen((sys.executable, '-c', (
+    with SpawnedProcess.launch((sys.executable, '-c', (
         'import socket, sys; s = socket.socket(socket.AF_UNIX); s.bind(sys.argv[1]); '
         's.listen(); print(flush=True); peer, _ = s.accept(); peer.makefile("rb").readline()'),
-        str(claim.socket_path)), stdout=subprocess.PIPE)
-    dying.stdout.readline()
-    me = ProcessIdentity.capture(os.getpid())
-    observed = SessionInformation(owner=me, process=ProcessIdentity.capture(dying.pid),
-                                  runtime=LaunchRuntime.running(),
-                                  transcript=tmp_path / 'dying-session.jsonl')
-    bridges = SessionBridgeResumes()
-    with bridges.holding((SessionBridge(claim, observed),)):
-        pass
-    assert dying.wait(10) == 0
+        str(claim.socket_path)), output=subprocess.PIPE) as dying:
+        dying.process.stdout.readline()
+        me = ProcessIdentity.capture(os.getpid())
+        observed = SessionInformation(owner=me, process=dying.identity,
+                                      runtime=LaunchRuntime.running(),
+                                      transcript=tmp_path / 'dying-session.jsonl')
+        bridges = SessionBridgeResumes()
+        with bridges.holding((SessionBridge(claim, observed),)):
+            pass
+        assert dying.process.wait(10) == 0
     assert bridges.failures == ()
     assert bridges.lines() == ('Session bridge dying-session: bridge exited during pause; '
                                'left stopped',)
diff --git a/tests/test_installer_route_lifetime.py b/tests/test_installer_route_lifetime.py
index 8890e46b45..1b1ae6724d 100644
--- a/tests/test_installer_route_lifetime.py
+++ b/tests/test_installer_route_lifetime.py
@@ -21,6 +21,7 @@
     route_directory,
 )
 from agent_comms.bus import Bus
+from agent_comms.child_process import SpawnedProcess
 from agent_comms.comms import Comms, wire
 from agent_comms.installer.backup import StoreBackup
 from agent_comms.installer.refusals import UpgradeRefused
@@ -262,24 +263,15 @@ def drain():
 """
 
 
-def route_holder(directory, program):
-    directory.mkdir(mode=0o700)
-    holder = subprocess.Popen(
-        (sys.executable, "-c", ROUTE_HOLDER, str(directory), program),
-        stdout=subprocess.PIPE, text=True)
-    assert holder.stdout.readline() == "held\n"
-    return holder
-
-
 def test_route_publication_neither_waits_for_nor_ends_an_old_toad_holding_the_directory(
     tmp_path,
 ):
     """An old runtime's Toad still flocks the route directory; publishers no longer do."""
-    holder = route_holder(tmp_path / "state", "toad")
-    try:
-        with route_directory(tmp_path / "state" / "active-route.json"):
+    directory = tmp_path / "state"
+    directory.mkdir(mode=0o700)
+    with SpawnedProcess.launch((sys.executable, "-c", ROUTE_HOLDER, str(directory), "toad"),
+                               output=subprocess.PIPE) as holder:
+        assert holder.process.stdout.readline() == b"held\n"
+        with route_directory(directory / "active-route.json"):
             pass
-        assert holder.poll() is None
-    finally:
-        holder.kill()
-        holder.wait()
+        assert holder.process.poll() is None
diff --git a/tests/test_native_evidence_boundary.py b/tests/test_native_evidence_boundary.py
index e7eccce464..9de0f98d72 100644
--- a/tests/test_native_evidence_boundary.py
+++ b/tests/test_native_evidence_boundary.py
@@ -6,7 +6,7 @@
 
 from agent_comms.native_entries import NativeEntry
 from agent_comms.native_failures import NativePiUnavailable
-from agent_comms.pi_payloads import PiOmitted, TextContent, UnknownContent
+from agent_comms.pi_payloads import PiOmitted, ReportedText, TextContent, UnknownContent
 
 
 def test_original_header_preserves_project_and_fork_lineage(tmp_path):
@@ -25,7 +25,7 @@ def test_original_header_preserves_project_and_fork_lineage(tmp_path):
     with NativeEntry.open_evidence(file) as evidence:
         header, entries = evidence.observe()
         assert entries == (header,)
-        assert header.cwd == raw["cwd"]
+        assert header.cwd == ReportedText(raw["cwd"])
         assert header.parent_session == raw["parentSession"]
         encoded = header.to_wire()
         assert all(encoded[name] == value for name, value in raw.items())
diff --git a/tests/test_store_guards.py b/tests/test_store_guards.py
index ad60dedff4..7701b93569 100644
--- a/tests/test_store_guards.py
+++ b/tests/test_store_guards.py
@@ -5,11 +5,12 @@
 import sqlite3
 import subprocess
 import sys
-from contextlib import closing, contextmanager
+from contextlib import ExitStack, closing, contextmanager
 from pathlib import Path
 
 import pytest
 
+from agent_comms.child_process import SpawnedProcess
 from agent_comms.locked_store import SQLiteFile
 from agent_comms.store_files import ClaimHeldError, ProcessSingleton, StoreBusy, StoreLockContention
 
@@ -38,27 +39,18 @@ def opening():
 
 @contextmanager
 def _exec_adopter(claim):
-    with claim.claimed() as held:
-        child = subprocess.Popen(
-            (sys.executable, __file__, str(claim.path), str(held.fileno())),
-            pass_fds=(held.fileno(),), stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-        )
-        try:
-            assert select.select((child.stdout,), (), (), 10)[0], "adopter did not start"
-            assert child.stdout.readline() == "held\n"
-        except BaseException:
-            child.kill()
-            child.communicate(timeout=10)
-            raise
-    try:
+    with ExitStack() as owned:
+        with claim.claimed() as held:
+            child = owned.enter_context(SpawnedProcess.launch(
+                (sys.executable, __file__, str(claim.path), str(held.fileno())),
+                pass_fds=(held.fileno(),), stdin=subprocess.PIPE,
+                output=subprocess.PIPE, stderr=subprocess.PIPE,
+            ))
+            assert select.select((child.process.stdout,), (), (), 10)[0], "adopter did not start"
+            assert child.process.stdout.readline() == b"held\n"
         yield
-        _, errors = child.communicate("release\n", timeout=10)
-        assert child.returncode == 0, errors
-    finally:
-        if child.poll() is None:
-            child.kill()
-            child.communicate(timeout=10)
+        _, errors = child.process.communicate(b"release\n", timeout=10)
+        assert child.process.returncode == 0, errors
 
 
 def test_process_singleton_refuses_competitors_and_keeps_an_exec_adoption(tmp_path):