fix: make the macOS CI cell green

docs/reference/remote-execution.md

@@ -600,7 +600,9 @@ machine, humanize books a meeting:
    else `$HUMANIZE_RENDEZVOUS`, else this machine's outward-facing IPv4 address (the source
    address of a route to `192.0.2.1`; `127.0.0.1` if there is none).
 3. It spawns `hmz internal anchor --target peer://TICKET@HOST:PORT` on the harness machine.
-4. Both halves dial the broker with the ticket and their role (`anchor` or `serve`).
+4. Both halves dial the broker with the ticket and their role (`anchor` or `serve`), and offer
+   the address they dialled from plus the IPv4 addresses their hostname resolves to -- left
+   out where resolving it takes longer than 1 s, as an unanswered macOS `.local` name does.
 
 | Phase | Behaviour | Time limit |
 | --- | --- | --- |

src/hmz/cli/anchor.py

@@ -135,7 +135,9 @@ def anchor(argv: list[str]) -> int:
         # this machine has not got, is a different thing gone wrong and is answered below.
         print(f"hmz: {exc.strerror or exc}", file=sys.stderr)
         return 127
-    except (ConnectionError, ProtocolError, OSError, ValueError) as exc:
+    except (ConnectionError, ProtocolError, OSError, RuntimeError, ValueError) as exc:
+        # RuntimeError for a host that cannot intercept syscalls at all -- a Mac -- which is
+        # said in a sentence of its own rather than a traceback ending in one.
         print(f"hmz: {exc}", file=sys.stderr)
         return 1
     else:

src/hmz/coganchor/rendezvous.py

@@ -95,6 +95,9 @@
 #: connection to something that is not listening for us.
 GREETING = 3.0
 
+#: How long a half waits for its own machine's name to resolve before going without it.
+_NAMING = 1.0
+
 #: How long a half waits on the broker's own answers, which are sent as soon as it has them.
 _ANSWERING = 30.0
 
@@ -1008,16 +1011,38 @@ def _mine(home: socket.socket, port: int) -> list[tuple[str, int]]:
     where the broker is somewhere else entirely and the two halves are neighbours.
     """
     addresses = [str(home.getsockname()[0])]
-    with contextlib.suppress(OSError):
-        for found in socket.getaddrinfo(
-            socket.gethostname(), port, socket.AF_INET, socket.SOCK_STREAM
-        ):
-            address = str(found[4][0])
-            if address not in addresses:
-                addresses.append(address)
+    for address in _named(port):
+        if address not in addresses:
+            addresses.append(address)
     return [(address, port) for address in addresses]
 
 
+def _named(port: int) -> list[str]:
+    """This machine's addresses by its own name, or none if the name is slow to say them.
+
+    Asked of the resolver, which may not know the name at all: a Mac's `<name>.local` is
+    looked up over multicast DNS and, where nothing answers, takes over half a minute to fail
+    -- longer than the broker waits for this half's first word, so a half that waited for it
+    would arrive to find the broker gone. The names are a nicety for neighbours, and are left
+    out rather than waited for.
+    """
+    found: list[str] = []
+    done = threading.Event()
+
+    def asking() -> None:
+        with contextlib.suppress(OSError):
+            found.extend(
+                str(one[4][0])
+                for one in socket.getaddrinfo(
+                    socket.gethostname(), port, socket.AF_INET, socket.SOCK_STREAM
+                )
+            )
+        done.set()
+
+    threading.Thread(target=asking, name="rendezvous-names", daemon=True).start()
+    return list(found) if done.wait(_NAMING) else []
+
+
 def _pairs(said: Any) -> list[tuple[str, int]]:
     """Reads a candidate list off the wire, keeping only the addresses that can be punched to.
 

src/hmz/runtime/flowing/environing.py

@@ -454,7 +454,15 @@ def started_error(
       `EnvUnavailable` for a workdir that is not there, and otherwise what
       :func:`env_error` makes of it.
     """
-    if error.filename == cwd and error.errno in (errno.ENOENT, errno.ENOTDIR):
+    from hmz.coganchor.proto import path_key
+
+    # Matched as the machine would name it: a Mac's `/private/var` comes back as `/var`.
+    named = error.filename
+    if (
+        isinstance(named, str)
+        and path_key(named) == path_key(cwd)
+        and error.errno in (errno.ENOENT, errno.ENOTDIR)
+    ):
         return EnvUnavailable(f"the workdir {cwd} is not there")
     return env_error(error, f"could not run {argv[0]!r} in {cwd}", reported=reported)
 

tests/integration/agents/test_agy_stream.py

@@ -25,7 +25,7 @@
     Failed,
 )
 from hmz.coganchor.agents.event import Usage
-from hmz.coganchor.providers import Provider
+from hmz.coganchor.providers import Provider, redirect
 from tests.agents import standins
 from tests.stubs import HereAnchor
 
@@ -479,14 +479,17 @@ def node() -> Provider:
     def swaps(_provider: Provider) -> tuple[tuple[str, str], ...]:
         return (("/native/agy/token.json", str(token)),)
 
-    def command(_provider: Provider, argv: list[str] | tuple[str, ...]) -> list[str]:
-        # The supervisor answering those paths is tested where it lives; what is in
-        # question here is only whether the file's contents reach the restart decision.
+    def command(
+        _swaps: object, argv: list[str] | tuple[str, ...], *_: object
+    ) -> list[str]:
+        # The supervisor answering those paths is tested where it lives -- and only where a
+        # ptrace supervisor can run; what is in question here is only whether the file's
+        # contents reach the restart decision.
         return list(argv)
 
     monkeypatch.setattr(agy.agent, "node", node)
     monkeypatch.setattr(Provider, "swaps", swaps)
-    monkeypatch.setattr(Provider, "command", command)
+    monkeypatch.setattr(redirect, "command", command)
     session = agy.agent.new()
     session("first")
     token.write_text("refreshed credential")

tests/integration/agents/test_watchdog.py

@@ -272,7 +272,13 @@ def test_a_process_that_was_stopped_is_named_as_stopped() -> None:
     session = agent.new()
     assert isinstance(session, _Stream)
     proc = session._start(session._command())
-    psutil.Process(proc.pid).suspend()
+    stopped = psutil.Process(proc.pid)
+    stopped.suspend()
+    # Stopped once the kernel says so, not once it was asked: a signal is taken when the
+    # process next runs, which on a loaded Mac can be past the first look.
+    deadline = time.monotonic() + 10
+    while stopped.status() != psutil.STATUS_STOPPED and time.monotonic() < deadline:
+        time.sleep(0.05)
     watch = Watchdog(session, riding=lambda: proc, window=0.2)
     try:
         with watch:

tests/integration/coganchor/test_fence_proxy.py

@@ -182,7 +182,13 @@ def test_connections_past_the_limit_are_turned_away(upstream: int) -> None:
     with Proxy(["localhost"], ports=[upstream], limit=1) as proxy:
         held, answer = _ask(proxy, f"CONNECT localhost:{upstream} HTTP/1.1\r\n\r\n")
         assert answer.startswith(b"HTTP/1.1 200")
-        extra, answer = _ask(proxy, f"CONNECT localhost:{upstream} HTTP/1.1\r\n\r\n")
+        # Closed unanswered: an end of file where the request had not yet arrived when it
+        # was, a reset where it had -- which is what a Mac's kernel says of unread bytes.
+        with socket.create_connection(("127.0.0.1", proxy.port), timeout=10) as extra:
+            extra.sendall(f"CONNECT localhost:{upstream} HTTP/1.1\r\n\r\n".encode())
+            try:
+                answer = extra.recv(4096)
+            except ConnectionResetError:
+                answer = b""
         assert answer == b""
-        extra.close()
         held.close()

tests/integration/flows/test_local_envs.py

@@ -14,8 +14,10 @@
 import asyncio
 import fcntl
 import os
+import shlex
 import stat
 import subprocess
+import sys
 import time
 from pathlib import Path, PurePosixPath
 
@@ -50,6 +52,7 @@
 from hmz.runtime.flowing.specs import parse_envs
 from hmz.runtime.flowing.spi import ENV_CAPABILITIES, Placement
 from tests.flows.contracts import check_env_driver
+from tests.stubs import CPUS
 
 #: Whether this process may read and write whatever it likes, which makes a permission test
 #: a test of nothing.
@@ -196,7 +199,7 @@ def test_a_directory_that_is_not_there_is_unavailable(tmp_path: Path) -> None:
 async def test_what_the_machine_has_is_what_the_kernel_says(tmp_path: Path) -> None:
     driver = _driver(tmp_path)
     await probe(driver)
-    assert driver.cpu_count == len(os.sched_getaffinity(0))
+    assert driver.cpu_count == CPUS
     assert driver.memory == psutil.virtual_memory().total
     assert driver.gpu_count >= 0
     assert driver.gpu_memory >= 0
@@ -280,8 +283,13 @@ async def test_a_timeout_kills_what_outlived_the_command_in_its_group(
     assert await _gone(child), "what the command left holding its output outlived it"
 
 
-#: A daemon that leaves the command's group and session, and keeps its output open.
-_DAEMON = "setsid sh -c 'echo $$ > daemon; exec sleep 30' & echo started"
+#: A daemon that leaves the command's group and session, and keeps its output open. Python's
+#: `setsid` rather than the command, which a Mac has not got.
+_DAEMON = (
+    f"{shlex.quote(sys.executable)} -c 'import os; os.setsid(); "
+    'os.write(os.open("daemon", os.O_WRONLY | os.O_CREAT), b"%d" % os.getpid()); '
+    'os.execvp("sleep", ["sleep", "30"])\' & echo started'
+)
 
 
 async def test_a_daemon_holding_a_command_s_output_does_not_hold_its_timeout(

tests/integration/flows/test_ssh_envs.py

@@ -23,6 +23,7 @@
 
 from hmz.coganchor.machines import AnchoredConfig, store
 from hmz.coganchor.machines.store import SSHRuntime
+from hmz.coganchor.proto import path_key
 from hmz.flows import (
     EnvBackendKind,
     EnvCommandTimeout,
@@ -36,6 +37,7 @@
 from hmz.runtime.flowing.environments import open_env, probe
 from hmz.runtime.flowing.specs import parse_envs
 from tests.flows.contracts import check_env_driver
+from tests.stubs import CPUS
 
 if TYPE_CHECKING:
     from hmz.runtime.flowing.spi import EnvDriver
@@ -103,7 +105,12 @@ def _open(host: str, workdir: Path | str) -> MachineEnvDriver:
 
 
 def _mem_total() -> int:
-    """The memory this machine's kernel says it has, as `/proc/meminfo` counts it."""
+    """The memory this machine's kernel says it has, as `/proc/meminfo` counts it.
+
+    Or as `sysctl hw.memsize` does where there is no `/proc` (macOS), which is psutil's count.
+    """
+    if not Path("/proc/meminfo").exists():
+        return int(psutil.virtual_memory().total)
     meminfo = Path("/proc/meminfo").read_text().splitlines()
     (total,) = [line.split()[1] for line in meminfo if line.startswith("MemTotal:")]
     return int(total) * 1024
@@ -192,7 +199,7 @@ async def test_what_a_host_has_is_learned_when_it_is_probed(
     try:
         await probe(driver)
         assert driver.available
-        assert driver.cpu_count == len(os.sched_getaffinity(0))
+        assert driver.cpu_count == CPUS
         assert driver.memory == _mem_total()
         assert abs(driver.memory - psutil.virtual_memory().total) < 1024 * 1024
         assert driver.gpu_count >= 0
@@ -212,7 +219,10 @@ async def test_a_workdir_under_home_is_found_where_home_is_there(
         assert isinstance(before, AnchoredConfig)
         assert before.anchor.remote_path == "~/work"
         await probe(driver)
-        assert await driver.exec(["pwd"], timeout=30) == (0, f"{far}/work\n", "")
+        # Whichever name for it the host's `pwd` prints: a Mac's says `/var` for `/private/var`.
+        status, out, err = await driver.exec(["pwd"], timeout=30)
+        assert (status, err) == (0, "")
+        assert path_key(out.strip()) == path_key(f"{far}/work")
         await driver.write("made.txt", b"made")
         assert (far / "work/made.txt").read_bytes() == b"made"
         assert await driver.read("~/work/made.txt") == b"made"

tests/integration/flows/test_swarm_envs.py

@@ -42,6 +42,7 @@
 from hmz.runtime.flowing.specs import parse_envs
 from hmz.sdk import Hmz
 from tests.flows.contracts import check_env_driver
+from tests.machines.fixtures import containered
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -108,6 +109,7 @@ class Envs(EnvCollection):
 elif command == "run" and "--rm" in rest:
     print(said("STANDIN_OWNER", "4242 4343"))
 elif command == "exec":
+    os.environ["PATH"] = os.environ["STANDIN_CONTAINER"] + os.pathsep + os.environ["PATH"]
     moved = os.environ["STANDIN_ROOT"]
     words = [word.replace("/tmp/humanize", moved) for word in rest[2:]]
     os.execvp(words[0], words)
@@ -198,6 +200,7 @@ def swarm(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Swarm:
     monkeypatch.setenv("PATH", f"{bin_}{os.pathsep}{os.environ['PATH']}")
     monkeypatch.setenv("STANDIN_LOG", str(tmp_path / "docker.log"))
     monkeypatch.setenv("STANDIN_ROOT", str(tmp_path / "container-tmp"))
+    containered(tmp_path / "container-bin", monkeypatch)
     monkeypatch.setenv(
         "STANDIN_NODES",
         json.dumps(

tests/integration/machines/test_runtimes.py

@@ -18,6 +18,7 @@
 from hmz.coganchor.machines import store
 from hmz.coganchor.machines.store import DockerRuntime, SSHRuntime
 from hmz.sdk import Hmz
+from tests.stubs import CPUS
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -167,7 +168,7 @@ def test_an_ssh_host_checked_says_what_a_run_would_learn(standins: Path) -> None
 
     assert checked.reached, checked.said
     assert checked.home == os.environ["HOME"]
-    assert checked.cpus == len(os.sched_getaffinity(0))
+    assert checked.cpus == CPUS
     assert checked.memory > 0
     asked = standins.read_text()  # one ssh, carrying the probe a run makes
     assert asked.count("-T -o BatchMode=no") == 1

tests/integration/tui/test_runtimes.py

@@ -66,7 +66,7 @@
     picks,
     rows,
 )
-from tests.stubs import written
+from tests.stubs import CPUS, written
 from tests.tui.fixtures import until
 
 if TYPE_CHECKING:
@@ -275,7 +275,7 @@ async def test_an_ssh_host_is_added_on_one_form_and_asked_what_it_has(
         )
         # Asked what a run would ask it, down the road a run takes: its home, its CPUs.
         assert f"home {os.environ['HOME']}" in _under(app)
-        assert f"{len(os.sched_getaffinity(0))} CPUs" in _under(app)
+        assert f"{CPUS} CPUs" in _under(app)
         assert sheet.under() == "ssh/gpu"
         assert "me@gpu.example:2200 · key ~/.ssh/gpu" in _drawn(app)
         assert "ssh/gpu saved to" in "\n".join(sheet._told)