Fix the ownerless main CI reds; Linux-only CI

.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]

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:

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:

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):

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]:

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):

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)
 
 

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)

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')

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(

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',)

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

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())