Make supervised tunnel startup deterministic

qualification/pre1-cases.json

@@ -144,6 +144,16 @@
         "tests/test_current_node_token.py::test_e050_client_credential_lifecycle_fixture"
       ]
     },
+    "dynamic-public-route-readiness": {
+      "assertion": "test_heartbeat_self_heals_from_empty_initial_token",
+      "command": [
+        "@python",
+        "-m",
+        "pytest",
+        "-q",
+        "tests/test_serve.py::test_heartbeat_self_heals_from_empty_initial_token"
+      ]
+    },
     "minimum-version": {
       "assertion": "test_minimum_python_version_is_declared_and_candidate_remains_pre1",
       "command": [

scripts/check_mypy_baseline.py

@@ -6,6 +6,7 @@
 import json
 import re
 import subprocess
+import sys
 from pathlib import Path
 
 ERROR = re.compile(r"^([^:]+):\d+: error: .*\[([^]]+)\]$")
@@ -26,7 +27,11 @@ def main() -> int:
     args = parser.parse_args()
     baseline = json.loads(args.baseline.read_text(encoding="utf-8"))
     result = subprocess.run(
-        ["mypy", "src"], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False
+        [sys.executable, "-m", "mypy", "src"],
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        check=False,
     )
     observed = observed_errors(result.stdout)
     allowed = collections.Counter(baseline["errors"])

src/iicp_client/cli.py

@@ -2150,7 +2150,12 @@ async def _serve(args: argparse.Namespace) -> int:
     # None is reserved for --skip-registration (no heartbeat by design).
     token: str | None = None
     if not args.skip_registration:
-        for attempt in range(1, 4):
+        # A dynamic public route becomes eligible only after the node is
+        # listening and the directory can dial it. One expected pre-listener
+        # attempt is enough; the empty-token heartbeat re-registers immediately
+        # after serve() binds.
+        registration_attempts = 1 if _tunnel is not None else 3
+        for attempt in range(1, registration_attempts + 1):
             try:
                 token = await node.register()
                 logger.info("Registered as %s (token=%s…)", node_id, (token or "")[:8])
@@ -2172,7 +2177,7 @@ async def _serve(args: argparse.Namespace) -> int:
                     _complete_handoff_for_node(args.node)
                 break
             except Exception as exc:  # noqa: BLE001
-                if attempt >= 3:
+                if attempt >= registration_attempts:
                     logger.warning(
                         "Registration failed after %d attempts: %s — starting heartbeat loop "
                         "anyway; it will re-register on the first 401",
@@ -3213,6 +3218,13 @@ def run_actions() -> None:
                 raise RuntimeError(f"service manager action failed ({completed.returncode}): {' '.join(action.argv)}")
 
     if cmd == "install":
+        from iicp_client.tunnel import cloudflared_path
+
+        if os.environ.get("IICP_CLOUDFLARED_PATH") is None and cloudflared_path() is None:
+            sys.stderr.write(
+                "WARNING: Quick Tunnel fallback is unavailable under the supervisor because "
+                "cloudflared could not be resolved. Direct reachability remains supported.\n"
+            )
         if dry_run:
             sys.stdout.write(f"# {unit.platform} service: {unit.name}\n# path: {unit.path}\n")
             sys.stdout.write(unit.content)

src/iicp_client/node.py

@@ -993,8 +993,13 @@ async def _heartbeat_loop(self, node_token: str) -> None:
         recovery_grace = env_grace_checks()
         recovery_check_every = env_check_every_heartbeats()
         recovery_supervised = supervised_recovery_enabled()
+        # Empty means startup registration was deferred until the listener
+        # existed. Recover immediately; registered nodes keep normal cadence.
+        wait_before_tick = bool(token)
         while True:
-            await asyncio.sleep(_HEARTBEAT_INTERVAL)
+            if wait_before_tick:
+                await asyncio.sleep(_HEARTBEAT_INTERVAL)
+            wait_before_tick = True
             self._runtime_health.advance_supervisor()
             seq += 1
             try:

src/iicp_client/service.py

@@ -123,6 +123,37 @@ def _env_value(key: str, default: str) -> str:
     return os.environ.get(key, default)
 
 
+def _supervisor_tunnel_environment() -> dict[str, str]:
+    from iicp_client.tunnel import cloudflared_path
+
+    configured_path = os.environ.get("IICP_CLOUDFLARED_PATH")
+    binary = cloudflared_path()
+    if configured_path is not None and binary is None:
+        raise ValueError("IICP_CLOUDFLARED_PATH must be an absolute path to an executable file")
+
+    explicit = os.environ.get("IICP_TUNNEL")
+    normalized: str | None = None
+    if explicit is not None:
+        value = explicit.strip().lower()
+        if value in {"1", "true", "yes"}:
+            normalized = "1"
+        elif value in {"0", "false", "no"}:
+            normalized = "0"
+        else:
+            raise ValueError("IICP_TUNNEL must be one of 1/true/yes or 0/false/no")
+    if normalized == "1" and binary is None:
+        raise ValueError(
+            "IICP_TUNNEL=1 requires cloudflared; set IICP_CLOUDFLARED_PATH to its absolute executable path"
+        )
+
+    result: dict[str, str] = {}
+    if binary is not None:
+        result["IICP_CLOUDFLARED_PATH"] = binary
+    if normalized is not None:
+        result["IICP_TUNNEL"] = normalized
+    return result
+
+
 def detect_platform(requested: str = "auto") -> str:
     if requested != "auto":
         if requested not in {"launchd", "systemd"}:
@@ -146,6 +177,7 @@ def render_launchd(node: str, *, name: str | None = None, executable: str = "iic
         "IICP_TUNNEL_DEAD_POLICY": _env_value("IICP_TUNNEL_DEAD_POLICY", "auto"),
         "IICP_LOG_DIR": str(log_dir),
     }
+    env.update(_supervisor_tunnel_environment())
     env_xml = "\n".join(f"    <key>{escape(k)}</key><string>{escape(v)}</string>" for k, v in env.items())
     content = f"""<?xml version="1.0" encoding="UTF-8"?>
 <!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
@@ -200,6 +232,7 @@ def render_systemd(node: str, *, name: str | None = None, executable: str = "iic
         "IICP_TUNNEL_DEAD_POLICY": _env_value("IICP_TUNNEL_DEAD_POLICY", "auto"),
         "IICP_LOG_DIR": str(log_dir),
     }
+    env.update(_supervisor_tunnel_environment())
     env_lines = "\n".join(f"Environment={k}={shlex.quote(v)}" for k, v in env.items())
     content = f"""[Unit]
 Description=IICP node {node}

src/iicp_client/tunnel.py

@@ -439,8 +439,27 @@ def _wait_until_reachable(
 
 
 def cloudflared_path() -> str | None:
-    """Locate the cloudflared binary, or None (we never auto-install it)."""
-    return shutil.which("cloudflared")
+    """Resolve cloudflared for interactive and supervisor-managed execution.
+
+    An explicit ``IICP_CLOUDFLARED_PATH`` is authoritative and must be an
+    absolute executable file. Invalid explicit configuration fails closed
+    instead of selecting another binary from ``PATH``.
+    """
+
+    configured = os.environ.get("IICP_CLOUDFLARED_PATH")
+    candidate = configured if configured is not None else shutil.which("cloudflared")
+    if not candidate:
+        return None
+    path = Path(candidate)
+    if configured is not None and not path.is_absolute():
+        return None
+    try:
+        resolved = path.resolve(strict=True)
+    except (OSError, RuntimeError):
+        return None
+    if not resolved.is_file() or not os.access(resolved, os.X_OK):
+        return None
+    return str(resolved)
 
 
 class QuickTunnel:

tests/test_sdk_quality_contract.py

@@ -34,3 +34,8 @@ def test_quality_runner_uses_the_shared_content_free_schema() -> None:
 def test_pull_request_quality_enforces_mypy_no_regression() -> None:
     workflow = (ROOT / ".github/workflows/quality.yml").read_text()
     assert "python scripts/check_mypy_baseline.py" in workflow
+
+
+def test_mypy_baseline_uses_the_active_python_environment() -> None:
+    source = (ROOT / "scripts/check_mypy_baseline.py").read_text()
+    assert '[sys.executable, "-m", "mypy", "src"]' in source

tests/test_serve.py

@@ -551,10 +551,20 @@ async def fake_reg():
 
     monkeypatch.setattr(n, "heartbeat", fake_hb)
     monkeypatch.setattr(n, "register", fake_reg)
+    recovered_heartbeat = asyncio.Event()
+
+    original_fake_hb = fake_hb
+
+    async def observed_hb(tok):
+        await original_fake_hb(tok)
+        if tok == "recovered-token":
+            recovered_heartbeat.set()
+
+    monkeypatch.setattr(n, "heartbeat", observed_hb)
 
     async def _run():
         task = asyncio.create_task(n._heartbeat_loop(""))  # started with empty token
-        await asyncio.sleep(0.06)
+        await asyncio.wait_for(recovered_heartbeat.wait(), timeout=1.0)
         task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
             await task

tests/test_service.py

@@ -77,3 +77,39 @@ def test_no_start_omits_start(monkeypatch, tmp_path):
     unit = render_systemd("mynode")
     commands = [a.argv for a in manager_actions(unit, "install", no_start=True)]
     assert not any(command[:3] == ("systemctl", "--user", "start") for command in commands)
+
+
+def test_service_preserves_only_explicit_tunnel_policy_and_resolved_binary(monkeypatch, tmp_path):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    binary = tmp_path / "cloudflared"
+    binary.write_text("#!/bin/sh\nexit 0\n")
+    binary.chmod(0o700)
+    monkeypatch.setenv("IICP_CLOUDFLARED_PATH", str(binary))
+    monkeypatch.setenv("IICP_TUNNEL", "yes")
+
+    launchd = render_launchd("mynode")
+    systemd = render_systemd("mynode")
+    resolved = str(binary.resolve())
+    assert f"<key>IICP_CLOUDFLARED_PATH</key><string>{resolved}</string>" in launchd.content
+    assert "<key>IICP_TUNNEL</key><string>1</string>" in launchd.content
+    assert f"Environment=IICP_CLOUDFLARED_PATH={resolved}" in systemd.content
+    assert "Environment=IICP_TUNNEL=1" in systemd.content
+
+    monkeypatch.delenv("IICP_TUNNEL")
+    automatic = render_launchd("mynode")
+    assert "<key>IICP_TUNNEL</key>" not in automatic.content
+
+
+def test_service_refuses_invalid_or_unavailable_forced_tunnel(monkeypatch, tmp_path):
+    import pytest
+
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("IICP_CLOUDFLARED_PATH", "relative/cloudflared")
+    with pytest.raises(ValueError, match="absolute path"):
+        render_launchd("mynode")
+
+    monkeypatch.delenv("IICP_CLOUDFLARED_PATH")
+    monkeypatch.setenv("PATH", "")
+    monkeypatch.setenv("IICP_TUNNEL", "1")
+    with pytest.raises(ValueError, match="requires cloudflared"):
+        render_systemd("mynode")

tests/test_tunnel.py

@@ -52,6 +52,22 @@
 """
 
 
+def test_cloudflared_override_is_absolute_executable_and_authoritative(monkeypatch, tmp_path):
+    binary = tmp_path / "cloudflared"
+    binary.write_text("#!/bin/sh\nexit 0\n")
+    binary.chmod(0o700)
+
+    monkeypatch.setenv("IICP_CLOUDFLARED_PATH", str(binary))
+    assert cloudflared_path() == str(binary.resolve())
+
+    monkeypatch.setenv("IICP_CLOUDFLARED_PATH", "relative/cloudflared")
+    monkeypatch.setenv("PATH", str(tmp_path))
+    assert cloudflared_path() is None
+
+    monkeypatch.delenv("IICP_CLOUDFLARED_PATH")
+    assert cloudflared_path() == str(binary.resolve())
+
+
 def _fake_bin(tmp_path, template: str, name: str = "fake-fox-1234", lifetime: float = 60.0) -> str:
     p = tmp_path / "cloudflared"
     p.write_text(template.format(python=sys.executable, name=name, lifetime=lifetime))