fix(node): a Windows daemon that stops properly, starts honestly and runs once

Found by installing the builds and driving every startup mode live:

- Stop through the node's own control API first (POST /api/shutdown, loopback
  and per-run token): the one channel that reaches a daemon in any session
  without elevation -- a service node runs in session 0 -- and the one that
  runs its shutdown. Then Task Scheduler, then a forced stop. Nine stops in a
  row used to log no shutdown at all: each was a TerminateProcess.
- The forced stop spares the command running it. The frozen meshbay-node.exe
  is the daemon and every CLI verb, so `taskkill /IM meshbay-node.exe` killed
  `autostart stop` and `restart-daemon` themselves: exit 1, no output, and no
  node after a restart. It excludes its own pid and its parent's, and /T takes
  a venv launcher's python child and a daemon's ffmpeg children with it.
- Start and restart report the version that answered, never "started" about a
  node nobody asked; `service start` says so when no node answered, and where
  the log is.
- A second instance fails before it touches anything. The daemon wrote
  ui-token, then failed to bind inside uvicorn's task and exited with the
  reason on a hidden console; the node still running then refused every stop
  and status, its token file naming a dead process. The control port is now
  bound first (exclusively on Windows, where SO_REUSEADDR would share it), and
  a refusal is logged and exits 2. Linux had the same order.
- The daemon logs to %LOCALAPPDATA%\meshbay\state\node.log: Task Scheduler
  discards its stderr. Only the daemon run opens it, never a CLI verb.
- Hub sign-in waits are interruptible, a stop requested before the node is up
  is honoured, and a hub that answers 429 or restarts leaves the node in
  waiting_for_hub rather than looking dead.
- operator_paired is null until the roster is read, instead of a false that
  showed "No operator paired" about a node whose pairing was intact.

The node test conftest also points HOME, USERPROFILE, LOCALAPPDATA and APPDATA
at a throwaway directory for every test, and keeps log_file() away from the
developer's own node: redirecting HOME alone isolates nothing on Windows, and
the CLI tests had been writing invite and pairing codes into the real profile.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

4 files changed, 556 insertions, 90 deletions

diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py
index 9ef23a3..32c9b3d 100644
--- a/packages/meshbay-node/tests/conftest.py
+++ b/packages/meshbay-node/tests/conftest.py
@@ -18,6 +18,48 @@ needs_subprocess = pytest.mark.skipif(
 )
 
 @pytest.fixture(autouse=True)
+def _no_log_file_in_the_developers_profile(monkeypatch):
+    """A test that runs the daemon's entry point must not open its log file.
+
+    On Windows that file is %LOCALAPPDATA%\\meshbay\\state\\node.log -- the
+    developer's own node's log. The first run of the suite with it appended
+    every later test's log lines there, through a handler left on the root
+    logger. A test about the log file points log_file() somewhere of its own.
+    """
+    import logging
+
+    from meshbay_node import platform as _plat
+    monkeypatch.setattr(_plat, "log_file", lambda: None)
+    root = logging.getLogger()
+    before = list(root.handlers)
+    yield
+    for h in root.handlers[:]:
+        if h not in before and getattr(h, "baseFilename", None):
+            root.removeHandler(h)
+            h.close()
+
+
+@pytest.fixture(scope="session")
+def _throwaway_profile(tmp_path_factory):
+    return tmp_path_factory.mktemp("profile")
+
+
+@pytest.fixture(autouse=True)
+def _no_test_touches_the_developers_profile(monkeypatch, _throwaway_profile):
+    """Every per-user location, pointed somewhere nobody lives.
+
+    Tests redirected HOME and nothing else, which isolates nothing on Windows:
+    Path.home() reads USERPROFILE there and the node's own directories come from
+    LOCALAPPDATA. `member invite` and `operator pair`, walked by the CLI tests,
+    wrote their codes into the developer's real node directory and a
+    ~/.local/share/meshbay nobody had, on every run. A test that sets its own
+    HOME still can; it just no longer falls through to the real one.
+    """
+    for var in ("HOME", "USERPROFILE", "LOCALAPPDATA", "APPDATA"):
+        monkeypatch.setenv(var, str(_throwaway_profile / var.lower()))
+
+
+@pytest.fixture(autouse=True)
 def _restore_media_tool_paths():
     """Put `platform`'s resolved ffmpeg/ffprobe paths back after every test.
 
diff --git a/packages/meshbay-node/tests/test_login_retry_is_resilient.py b/packages/meshbay-node/tests/test_login_retry_is_resilient.py
index e37a413..39358bb 100644
--- a/packages/meshbay-node/tests/test_login_retry_is_resilient.py
+++ b/packages/meshbay-node/tests/test_login_retry_is_resilient.py
@@ -14,8 +14,7 @@ import asyncio
 
 import httpx
 import pytest
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-from meshbay_node.config import Config, GroupConfig, HubConfig, KeystoreConfig, NodeConfig
+from meshbay_node.config import Config, HubConfig, KeystoreConfig, NodeConfig
 from meshbay_node.daemon import NodeDaemon
 
 
@@ -96,6 +95,23 @@ async def test_a_401_still_retries_and_stays_alive(tmp_path, monkeypatch):
     assert daemon._state.get("status") in ("waiting_for_node_key", "waiting_for_account")
 
 
+def test_a_node_waiting_for_its_link_stays_under_the_hubs_sign_in_limit():
+    """At 5s a node waiting for its key made 12 attempts a minute against a
+    limit of 10: it put itself in 429 back-off within a minute, every minute,
+    and the desktop app did not recognise that state as a node at all."""
+    import re
+    from pathlib import Path
+
+    from meshbay_node.daemon import NodeDaemon
+
+    nodes_py = (Path(__file__).resolve().parents[2] / "meshbay-hub" / "src" / "meshbay_hub"
+                / "api" / "nodes.py").read_text(encoding="utf-8")
+    m = re.search(r'@router\.post\("/auth"\)\s*@limiter\.limit\("(\d+)/minute"\)', nodes_py)
+    assert m, "the hub's node sign-in limit moved; update this test"
+    per_minute = int(m.group(1))
+    assert 60 / NodeDaemon.LINK_WAIT_S < per_minute
+
+
 @pytest.mark.asyncio
 async def test_a_genuine_client_error_still_raises(tmp_path, monkeypatch):
     """A 400/422 is a bug, not a transient state — it must not be swallowed."""
diff --git a/packages/meshbay-node/tests/test_platform.py b/packages/meshbay-node/tests/test_platform.py
index bebe8d9..8856962 100644
--- a/packages/meshbay-node/tests/test_platform.py
+++ b/packages/meshbay-node/tests/test_platform.py
@@ -8,6 +8,7 @@ suite happens to run on.
 import asyncio
 import os
 import sys
+import time
 from pathlib import Path
 from unittest.mock import Mock
 
@@ -168,10 +169,10 @@ def test_autostart_install_refuses_off_windows(monkeypatch):
         plat.autostart_install(exe="/usr/bin/meshbay-node")
 
 
-def test_autostart_run_launches_the_resolved_exe_windowless_and_records_its_pid(
-        win_startup, monkeypatch, tmp_path):
+def test_autostart_run_launches_windowless_and_inherits_nothing(win_startup, monkeypatch):
+    """close_fds: the desktop app starts its node through here because a child
+    of Electron inherited Electron's sockets and held them after the app quit."""
     monkeypatch.setattr(plat, "_node_exe", lambda: r"C:\x\meshbay-node.exe")
-    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))     # state_dir() -> pidfile location
     calls = {}
 
     def fake_popen(argv, **kw):
@@ -181,13 +182,23 @@ def test_autostart_run_launches_the_resolved_exe_windowless_and_records_its_pid(
     monkeypatch.setattr(plat.subprocess, "Popen", fake_popen)
     plat.autostart_run()
     assert calls["argv"] == [r"C:\x\meshbay-node.exe"]
-    flags = calls["kw"]["creationflags"]
-    assert flags & 0x08000000      # CREATE_NO_WINDOW
-    assert flags & 0x00000200      # CREATE_NEW_PROCESS_GROUP
-    assert not flags & 0x00000008  # not DETACHED_PROCESS -- that has no
-                                   # console at all, so CTRL_BREAK_EVENT
-                                   # would have nothing to signal
-    assert plat._pid_file().read_text(encoding="utf-8") == "4242"
+    assert calls["kw"]["creationflags"] & 0x08000000      # CREATE_NO_WINDOW
+    assert not calls["kw"]["creationflags"] & 0x00000008  # a console of its own:
+    # CTRL_LOGOFF/SHUTDOWN reach install_console_close_handler
+    assert calls["kw"]["close_fds"] is True
+
+
+def test_the_launcher_is_this_very_executable_when_frozen(monkeypatch, tmp_path):
+    """PATH came first, and with two copies on it `autostart start` from one
+    install launched the other -- found on a machine with an MSIX build on PATH."""
+    stray = tmp_path / "stray"
+    stray.mkdir()
+    (stray / "meshbay-node.exe").write_text("", encoding="utf-8")
+    monkeypatch.setenv("PATH", str(stray))
+    monkeypatch.setattr(plat.sys, "frozen", True, raising=False)
+    frozen = r"C:\Programs\MeshBay\node-runtime\meshbay-node.exe"
+    monkeypatch.setattr(plat.sys, "executable", frozen)
+    assert plat._node_exe() == frozen
 
 
 def test_autostart_run_refuses_off_windows(monkeypatch):
@@ -196,100 +207,165 @@ def test_autostart_run_refuses_off_windows(monkeypatch):
         plat.autostart_run()
 
 
-# ── Graceful stop (CTRL_BREAK_EVENT + taskkill fallback) ────────────────────
+# ── Stopping ────────────────────────────────────────────────────────────────
 #
-# autostart_end() references signal.CTRL_BREAK_EVENT, which genuinely does not
-# exist in the `signal` module off Windows -- monkeypatching sys.platform
-# cannot manufacture it, unlike the pure-Python behaviour tested above. Skip
-# rather than mock around it, matching test_configure_event_loop_selector_opt_in.
+# The CTRL_BREAK_EVENT path these tests used to pin was the bug: aimed at a
+# node with a console of its own, GenerateConsoleCtrlEvent reached every process
+# on the caller's console, so `autostart stop` and `restart-daemon` killed
+# themselves. The tests mocked os.kill, so they agreed with it by construction.
 
-@pytest.mark.skipif(sys.platform != "win32",
-                    reason="signal.CTRL_BREAK_EVENT exists only on win32")
-def test_autostart_end_stops_gracefully_when_ctrl_break_is_enough(monkeypatch, tmp_path):
+def test_autostart_end_force_stops_by_image_name(monkeypatch):
     monkeypatch.setattr(sys, "platform", "win32")
-    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
-    plat._pid_file().parent.mkdir(parents=True, exist_ok=True)
-    plat._pid_file().write_text("4242", encoding="utf-8")
-
-    kill_calls = []
-    monkeypatch.setattr(plat.os, "kill", lambda pid, sig: kill_calls.append((pid, sig)))
-    # Alive (our exe) on the pre-signal check, gone by the first poll after --
-    # a plain constant can't tell those two calls apart.
-    seen = {"n": 0}
-
-    def fake_check(pid):
-        seen["n"] += 1
-        return seen["n"] == 1
-
-    monkeypatch.setattr(plat, "_pid_is_meshbay_node", fake_check)
     run_calls = []
     monkeypatch.setattr(plat.subprocess, "run",
                         lambda argv, **kw: run_calls.append(argv))
-
     plat.autostart_end()
-
-    assert kill_calls == [(4242, plat.signal.CTRL_BREAK_EVENT)]
-    assert run_calls == []                       # no taskkill needed
-    assert not plat._pid_file().exists()
+    [argv] = run_calls
+    assert argv[:5] == ["taskkill", "/F", "/T", "/IM", "meshbay-node.exe"]
 
 
-@pytest.mark.skipif(sys.platform != "win32",
-                    reason="signal.CTRL_BREAK_EVENT exists only on win32")
-def test_autostart_end_falls_back_to_taskkill_when_the_pid_never_exits(
-        monkeypatch, tmp_path):
+def test_the_forced_stop_spares_the_command_that_runs_it(monkeypatch):
+    """The CLI is meshbay-node.exe too: `taskkill /IM meshbay-node.exe` killed
+    `autostart stop` and `restart-daemon` themselves -- found on a real install."""
     monkeypatch.setattr(sys, "platform", "win32")
-    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
-    plat._pid_file().parent.mkdir(parents=True, exist_ok=True)
-    plat._pid_file().write_text("4242", encoding="utf-8")
-
-    monkeypatch.setattr(plat.os, "kill", lambda pid, sig: None)
-    monkeypatch.setattr(plat, "_pid_is_meshbay_node", lambda pid: True)  # never exits
-    monkeypatch.setattr(plat.time, "sleep", lambda s: None)       # don't really wait
-    clock = iter([0.0, 1.0, 6.0])       # deadline = 0.0 + 5.0; third read is past it
-    monkeypatch.setattr(plat.time, "monotonic", lambda: next(clock))
     run_calls = []
     monkeypatch.setattr(plat.subprocess, "run",
                         lambda argv, **kw: run_calls.append(argv))
-
     plat.autostart_end()
+    [argv] = run_calls
+    assert f"PID ne {os.getpid()}" in argv and f"PID ne {os.getppid()}" in argv
 
-    assert run_calls == [["taskkill", "/IM", "meshbay-node.exe", "/F"]]
-    assert not plat._pid_file().exists()
 
+@pytest.mark.skipif(sys.platform != "win32", reason="taskkill")
+def test_the_forced_stop_really_spares_its_caller(tmp_path):
+    """For real: a process with the node's image name runs the forced stop, and
+    must survive it while another process of that name does not. Under a name of
+    its own, or it would also kill the developer's running node."""
+    import shutil
+    import subprocess
+    import textwrap
+    import uuid
 
-def test_autostart_end_falls_back_to_taskkill_without_a_pidfile(monkeypatch, tmp_path):
-    """No CTRL_BREAK_EVENT dependency here -- there is no pid to signal, so
-    this one runs everywhere, same as the pre-existing behaviour it replaces."""
-    monkeypatch.setattr(sys, "platform", "win32")
-    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
-    run_calls = []
-    monkeypatch.setattr(plat.subprocess, "run",
-                        lambda argv, **kw: run_calls.append(argv))
-    plat.autostart_end()
-    assert run_calls == [["taskkill", "/IM", "meshbay-node.exe", "/F"]]
+    image = f"mb-test-{uuid.uuid4().hex[:8]}.exe"
+    base = Path(getattr(sys, "_base_executable", sys.executable))
+    env = {**os.environ, "PYTHONHOME": sys.base_prefix}
 
+    def renamed_python(where: str) -> Path:
+        d = tmp_path / where
+        d.mkdir()
+        shutil.copy(base, d / image)
+        for dll in base.parent.glob("*.dll"):
+            shutil.copy(dll, d)
+        return d / image
 
-def test_autostart_end_ignores_a_stale_pid_reused_by_another_process(monkeypatch, tmp_path):
-    """The recorded pid is alive but is not meshbay-node.exe -- Windows reused
-    it after the daemon exited. Must not send CTRL_BREAK_EVENT to whatever
-    that is; falls straight to taskkill (by image name, so harmless here)."""
-    monkeypatch.setattr(sys, "platform", "win32")
-    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
-    plat._pid_file().parent.mkdir(parents=True, exist_ok=True)
-    plat._pid_file().write_text("4242", encoding="utf-8")
+    caller = renamed_python("caller")
+    victim = renamed_python("victim")               # a "daemon" with a "transcode"
+    child_pid_file = tmp_path / "child.pid"
+    src = Path(plat.__file__).resolve().parents[1]
+    script = textwrap.dedent(f"""
+        import sys; sys.path.insert(0, {str(src)!r})
+        from meshbay_node import platform as p
+        p.NODE_IMAGE = {image!r}
+        p.autostart_end()
+        print("survived")
+    """)
+    other = subprocess.Popen(
+        [str(victim), "-c",
+         "import subprocess, sys; p = subprocess.Popen(['ping', '-n', '300', '127.0.0.1'],"
+         " stdout=subprocess.DEVNULL); open(sys.argv[1], 'w').write(str(p.pid)); p.wait()",
+         str(child_pid_file)], env=env)
+    child_pid = 0
+    try:
+        for _ in range(100):
+            if child_pid_file.exists() and child_pid_file.read_text(encoding="utf-8"):
+                break
+            time.sleep(0.1)
+        child_pid = int(child_pid_file.read_text(encoding="utf-8"))
+        r = subprocess.run([str(caller), "-c", script], capture_output=True, text=True,
+                           timeout=60, env=env)
+        assert "survived" in r.stdout, (r.returncode, r.stdout, r.stderr)
+        assert other.wait(timeout=10) is not None, "the other process must be stopped"
+        assert not plat._pid_alive(child_pid), "and its children with it"
+    finally:
+        if other.poll() is None:
+            other.kill()
+        if child_pid:
+            subprocess.run(["taskkill", "/F", "/PID", str(child_pid)], capture_output=True)
 
-    monkeypatch.setattr(plat, "_pid_is_meshbay_node", lambda pid: False)
-    kill_calls = []
-    monkeypatch.setattr(plat.os, "kill", lambda pid, sig: kill_calls.append((pid, sig)))
-    run_calls = []
-    monkeypatch.setattr(plat.subprocess, "run",
-                        lambda argv, **kw: run_calls.append(argv))
 
-    plat.autostart_end()
+def test_nothing_signals_a_console_group_any_more():
+    """Read as code, not text: the docstrings say why CTRL_BREAK went."""
+    import ast
+    import inspect
+    tree = ast.parse(inspect.getsource(plat))
+    for node in ast.walk(tree):
+        assert not (isinstance(node, ast.Attribute) and node.attr == "CTRL_BREAK_EVENT")
+        if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
+            assert not (node.func.attr == "kill" and getattr(node.func.value, "id", "") == "os")
+
+
+def _serve_node(tmp_path, pid):
+    """A control API that answers status and shutdown, as the daemon's does."""
+    import json
+    import threading
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+
+    asked = []
+
+    class H(BaseHTTPRequestHandler):
+        def _send(self, body):
+            data = json.dumps(body).encode()
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(data)))
+            self.end_headers()
+            self.wfile.write(data)
+
+        def do_GET(self):  # noqa: N802
+            self._send({"version": "x", "status": "running", "pid": pid})
+
+        def do_POST(self):  # noqa: N802
+            asked.append(self.path)
+            self._send({"stopping": True})
+
+        def log_message(self, *a):
+            pass
+
+    server = HTTPServer(("127.0.0.1", 0), H)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    (tmp_path / "ui-token").write_text("tok", encoding="utf-8")
+    return server, asked
+
+
+def test_a_graceful_stop_asks_the_node_and_waits_for_its_process(monkeypatch, tmp_path):
+    server, asked = _serve_node(tmp_path, pid=4242)
+    alive = iter([True, True, False])
+    monkeypatch.setattr(plat, "_pid_alive", lambda pid: next(alive))
+    try:
+        assert plat.request_graceful_stop(tmp_path, server.server_address[1], timeout=10)
+    finally:
+        server.shutdown()
+    assert asked == ["/api/shutdown?t=tok"]
+
+
+def test_a_node_that_does_not_exit_is_reported_for_the_caller_to_force(monkeypatch, tmp_path):
+    server, _ = _serve_node(tmp_path, pid=4242)
+    monkeypatch.setattr(plat, "_pid_alive", lambda pid: True)
+    try:
+        assert not plat.request_graceful_stop(tmp_path, server.server_address[1], timeout=1)
+    finally:
+        server.shutdown()
+
+
+def test_nothing_to_stop_when_no_node_answers(tmp_path):
+    (tmp_path / "ui-token").write_text("tok", encoding="utf-8")
+    assert not plat.request_graceful_stop(tmp_path, 1, timeout=1)
+
 
-    assert kill_calls == []                      # never signalled the reused pid
-    assert run_calls == [["taskkill", "/IM", "meshbay-node.exe", "/F"]]
-    assert not plat._pid_file().exists()
+@pytest.mark.skipif(sys.platform != "win32", reason="tasklist")
+def test_pid_alive_sees_this_process_and_not_a_dead_one():
+    import os
+    assert plat._pid_alive(os.getpid())
+    assert not plat._pid_alive(9_999_991)
 
 
 def test_autostart_end_is_a_noop_off_windows(monkeypatch):
@@ -413,7 +489,7 @@ def test_frozen_build_finds_default_env_beside_the_executable(monkeypatch, tmp_p
     """Where build-node-runtime.ps1 puts it, alongside ffmpeg."""
     exe = tmp_path / "meshbay-node.exe"
     exe.write_bytes(b"")
-    (tmp_path / "default.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJtest\n")
+    (tmp_path / "default.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJtest\n", encoding="utf-8")
     monkeypatch.setattr(sys, "frozen", True, raising=False)
     monkeypatch.setattr(sys, "executable", str(exe))
     assert plat.packaged_default_env() == tmp_path / "default.env"
@@ -434,7 +510,7 @@ def test_load_node_env_sets_names(monkeypatch, tmp_path):
         "\n"
         "MESHBAY_TMDB_DEFAULT_TOKEN=eyJloaded\n"
         'QUOTED="value"\n'
-    )
+    , encoding="utf-8")
     monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
     monkeypatch.delenv("QUOTED", raising=False)
     assert plat.load_node_env(tmp_path) == 2
@@ -445,7 +521,7 @@ def test_load_node_env_sets_names(monkeypatch, tmp_path):
 def test_load_node_env_does_not_override_the_environment(monkeypatch, tmp_path):
     """systemd may have loaded the same file already, and an operator export
     must win over a packaged default."""
-    (tmp_path / "node.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJfromfile\n")
+    (tmp_path / "node.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJfromfile\n", encoding="utf-8")
     monkeypatch.setenv("MESHBAY_TMDB_DEFAULT_TOKEN", "eyJfromenv")
     assert plat.load_node_env(tmp_path) == 0
     assert os.environ["MESHBAY_TMDB_DEFAULT_TOKEN"] == "eyJfromenv"
@@ -460,7 +536,7 @@ def test_load_node_env_reads_the_packaged_default_without_a_node_env(monkeypatch
     """A node onboarded by the desktop client has no node.env: nothing ran
     `init` to copy one. The packaged token must reach it anyway."""
     src = tmp_path / "default.env"
-    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJpackaged\n")
+    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJpackaged\n", encoding="utf-8")
     monkeypatch.setattr(plat, "packaged_default_env", lambda: src)
     monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
     cfg = tmp_path / "config"
@@ -472,12 +548,12 @@ def test_load_node_env_reads_the_packaged_default_without_a_node_env(monkeypatch
 
 def test_load_node_env_prefers_the_operator_node_env(monkeypatch, tmp_path):
     src = tmp_path / "default.env"
-    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJpackaged\n")
+    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJpackaged\n", encoding="utf-8")
     monkeypatch.setattr(plat, "packaged_default_env", lambda: src)
     monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
     cfg = tmp_path / "config"
     cfg.mkdir()
-    (cfg / "node.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJoperator\n")
+    (cfg / "node.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJoperator\n", encoding="utf-8")
 
     plat.load_node_env(cfg)
     assert os.environ["MESHBAY_TMDB_DEFAULT_TOKEN"] == "eyJoperator"
diff --git a/packages/meshbay-node/tests/test_windows_service_diagnosis.py b/packages/meshbay-node/tests/test_windows_service_diagnosis.py
new file mode 100644
index 0000000..84ba0b6
--- /dev/null
+++ b/packages/meshbay-node/tests/test_windows_service_diagnosis.py
@@ -0,0 +1,332 @@
+"""
+What a Windows node says about itself when something is wrong.
+
+A 0.16 upgrade in service mode left the previous version's node running, and
+every message on the way out pointed elsewhere: `meshbay-node service start`
+printed "started" for a node that had not, the app said "the node started but
+could not link to your hub account" about one that had stopped answering, the
+Node page said "No operator paired" about one whose pairing was intact, and the
+daemon itself had logged nowhere at all -- Task Scheduler discards its stderr.
+"""
+
+import argparse
+import json
+import logging
+import os
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import pytest
+from meshbay_node import __version__
+from meshbay_node import platform as plat
+from meshbay_node.cli import dispatch, lifecycle
+
+# ── the daemon's log file ──────────────────────────────────────────────────
+
+
+@pytest.fixture
+def root_logger():
+    """add_log_file changes the root logger: pin it and put it back."""
+    root = logging.getLogger()
+    handlers, level, disabled = list(root.handlers), root.level, root.disabled
+    root.setLevel(logging.INFO)
+    root.disabled = False
+    yield root
+    for h in root.handlers:
+        if h not in handlers:
+            h.close()
+    root.handlers[:] = handlers
+    root.setLevel(level)
+    root.disabled = disabled
+
+
+def test_the_daemon_logs_to_a_file_where_there_is_no_console(tmp_path, monkeypatch, root_logger):
+    log_path = tmp_path / "state" / "node.log"
+    monkeypatch.setattr(plat, "log_file", lambda: log_path)
+
+    dispatch.add_log_file()
+    logging.getLogger("meshbay_node.daemon").warning("hub login failed: probe")
+    for h in root_logger.handlers:
+        h.flush()
+
+    assert log_path.exists(), "the log file's directory must be created, not assumed"
+    assert "hub login failed: probe" in log_path.read_text(encoding="utf-8")
+
+
+def test_no_log_file_where_journald_has_the_output(monkeypatch, root_logger):
+    monkeypatch.setattr(plat, "log_file", lambda: None)
+    before = list(root_logger.handlers)
+    dispatch.add_log_file()
+    assert root_logger.handlers == before
+
+
+def test_the_log_file_is_under_localappdata_on_windows(monkeypatch, tmp_path):
+    monkeypatch.undo()   # the conftest keeps log_file() away from the real profile
+    monkeypatch.setattr(plat.sys, "platform", "win32")
+    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
+    assert plat.log_file() == tmp_path / "meshbay" / "state" / "node.log"
+    monkeypatch.setattr(plat.sys, "platform", "linux")
+    assert plat.log_file() is None
+
+
+def test_only_the_daemon_run_opens_the_log_file(monkeypatch):
+    """CLI verbs print reports; appending them to the daemon's log would bury
+    the daemon's own lines."""
+    calls = []
+    monkeypatch.setattr(dispatch, "add_log_file", lambda: calls.append(1))
+    monkeypatch.setattr(plat, "load_node_env", lambda _d: 0)
+    for argv, expected in ((["meshbay-node", "status"], 0), (["meshbay-node"], 1)):
+        calls.clear()
+        monkeypatch.setattr("sys.argv", argv)
+        dispatch.start()
+        assert len(calls) == expected, argv
+
+
+# ── `service start` says what happened ──────────────────────────────────────
+
+
+class _Status(BaseHTTPRequestHandler):
+    body = {}
+
+    def do_GET(self):  # noqa: N802
+        data = json.dumps(self.body).encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def log_message(self, *_a):
+        pass
+
+
+@pytest.fixture
+def fake_daemon(tmp_path):
+    server = HTTPServer(("127.0.0.1", 0), _Status)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    cfg = argparse.Namespace(
+        data_dir=data_dir,
+        node=argparse.Namespace(ui_port=server.server_address[1]))
+    yield cfg
+    server.shutdown()
+
+
+def test_a_token_from_the_stopped_instance_is_not_taken_for_the_new_one(fake_daemon):
+    _Status.body = {"version": "0.16.0", "status": "running"}
+    token = fake_daemon.data_dir / "ui-token"
+    token.write_text("t", encoding="utf-8")
+    old = time.time() - 60
+    os.utime(token, (old, old))
+
+    assert lifecycle._await_daemon(fake_daemon, since=time.time() - 1, timeout=1.5) is None
+
+
+def test_a_daemon_that_started_is_seen(fake_daemon):
+    _Status.body = {"version": "0.16.0", "status": "waiting_for_account"}
+    since = time.time() - 1
+    (fake_daemon.data_dir / "ui-token").write_text("t", encoding="utf-8")
+
+    got = lifecycle._await_daemon(fake_daemon, since=since, timeout=5)
+    assert got == _Status.body
+
+
+def _service_start(monkeypatch, awaited, capsys):
+    monkeypatch.setattr(plat, "service_status", lambda: {"installed": True, "state": "Ready"})
+    monkeypatch.setattr(plat, "service_run", lambda: None)
+    monkeypatch.setattr(plat, "service_end", lambda: None)
+    monkeypatch.setattr(plat, "log_file", lambda: "C:/x/meshbay/state/node.log")
+    monkeypatch.setattr(lifecycle, "load_config", lambda _p: object())
+    monkeypatch.setattr(lifecycle, "_running_status", lambda _cfg: None)
+    monkeypatch.setattr(lifecycle, "_await_daemon", lambda _cfg, _since, **_kw: awaited)
+    monkeypatch.setattr(lifecycle.sys, "platform", "win32")
+    lifecycle.service(argparse.Namespace(subcommand="start", config=None))
+    return capsys.readouterr().out
+
+
+def test_service_start_does_not_report_a_node_that_never_answered(monkeypatch, capsys):
+    with pytest.raises(SystemExit) as exc:
+        _service_start(monkeypatch, None, capsys)
+    assert exc.value.code == 1
+    out = capsys.readouterr().out
+    assert not out.startswith("started"), "it must not claim the node started"
+    assert "no node answered" in out
+    assert "node.log" in out, "and must say where to look"
+
+
+def test_service_start_names_the_version_that_answered(monkeypatch, capsys):
+    out = _service_start(monkeypatch, {"version": __version__, "status": "running"}, capsys)
+    assert f"node {__version__}" in out and "warning" not in out
+
+
+def test_service_start_warns_when_the_service_runs_an_older_node(monkeypatch, capsys):
+    out = _service_start(monkeypatch, {"version": "0.15.0", "status": "waiting_for_account"},
+                         capsys)
+    assert "0.15.0" in out and "warning" in out and "not replaced" in out
+
+
+# ── a graceful stop, through the control API ───────────────────────────────
+
+
+def test_the_control_api_asks_the_daemon_to_stop():
+    from fastapi.testclient import TestClient
+    from meshbay_node.ui import create_ui_app
+
+    asked = []
+    app = create_ui_app({"ui_token": "tok", "request_shutdown": lambda: asked.append(1)})
+    client = TestClient(app)
+    assert client.post("/api/shutdown").status_code in (401, 403), "behind the token"
+    assert asked == []
+    r = client.post("/api/shutdown?t=tok")
+    assert r.status_code == 200 and r.json() == {"stopping": True}
+    assert asked == [1]
+
+
+def test_a_stop_request_ends_the_wait_for_the_hub():
+    """A node waiting for its account is the one people stop and restart."""
+    import asyncio
+
+    from meshbay_node.daemon import NodeDaemon
+
+    async def go():
+        daemon = NodeDaemon.__new__(NodeDaemon)
+        daemon._stop_event = asyncio.Event()
+        asyncio.get_running_loop().call_later(0.1, daemon._stop_event.set)
+        t0 = time.monotonic()
+        stopped = await daemon._pause(30)
+        return stopped, time.monotonic() - t0
+
+    stopped, took = asyncio.run(go())
+    assert stopped and took < 5
+
+
+class _RealNode:
+    """A real daemon process in a throwaway profile, pointed at a hub that is
+    not there (it waits for it, control API up)."""
+
+    def __init__(self, tmp_path):
+        import socket
+
+        with socket.socket() as s:
+            s.bind(("127.0.0.1", 0))
+            self.port = s.getsockname()[1]
+        self.tmp = tmp_path
+        home = tmp_path / "home"
+        self.conf = tmp_path / "node.toml"
+        unlock = tmp_path / "unlock.key"
+        unlock.write_text("x" * 43, encoding="utf-8")
+        self.data = tmp_path / "data"
+        self.conf.write_text(
+            f'data_dir = "{self.data.as_posix()}"\n'
+            '[hub]\nurl = "http://127.0.0.1:1"\nusername = "probe"\n'
+            f'[node]\nquic_enabled = false\nui_port = {self.port}\n'
+            f'[keystore]\npath = "{(tmp_path / "keystore.enc").as_posix()}"\n'
+            f'unlock_file = "{unlock.as_posix()}"\n', encoding="utf-8")
+        self.env = {**os.environ, "HOME": str(home), "LOCALAPPDATA": str(home),
+                    "USERPROFILE": str(home)}
+        self.procs = []
+
+    def spawn(self, name: str):
+        import subprocess
+        import sys
+
+        out = self.tmp / f"{name}.txt"
+        with open(out, "w", encoding="utf-8") as f:
+            proc = subprocess.Popen(
+                [sys.executable, "-c", "from meshbay_node.daemon import main; main()",
+                 "--config", str(self.conf)],
+                env=self.env, stderr=f, stdout=f)
+        proc.out = out
+        self.procs.append(proc)
+        return proc
+
+    def wait_up(self, proc) -> None:
+        deadline = time.monotonic() + 40
+        while not (self.data / "ui-token").exists() and time.monotonic() < deadline:
+            assert proc.poll() is None, proc.out.read_text(encoding="utf-8")
+            time.sleep(0.2)
+        time.sleep(1)
+
+    def kill_all(self) -> None:
+        for p in self.procs:
+            if p.poll() is None:
+                p.kill()
+
+
+def test_a_real_daemon_shuts_down_properly_when_asked(tmp_path):
+    """The whole path, on a real process: before this, nine stops of a Windows
+    node in a row logged not one shutdown -- every one was a TerminateProcess."""
+    node = _RealNode(tmp_path)
+    proc = node.spawn("first")
+    try:
+        node.wait_up(proc)
+        assert plat.request_graceful_stop(node.data, node.port, timeout=15)
+        assert proc.wait(timeout=10) is not None
+        err = proc.out.read_text(encoding="utf-8")
+        assert "Node stopped" in err, err[-2000:]
+    finally:
+        node.kill_all()
+
+
+def test_a_second_node_leaves_the_running_one_alone_and_says_why(tmp_path):
+    """Found on a real install: the sign-in launcher run while a node was up
+    started a second one, which wrote its token over the first's, failed to bind
+    inside uvicorn's task and exited with nothing in the log. The node that kept
+    running then refused every stop, status and restart -- the token file named
+    a process that no longer existed."""
+    node = _RealNode(tmp_path)
+    first = node.spawn("first")
+    try:
+        node.wait_up(first)
+        token = (node.data / "ui-token").read_text(encoding="utf-8")
+
+        second = node.spawn("second")
+        assert second.wait(timeout=40) == 2
+        said = second.out.read_text(encoding="utf-8")
+        assert f"127.0.0.1:{node.port} is taken" in said, said[-2000:]
+        assert "already running" in said
+
+        assert (node.data / "ui-token").read_text(encoding="utf-8") == token
+        assert first.poll() is None
+        assert plat.request_graceful_stop(node.data, node.port, timeout=15), (
+            "the running node can no longer be asked to stop")
+    finally:
+        node.kill_all()
+
+
+def test_the_control_port_is_exclusive():
+    """On Windows SO_REUSEADDR would let a second socket share a port in use."""
+    from meshbay_node.daemon import ControlPortTaken, bind_control_port
+
+    first = bind_control_port(0)
+    try:
+        port = first.getsockname()[1]
+        with pytest.raises(ControlPortTaken, match="already running"):
+            bind_control_port(port)
+    finally:
+        first.close()
+
+
+# ── "No operator paired" only when the roster says so ──────────────────────
+
+
+async def test_a_node_that_has_not_read_its_roster_does_not_claim_there_is_no_operator():
+    from meshbay_node import ops
+    result = await ops.list_groups({"config": None, "groups_ctx": {}})
+    assert result["operator_paired"] is None
+
+
+async def test_a_node_with_a_roster_reports_the_operator(tmp_path):
+    from meshbay_node import ops
+    from meshbay_node.roster import Roster
+
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        result = await ops.list_groups({"config": None, "groups_ctx": {}, "roster": roster})
+    finally:
+        await roster.close()
+    assert result["operator_paired"] is False