runtime: the managed owner runtime on Windows (loopback TCP + token, job object, byte-range leases)

strata runtime start/status/stop/verify/restart now work on Windows. The owner transport (loopback-only TCP, 256-bit
token, state file in a private directory) was already there; the supervisor around it was fcntl/getuid/killpg only.

- strata_control/winproc.py: LockFileEx byte-range leases (shared and exclusive, on a byte beyond the data so the file
  stays readable) and process identity (OpenProcess + GetProcessTimes). Liveness never uses os.kill(pid, 0): on
  Windows that is TerminateProcess.
- runtime.py: state directory %LOCALAPPDATA%\Strata\runtime with a protected one-entry ACL (existing ones must be
  unshared); owner-only checks through private_files.is_unshared; the state file's address must be a loopback port;
  the owner host is detached with CREATE_NEW_PROCESS_GROUP|CREATE_NO_WINDOW; the server runs in a kill-on-close job
  object (serve/winjob.py, terminate() added), so the whole tree ends with the host however the host ends; a graceful
  stop closes the server's stdin pipe (--owner-fd 0, serve/owner_guard.py turns the EOF into the Ctrl+C drain path);
  the deadline kill ends the job; the ownership marker records the child's creation time, so a PID that was reused
  by another process is not mistaken for the recorded child. A PID from a state file is only compared, never signalled.
- A connect failure to a lost owner is a typed ProfileError (exit 3) instead of a raw OSError.
- `strata serve` and calibration take the activity lease on Windows too.
- tests/terminal/test_runtime.py runs on Windows (32 tests were skipped): a .cmd stands in for the fake engine, the
  signal tests have Windows variants; new tests: a second client attaches from another process, wrong/stale tokens
  cannot stop the owner, a recorded PID of an unrelated live process is never touched, a reused PID with another
  creation time is not the recorded child, the liveness check does not end its target, the directory ACL, shared and
  exclusive leases.

serve/owner_guard.py

@@ -1,8 +1,15 @@
-"""Emergency parent-loss guard for a server launched by the local runtime host.
+"""Parent-loss guard and graceful-stop channel for a server launched by the local runtime host.
 
 The inherited pipe has exactly one writer, held by the host.  EOF therefore
-means the host exited, including after SIGKILL.  The server has its own process
-group so its engine subprocesses receive the emergency stop as well.
+means the host exited, including after SIGKILL.
+
+POSIX: the server has its own process group so its engine subprocesses receive
+the emergency stop as well.  Windows: the host passes the pipe as the server's
+standard input (``--owner-fd 0``) and ends the server's whole process tree
+through a kill-on-close job object if it dies; the pipe is also how the host
+asks the server to drain and exit (it closes its end), because Windows has no
+SIGTERM to send.  EOF is turned into a Ctrl+C in the main thread, which is the
+path the server already drains and stops cleanly on.
 """
 
 from __future__ import annotations
@@ -13,7 +20,26 @@ import stat
 import threading
 
 
-def start(fd: int) -> None:
+def _stop_gracefully() -> None:
+    import _thread
+    _thread.interrupt_main()
+
+
+def start(fd: int, on_loss=None) -> None:
+    if os.name == "nt":
+        if fd < 0 or not stat.S_ISFIFO(os.fstat(fd).st_mode):
+            raise ValueError("--owner-fd must be an inherited pipe")
+
+        def watch_windows() -> None:
+            try:
+                while os.read(fd, 1):
+                    pass
+            except OSError:
+                pass
+            (on_loss or _stop_gracefully)()
+
+        threading.Thread(target=watch_windows, name="strata-owner-guard", daemon=True).start()
+        return
     if os.name != "posix" or fd < 0 or os.getpgrp() != os.getpid():
         raise ValueError("--owner-fd requires a dedicated POSIX server process group")
     if not stat.S_ISFIFO(os.fstat(fd).st_mode):

serve/winjob.py

@@ -45,6 +45,7 @@ if os.name == "nt":
     _k32.CreateJobObjectW.argtypes = (wintypes.LPVOID, wintypes.LPCWSTR)
     _k32.SetInformationJobObject.argtypes = (wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID, wintypes.DWORD)
     _k32.AssignProcessToJobObject.argtypes = (wintypes.HANDLE, wintypes.HANDLE)
+    _k32.TerminateJobObject.argtypes = (wintypes.HANDLE, wintypes.UINT)
 
     def _make_job():
         job = _k32.CreateJobObjectW(None, None)     # no security attributes: the handle is not inherited
@@ -80,6 +81,14 @@ else:
         return False
 
 
+def terminate() -> bool:
+    """End every process in the job now (the deadline kill of an owner that must not leave a tree behind). False when
+    there is no job (nothing was contained, or not Windows)."""
+    if os.name != "nt" or not _job:
+        return False
+    return bool(_k32.TerminateJobObject(_job, 1))
+
+
 def contain(proc) -> bool:
     """Put a started subprocess.Popen in the kill-on-close job. True if it is in; False (harmlessly) otherwise."""
     global _job

strata_cli/cli.py

@@ -739,7 +739,7 @@ def _serve(a: argparse.Namespace) -> int:
     if a.plan:
         return _success({"plan": plan}, machine=a.output == "json")
     from serve.server import main as server_main
-    if os.name != "posix":
+    if os.name not in ("posix", "nt"):
         return _serve_foreground(server_main, args, plan, output=a.output, quiet=a.quiet)
     from strata_control import runtime
     try:

strata_control/calibration.py

@@ -167,7 +167,7 @@ def _exclusive_preflight(cfg: dict, profile: dict | None = None) -> None:
 @contextmanager
 def _managed_exclusive():
     """Keep new owner hosts out for the full calibration engine action."""
-    if os.name != "posix":
+    if os.name not in ("posix", "nt"):
         yield
         return
     from strata_control import runtime

strata_control/runtime.py

@@ -1,9 +1,16 @@
 """Local owner-controlled supervisor for one saved Strata serving profile.
 
 The supervisor holds the Popen handle for the server it started. A client never
-sends a signal to a PID found in a ready file. The control socket and token live
-in an owner-private directory. This first implementation supports POSIX; Windows
-needs a qualified owner-only pipe/ACL implementation before it may manage work.
+sends a signal to a PID found in a ready file. The control channel and token live
+in an owner-private directory.
+
+POSIX: an owner-private AF_UNIX socket, flock leases, SIGTERM to the server's own
+process group. Windows: a loopback-only TCP listener authorised by the 256-bit
+token (strata_control/owner_transport.py), byte-range leases (winproc.py), a
+protected one-entry ACL on the state directory, and a kill-on-close job object
+around the server tree. A PID in a state file is never signalled: it is only
+compared (with its creation time on Windows) to decide whether the recorded
+child is still there.
 """
 
 from __future__ import annotations
@@ -30,16 +37,46 @@ try:
 except ImportError:  # Windows has no qualified owner transport yet.
     fcntl = None
 
-from strata_control import owner_transport, profiles, readiness
+from strata_control import owner_transport, private_files, profiles, readiness, winproc
 from serve import launch_identity
 from strata_control.profiles import ProfileError
 
 
 MAX_MESSAGE = 64 * 1024
 ROOT = Path(__file__).resolve().parents[1]
+WINDOWS = os.name == "nt"
 _STARTED_HOSTS: dict[str, subprocess.Popen] = {}
 
 
+def _is_private(details: os.stat_result, path: Path) -> bool:
+    """Owner-only: POSIX owner and no group/other bits; Windows an ACL that shuts out every other account."""
+    if WINDOWS:
+        try:
+            return private_files.is_unshared(path)
+        except OSError:
+            return False
+    return details.st_uid == os.getuid() and not stat.S_IMODE(details.st_mode) & 0o077
+
+
+def _lock_file(descriptor: int, *, exclusive: bool) -> bool:
+    """Take a lease without waiting: True when taken, False when another holder has it."""
+    if WINDOWS:
+        return winproc.try_lock(descriptor, exclusive=exclusive)
+    try:
+        fcntl.flock(descriptor, (fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH) | fcntl.LOCK_NB)
+    except OSError as exc:
+        if exc.errno in (errno.EWOULDBLOCK, errno.EAGAIN):
+            return False
+        raise
+    return True
+
+
+def _unlock_file(descriptor: int) -> None:
+    """Windows releases a lock only some time after the handle closes: release it now. POSIX closes."""
+    if WINDOWS:
+        winproc.unlock(descriptor)
+
+
 def _supported() -> None:
     """Refuse only when no transport exists at all.
 
@@ -63,14 +100,20 @@ def _name(value: str) -> str:
 def runtime_dir() -> Path:
     _supported()
     override = os.environ.get("STRATA_RUNTIME_DIR")
-    base = Path(override).expanduser() if override else \
-        Path(os.environ.get("XDG_STATE_HOME") or Path.home() / ".local" / "state") / "strata" / "runtime"
+    if override:
+        base = Path(override).expanduser()
+    elif WINDOWS:
+        base = Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local") / "Strata" / "runtime"
+    else:
+        base = Path(os.environ.get("XDG_STATE_HOME") or Path.home() / ".local" / "state") / "strata" / "runtime"
     if not base.is_absolute():
         raise ProfileError("STRATA_RUNTIME_DIR must be absolute")
+    existed = base.exists()
     base.mkdir(parents=True, mode=0o700, exist_ok=True)
+    if WINDOWS and not existed:
+        private_files.restrict_to_current_user(base, directory=True)   # no mode bits on NTFS: a protected ACL
     details = base.lstat()
-    if not stat.S_ISDIR(details.st_mode) or details.st_uid != os.getuid() or \
-            stat.S_IMODE(details.st_mode) & 0o077:
+    if not stat.S_ISDIR(details.st_mode) or not _is_private(details, base):
         raise ProfileError("runtime directory must be owner-only (mode 0700)", 5)
     return base
 
@@ -82,15 +125,10 @@ def acquire_activity_lease(*, shared: bool) -> int:
     try:
         descriptor = os.open(directory / ".activity.lock", flags, 0o600)
         details = os.fstat(descriptor)
-        if not stat.S_ISREG(details.st_mode) or details.st_uid != os.getuid() or \
-                stat.S_IMODE(details.st_mode) & 0o077:
+        if not stat.S_ISREG(details.st_mode) or not _is_private(details, directory / ".activity.lock"):
             raise ProfileError("runtime activity lock must be owner-only", 5)
-        try:
-            fcntl.flock(descriptor, (fcntl.LOCK_SH if shared else fcntl.LOCK_EX) | fcntl.LOCK_NB)
-        except OSError as exc:
-            if exc.errno in (errno.EWOULDBLOCK, errno.EAGAIN):
-                raise ProfileError("a managed runtime or calibration owns the local engine lease", 6) from None
-            raise
+        if not _lock_file(descriptor, exclusive=not shared):
+            raise ProfileError("a managed runtime or calibration owns the local engine lease", 6)
         return descriptor
     except BaseException:
         if "descriptor" in locals():
@@ -128,15 +166,18 @@ def _read_state(name: str) -> dict | None:
         details = path.lstat()
     except FileNotFoundError:
         return None
-    if not stat.S_ISREG(details.st_mode) or details.st_uid != os.getuid() or \
-            stat.S_IMODE(details.st_mode) & 0o077 or details.st_size > 16 * 1024:
+    if not stat.S_ISREG(details.st_mode) or not _is_private(details, path) or details.st_size > 16 * 1024:
         raise ProfileError("runtime state file is not private or is invalid", 5)
     try:
         state = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, UnicodeError, ValueError):
         raise ProfileError("runtime state file is unreadable", 5) from None
+    # POSIX: the address is the one derived from the profile. Windows: the port is chosen when the owner binds, so
+    # the state file carries it, and the only address a client may be sent to is a loopback port.
+    address_ok = bool(re.fullmatch(r"tcp:127\.0\.0\.1:[1-9][0-9]{0,4}", str(state.get("socket", "")))) \
+        if WINDOWS and isinstance(state, dict) else (isinstance(state, dict) and state.get("socket") == socket_path)
     if not isinstance(state, dict) or state.get("schema_version") != 1 or state.get("profile") != name or \
-            state.get("socket") != socket_path or not isinstance(state.get("token"), str) or \
+            not address_ok or not isinstance(state.get("token"), str) or \
             not re.fullmatch(r"[0-9a-f]{64}", state["token"]):
         raise ProfileError("runtime state file has an unsupported shape", 5)
     return state
@@ -146,7 +187,12 @@ def _child_marker(name: str, token: str) -> Path:
     return runtime_dir() / f"{name}-{token}.child.json"
 
 
-def _pid_exists(pid: int) -> bool:
+def _pid_exists(pid: int, started: int | None = None) -> bool:
+    """Whether the recorded child is still there. Never signals: on Windows `os.kill(pid, 0)` would END the process,
+    so liveness is a handle query, and `started` (the creation time recorded with the PID) tells the recorded
+    process from a later one that reused the number."""
+    if WINDOWS:
+        return winproc.pid_alive(pid, started)
     try:
         os.kill(pid, 0)  # Liveness check only; never signal a PID from saved state.
     except ProcessLookupError:
@@ -160,12 +206,9 @@ def _host_lease_active(name: str) -> bool:
     path, _, _ = _paths(name)
     descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
     try:
-        try:
-            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except OSError as exc:
-            if exc.errno in (errno.EWOULDBLOCK, errno.EAGAIN):
-                return True
-            raise
+        if not _lock_file(descriptor, exclusive=True):
+            return True
+        _unlock_file(descriptor)
         return False
     finally:
         os.close(descriptor)
@@ -182,17 +225,12 @@ def _recover_stale(name: str, *, wait: float) -> bool:
     try:
         if os.fstat(descriptor).st_ino != state_path.lstat().st_ino:
             raise ProfileError("runtime state changed during recovery", 6)
-        try:
-            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except OSError as exc:
-            if exc.errno in (errno.EWOULDBLOCK, errno.EAGAIN):
-                raise ProfileError("this profile already has a live managed runtime", 6) from None
-            raise
+        if not _lock_file(descriptor, exclusive=True):
+            raise ProfileError("this profile already has a live managed runtime", 6)
         marker = _child_marker(name, state["token"])
         try:
             details = marker.lstat()
-            if not stat.S_ISREG(details.st_mode) or details.st_uid != os.getuid() or \
-                    stat.S_IMODE(details.st_mode) & 0o077 or details.st_size > 4096:
+            if not stat.S_ISREG(details.st_mode) or not _is_private(details, marker) or details.st_size > 4096:
                 raise ValueError("invalid owner marker")
             child = json.loads(marker.read_text(encoding="utf-8"))
         except (OSError, ValueError):
@@ -201,9 +239,10 @@ def _recover_stale(name: str, *, wait: float) -> bool:
                 child.get("phase") != "owned" or type(child.get("pid")) is not int or child["pid"] <= 0:
             raise ProfileError("lost host child launch is unverified; inspect before recovery", 6)
         deadline = time.monotonic() + min(45.0, max(5.0, wait))
-        while _pid_exists(child["pid"]) and time.monotonic() < deadline:
+        started = child.get("started") if type(child.get("started")) is int else None
+        while _pid_exists(child["pid"], started) and time.monotonic() < deadline:
             time.sleep(0.1)
-        if _pid_exists(child["pid"]):
+        if _pid_exists(child["pid"], started):
             raise ProfileError("lost host child is still present; retry after it exits", 6)
         if not isinstance(child.get("ready"), str):
             raise ProfileError("lost host ready path is invalid; inspect before recovery", 6)
@@ -213,8 +252,7 @@ def _recover_stale(name: str, *, wait: float) -> bool:
             raise ProfileError("lost host ready path is invalid; inspect before recovery", 6)
         try:
             details = ready.lstat()
-            if not stat.S_ISREG(details.st_mode) or details.st_uid != os.getuid() or \
-                    stat.S_IMODE(details.st_mode) & 0o077:
+            if not stat.S_ISREG(details.st_mode) or not _is_private(details, ready):
                 raise ProfileError("lost host ready file is invalid; inspect before recovery", 6)
             ready.unlink()
         except FileNotFoundError:
@@ -248,15 +286,19 @@ def _recover_stale(name: str, *, wait: float) -> bool:
             raise ProfileError("too many lost-host config snapshots; inspect before recovery", 6)
         for snapshot in snapshots:
             details = snapshot.lstat()
-            if not stat.S_ISREG(details.st_mode) or details.st_uid != os.getuid() or \
-                    stat.S_IMODE(details.st_mode) & 0o077:
+            if not stat.S_ISREG(details.st_mode) or not _is_private(details, snapshot):
                 raise ProfileError("lost-host config snapshot is not private; inspect before recovery", 6)
             snapshot.unlink()
         marker.unlink()
+        if WINDOWS:                    # an open file cannot be deleted on Windows: release the lease first
+            _unlock_file(descriptor)
+            os.close(descriptor)
+            descriptor = -1
         state_path.unlink()
         return True
     finally:
-        os.close(descriptor)
+        if descriptor >= 0:
+            os.close(descriptor)
 
 
 def _write_new(path: Path, data: bytes) -> int:
@@ -264,7 +306,8 @@ def _write_new(path: Path, data: bytes) -> int:
     try:
         # Keep this lease open in the owner host. The kernel releases it even
         # after SIGKILL, unlike a PID file or a stale Unix socket.
-        fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if not _lock_file(descriptor, exclusive=True):
+            raise OSError(errno.EWOULDBLOCK, "the runtime lease is held")
         with os.fdopen(os.dup(descriptor), "wb") as stream:
             stream.write(data)
             stream.flush()
@@ -286,7 +329,10 @@ def _rpc(name: str, command: str, *, timeout: float = 5, params: dict | None = N
     raw = (json.dumps(request) + "\n").encode("utf-8")
     # The transport decides the address family: a private POSIX socket, or the loopback listener
     # on Windows. The token in `raw` is what authorises the request either way.
-    client = owner_transport.connect(state["socket"], timeout=timeout)
+    try:
+        client = owner_transport.connect(state["socket"], timeout=timeout)
+    except OSError:      # nothing listens there any more (a lost owner), or it did not answer in time
+        raise ProfileError("runtime control is unavailable; inspect the private host log", 3) from None
     try:
         try:
             client.sendall(raw)
@@ -378,9 +424,11 @@ def start(name: str, *, wait: float = 300, shutdown_timeout: float = 30,
                 command += ["--expected-profile-revision", expected_profile_revision]
             if expected_source_sha256 is not None:
                 command += ["--expected-source-sha256", expected_source_sha256]
+            detach = ({"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW}
+                      if WINDOWS else {"start_new_session": True})
             proc = subprocess.Popen(command,
                                     cwd=ROOT, stdin=subprocess.DEVNULL, stdout=stream,
-                                    stderr=subprocess.STDOUT, start_new_session=True, close_fds=True)
+                                    stderr=subprocess.STDOUT, close_fds=True, **detach)
         _STARTED_HOSTS[name] = proc
         deadline = time.monotonic() + max(5.0, wait)
         while _read_state(name) is None and proc.poll() is None and time.monotonic() < deadline:
@@ -518,6 +566,7 @@ class Host:
         self.running = True
         self.state_fd: int | None = None
         self.owner_write_fd: int | None = None
+        self.owner_write = None            # Windows: the server's stdin pipe, whose EOF asks it to drain and stop
         self.probe_timeout = startup_wait
         self.probe_deadline = 0.0
         self.probe_key: tuple[int, str] | None = None
@@ -528,6 +577,10 @@ class Host:
         payload = {"host_token": self.token, "phase": phase}
         if pid is not None and ready is not None:
             payload.update(pid=pid, ready=str(ready))
+            if WINDOWS:
+                started = winproc.process_started(pid)
+                if started is not None:
+                    payload["started"] = started   # tells this process from a later one with the same number
         profiles._atomic(_child_marker(self.name, self.token), (json.dumps(payload) + "\n").encode())
 
     def _plan(self) -> tuple[dict, list[str], Path, bytes]:
@@ -592,32 +645,47 @@ class Host:
         log = self.state_path.parent / f"{self.name}.server.log"
         self._mark_child("launching")
         descriptor = os.open(log, flags, 0o600)
-        reader, writer = os.pipe()
+        # POSIX: an inherited pipe with one writer (this host). Windows: the server's standard input is that pipe.
+        reader, writer = (None, None) if WINDOWS else os.pipe()
         server_args = [*args, "--ready-file", str(ready),
                        "--shutdown-timeout", str(self.shutdown_timeout),
-                       "--owner-fd", str(reader)]
+                       "--owner-fd", "0" if WINDOWS else str(reader)]
         attestation_key = secrets.token_bytes(32)
         expected_attestation = launch_identity.attestation(attestation_key, source_bytes, server_args)
         child_env = os.environ.copy()
         child_env["STRATA_OWNER_ATTEST_KEY"] = attestation_key.hex()
         try:
             with os.fdopen(descriptor, "ab") as stream:
-                self.child = subprocess.Popen([sys.executable, "-m", "serve.server", *server_args],
-                                              cwd=ROOT, stdin=subprocess.DEVNULL, stdout=stream,
-                                              stderr=subprocess.STDOUT, close_fds=True,
-                                              pass_fds=(reader,), start_new_session=True, env=child_env)
+                if WINDOWS:
+                    self.child = subprocess.Popen([sys.executable, "-m", "serve.server", *server_args],
+                                                  cwd=ROOT, stdin=subprocess.PIPE, stdout=stream,
+                                                  stderr=subprocess.STDOUT, close_fds=True, env=child_env,
+                                                  creationflags=subprocess.CREATE_NO_WINDOW)
+                    # the whole server tree ends with this host, however the host ends (a kill-on-close job)
+                    from serve import winjob
+                    winjob.contain(self.child)
+                else:
+                    self.child = subprocess.Popen([sys.executable, "-m", "serve.server", *server_args],
+                                                  cwd=ROOT, stdin=subprocess.DEVNULL, stdout=stream,
+                                                  stderr=subprocess.STDOUT, close_fds=True,
+                                                  pass_fds=(reader,), start_new_session=True, env=child_env)
         except BaseException:
-            os.close(writer)
+            if writer is not None:
+                os.close(writer)
             raise
         finally:
-            os.close(reader)
-        self.owner_write_fd = writer
+            if reader is not None:
+                os.close(reader)
+        if WINDOWS:
+            self.owner_write = self.child.stdin
+        else:
+            self.owner_write_fd = writer
         try:
             self._mark_child("owned", pid=self.child.pid, ready=ready)
         except BaseException:
             # No ownership record means a future start cannot safely reclaim
             # this instance. Stop this child immediately instead.
-            os.killpg(self.child.pid, signal.SIGKILL)
+            self._kill_child_tree()
             self.child.wait(timeout=10)
             self._close_owner_pipe()
             raise
@@ -702,11 +770,14 @@ class Host:
             if self.ready_path is not None:
                 self.ready_path.unlink(missing_ok=True)
             return {"ok": True, "state": "stopped", "exit_code": child.returncode if child else 0}
-        child.send_signal(signal.SIGTERM)
+        if WINDOWS:
+            self._close_owner_pipe()             # EOF on the server's stdin: it drains and stops (no SIGTERM here)
+        else:
+            child.send_signal(signal.SIGTERM)
         try:
             code = child.wait(timeout=self.shutdown_timeout + 8)
         except subprocess.TimeoutExpired:
-            os.killpg(child.pid, signal.SIGKILL)
+            self._kill_child_tree()
             child.wait(timeout=10)
             self._close_owner_pipe()
             return {"ok": False, "state": "failed", "error": "server_stop_deadline_expired", "exit_code": 8}
@@ -720,6 +791,30 @@ class Host:
         if self.owner_write_fd is not None:
             os.close(self.owner_write_fd)
             self.owner_write_fd = None
+        if self.owner_write is not None:
+            try:
+                self.owner_write.close()
+            except OSError:
+                pass
+            self.owner_write = None
+
+    def _kill_child_tree(self) -> None:
+        """The deadline kill of the server this host started, and everything it started. Only our own child: POSIX
+        signals its process group, Windows ends the job the child was put in (taskkill of our own child's tree if
+        the job could not be made)."""
+        child = self.child
+        if child is None:
+            return
+        if WINDOWS:
+            from serve import winjob
+            if not winjob.terminate():
+                subprocess.run(["taskkill", "/T", "/F", "/PID", str(child.pid)], capture_output=True)
+            try:
+                child.kill()
+            except OSError:
+                pass
+        else:
+            os.killpg(child.pid, signal.SIGKILL)
 
     def _restart(self, startup_wait: float, expected_profile_revision: str | None = None,
                  allow_source_switch: bool = False,
@@ -955,6 +1050,8 @@ class Host:
                 self.running = False
             signal.signal(signal.SIGTERM, shutdown)
             signal.signal(signal.SIGINT, shutdown)
+            if hasattr(signal, "SIGBREAK"):      # Windows: the signal a console Ctrl+Break / a closing console sends
+                signal.signal(signal.SIGBREAK, shutdown)
             self.state_fd = _write_new(self.state_path, (json.dumps({"schema_version": 1, "profile": self.name,
                                                                      "socket": bound_address, "token": self.token,
                                                                      "host_pid": os.getpid()}) + "\n").encode())
@@ -1006,6 +1103,11 @@ class Host:
                 self._stop_child()
             self._close_owner_pipe()
             listener.close()
+            if WINDOWS and self.state_fd is not None:
+                # an open file cannot be deleted on Windows: the lease ends just before the state file does
+                _unlock_file(self.state_fd)
+                os.close(self.state_fd)
+                self.state_fd = None
             try:
                 current = _read_state(self.name)
                 if current and current.get("token") == self.token:
@@ -1024,6 +1126,7 @@ class Host:
                 self._retire_snapshot(snapshot)
             if self.state_fd is not None:
                 os.close(self.state_fd)
+            _unlock_file(activity_lease)
             os.close(activity_lease)
         return 0
 

strata_control/winproc.py

@@ -0,0 +1,120 @@
+"""Windows file-lock and process-identity primitives for the managed runtime (ctypes; no pywin32).
+
+POSIX gets these from ``fcntl.flock`` and ``os.kill(pid, 0)``. Neither exists on Windows, and ``os.kill`` there is
+``TerminateProcess``: calling it with signal 0 would END the process, so a liveness check must never be written that way.
+
+* ``try_lock`` / ``unlock``: a byte-range lock (``LockFileEx``) on one byte far beyond the data, so the file stays
+  readable by other processes while a lock is held (a lock that covered the data would make reads fail). It supports
+  shared locks, which ``msvcrt.locking`` does not, and is released by the OS when the process ends, however it ends.
+* ``process_started`` / ``pid_alive``: whether a process exists and when it was created. A PID recorded in a state
+  file is only a hint; the creation time is what tells the process that was recorded from a later one that reused
+  the number.
+"""
+
+from __future__ import annotations
+
+import os
+
+WINDOWS = os.name == "nt"
+
+#: the locked byte. Beyond any data file the runtime writes (they are all far smaller).
+LOCK_OFFSET = 1 << 16
+
+if WINDOWS:
+    import ctypes
+    import msvcrt
+    from ctypes import wintypes
+
+    _k32 = ctypes.WinDLL("kernel32", use_last_error=True)
+
+    class _Overlapped(ctypes.Structure):
+        _fields_ = [("Internal", ctypes.c_void_p), ("InternalHigh", ctypes.c_void_p),
+                    ("Offset", wintypes.DWORD), ("OffsetHigh", wintypes.DWORD), ("hEvent", wintypes.HANDLE)]
+
+    _k32.LockFileEx.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD,
+                                ctypes.POINTER(_Overlapped)]
+    _k32.LockFileEx.restype = wintypes.BOOL
+    _k32.UnlockFileEx.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD,
+                                  ctypes.POINTER(_Overlapped)]
+    _k32.UnlockFileEx.restype = wintypes.BOOL
+    _k32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
+    _k32.OpenProcess.restype = wintypes.HANDLE
+    _k32.CloseHandle.argtypes = [wintypes.HANDLE]
+    _k32.GetExitCodeProcess.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)]
+    _k32.GetExitCodeProcess.restype = wintypes.BOOL
+    _k32.GetProcessTimes.argtypes = [wintypes.HANDLE] + [ctypes.POINTER(wintypes.FILETIME)] * 4
+    _k32.GetProcessTimes.restype = wintypes.BOOL
+
+    _LOCKFILE_FAIL_IMMEDIATELY = 0x1
+    _LOCKFILE_EXCLUSIVE_LOCK = 0x2
+    _ERROR_LOCK_VIOLATION = 33
+    _ERROR_IO_PENDING = 997
+    _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
+    _STILL_ACTIVE = 259
+    _ERROR_ACCESS_DENIED = 5
+
+    def _overlapped() -> _Overlapped:
+        o = _Overlapped()
+        o.Offset = LOCK_OFFSET & 0xFFFFFFFF
+        o.OffsetHigh = LOCK_OFFSET >> 32
+        return o
+
+    def try_lock(fd: int, *, exclusive: bool) -> bool:
+        """Take the lock without waiting. False when another holder has it."""
+        handle = msvcrt.get_osfhandle(fd)
+        flags = _LOCKFILE_FAIL_IMMEDIATELY | (_LOCKFILE_EXCLUSIVE_LOCK if exclusive else 0)
+        if _k32.LockFileEx(handle, flags, 0, 1, 0, ctypes.byref(_overlapped())):
+            return True
+        code = ctypes.get_last_error()
+        if code in (_ERROR_LOCK_VIOLATION, _ERROR_IO_PENDING):
+            return False
+        raise ctypes.WinError(code, "LockFileEx")
+
+    def unlock(fd: int) -> None:
+        _k32.UnlockFileEx(msvcrt.get_osfhandle(fd), 0, 1, 0, ctypes.byref(_overlapped()))
+
+    def process_started(pid: int) -> int | None:
+        """The creation time (100 ns ticks since 1601) of a running process, or None when there is no such
+        process, it has exited, or its times cannot be read."""
+        if type(pid) is not int or pid <= 0:
+            return None
+        handle = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
+        if not handle:
+            return None
+        try:
+            code = wintypes.DWORD()
+            if not _k32.GetExitCodeProcess(handle, ctypes.byref(code)) or code.value != _STILL_ACTIVE:
+                return None
+            created, exited, kernel, user = (wintypes.FILETIME() for _ in range(4))
+            if not _k32.GetProcessTimes(handle, ctypes.byref(created), ctypes.byref(exited),
+                                        ctypes.byref(kernel), ctypes.byref(user)):
+                return None
+            return (created.dwHighDateTime << 32) | created.dwLowDateTime
+        finally:
+            _k32.CloseHandle(handle)
+
+    def pid_alive(pid: int, started: int | None = None) -> bool:
+        """Whether `pid` is a running process, and, when `started` is given, the one created at that time."""
+        if type(pid) is not int or pid <= 0:
+            return False
+        now = process_started(pid)
+        if now is None:
+            # OpenProcess can be refused for a protected process that does exist; treat that as present.
+            handle = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
+            if handle:
+                _k32.CloseHandle(handle)
+                return False                          # opened, but exited or unreadable: not running
+            return ctypes.get_last_error() == _ERROR_ACCESS_DENIED
+        return started is None or now == started
+else:
+    def try_lock(fd: int, *, exclusive: bool) -> bool:
+        raise OSError("Windows only")
+
+    def unlock(fd: int) -> None:
+        raise OSError("Windows only")
+
+    def process_started(pid: int) -> int | None:
+        return None
+
+    def pid_alive(pid: int, started: int | None = None) -> bool:
+        raise OSError("Windows only")

tests/terminal/test_runtime.py

@@ -21,10 +21,13 @@ from unittest import mock
 from strata_cli.cli import main
 from strata_cli import bench
 from strata_cli.client import Endpoint
-from strata_control import calibration, journal, management, profiles, recipes, runtime
+from strata_control import calibration, journal, management, owner_transport, profiles, recipes, runtime
 
 
-@unittest.skipUnless(os.name == "posix", "owner-only Unix socket transport is POSIX-only")
+WINDOWS = os.name == "nt"
+
+
+@unittest.skipUnless(os.name in ("posix", "nt"), "no qualified owner transport on this platform")
 class RuntimeTests(unittest.TestCase):
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory(prefix="strata-runtime-test-")
@@ -56,6 +59,11 @@ class RuntimeTests(unittest.TestCase):
                           "        print('DONE 1 1 1.0 1.0 length 0 0 0', flush=True)\n"
                           "        if slot is not None: print(f'BADM {slot} 0', flush=True)\n", encoding="utf-8")
         engine.chmod(0o700)
+        if WINDOWS:
+            # a script cannot be an engine exe on Windows: a .cmd that runs it stands in for the binary
+            wrapper = self.root / "engine.cmd"
+            wrapper.write_bytes(f'@echo off\r\n"{sys.executable}" "{engine}" %*\r\n'.encode())
+            engine = wrapper
         from tools.strata_tokenizer import BYTE_TO_UNICODE
         tokenizer = self.root / "tokenizer"
         tokenizer.mkdir()
@@ -108,7 +116,7 @@ class RuntimeTests(unittest.TestCase):
         code, invalid = send("true")
         self.assertEqual((code, invalid["ok"]), (2, False))
         self.assertIn("live must be true or false", invalid["error"])
-        engine = self.root / "engine.py"
+        engine = self.root / ("engine.cmd" if WINDOWS else "engine.py")
         engine.unlink()
         code, missing = send(False)
         self.assertEqual((code, missing["ok"], missing["report"]["ok"]), (0, True, False))
@@ -416,6 +424,7 @@ class RuntimeTests(unittest.TestCase):
                 finally:
                     host.wait(timeout=10)
 
+    @unittest.skipIf(WINDOWS, "Windows console stop needs a platform-specific integration runner (no SIGTERM/SIGINT)")
     def test_foreground_jsonl_emits_ready_and_stop_without_human_stdout(self):
         for stop_signal, expected_code, expected_reason, quiet in (
                 (signal.SIGTERM, 0, "sigterm", False),
@@ -676,7 +685,11 @@ class RuntimeTests(unittest.TestCase):
         self.assertEqual(json.loads(self.config.read_text())["port"], edited["port"])
         snapshot = Path(result["rollback_source_config"])
         self.assertEqual(snapshot.parent, self.root / "runtime")
-        self.assertEqual(snapshot.stat().st_mode & 0o077, 0)
+        if WINDOWS:
+            from strata_control import private_files
+            self.assertTrue(private_files.is_unshared(snapshot))      # NTFS has no mode bits: the ACL is the promise
+        else:
+            self.assertEqual(snapshot.stat().st_mode & 0o077, 0)
         self.assertEqual(json.loads(snapshot.read_text())["port"], 0)
         self.assertEqual(json.loads(snapshot.read_text())["api_key"], secret)
         self.assertNotIn(secret, json.dumps(result))
@@ -970,6 +983,7 @@ class RuntimeTests(unittest.TestCase):
             code = main(["ops", *args, "--profile", "local", "--output", "json"])
         return code, json.loads(output.getvalue())
 
+    @unittest.skipIf(WINDOWS, "SIGTERM to a Windows process is TerminateProcess: the hard-kill test below covers it")
     def test_terminating_owner_host_stops_its_child_and_removes_state(self):
         started = runtime.start("local", wait=10, shutdown_timeout=2)
         self.assertTrue(started["ok"])
@@ -990,9 +1004,18 @@ class RuntimeTests(unittest.TestCase):
         orphan = self.root / "runtime" / f"local-{state['token']}-{'a' * 32}.rollback.json"
         orphan.write_bytes(self.config.read_bytes())
         orphan.chmod(0o600)
-        os.kill(started["host_pid"], signal.SIGKILL)
         owned = runtime._STARTED_HOSTS.pop("local")
-        self.assertEqual(owned.wait(timeout=3), -signal.SIGKILL)
+        if WINDOWS:
+            owned.kill()                          # TerminateProcess: the host cannot clean up after itself
+            owned.wait(timeout=3)
+            # the job object ends the server tree with the host, no matter how the host ended
+            deadline = time.monotonic() + 15
+            while runtime._pid_exists(old_child) and time.monotonic() < deadline:
+                time.sleep(0.1)
+            self.assertFalse(runtime._pid_exists(old_child))
+        else:
+            os.kill(started["host_pid"], signal.SIGKILL)
+            self.assertEqual(owned.wait(timeout=3), -signal.SIGKILL)
         stale = runtime.status("local")
         self.assertEqual((stale["state"], stale["error"]), ("stale", "owner_host_lost"))
         # The next start waits for the inherited-pipe guard to stop the old
@@ -1006,7 +1029,8 @@ class RuntimeTests(unittest.TestCase):
     def test_recovery_refuses_a_still_present_child(self):
         state_path, socket_path, _ = runtime._paths("local")
         token = "a" * 64
-        state = {"schema_version": 1, "profile": "local", "socket": str(socket_path),
+        state = {"schema_version": 1, "profile": "local",
+                 "socket": "tcp:127.0.0.1:50000" if WINDOWS else str(socket_path),
                  "token": token, "host_pid": 999999}
         lease = runtime._write_new(state_path, (json.dumps(state) + "\n").encode())
         os.close(lease)
@@ -1019,6 +1043,137 @@ class RuntimeTests(unittest.TestCase):
             runtime.start("local", wait=0, shutdown_timeout=2)
         self.assertTrue(state_path.exists())
 
+    # ---- owner transport, attach and identity (POSIX socket and Windows loopback alike)
+
+    def _stale_host(self, child_pid, *, started=None, token="a" * 64):
+        """A state file whose host is gone, and the ownership marker naming `child_pid` as its server."""
+        state_path, socket_path, _ = runtime._paths("local")
+        state = {"schema_version": 1, "profile": "local",
+                 "socket": "tcp:127.0.0.1:50000" if WINDOWS else str(socket_path),
+                 "token": token, "host_pid": 999999}
+        lease = runtime._write_new(state_path, (json.dumps(state) + "\n").encode())
+        runtime._unlock_file(lease)
+        os.close(lease)
+        marker = runtime._child_marker("local", token)
+        record = {"host_token": token, "phase": "owned", "pid": child_pid,
+                  "ready": str(state_path.parent / "local-fake.ready.json")}
+        if started is not None:
+            record["started"] = started
+        marker.write_text(json.dumps(record), encoding="utf-8")
+        marker.chmod(0o600)
+        return state_path, marker
+
+    def _send(self, state, token, command):
+        client = owner_transport.connect(state["socket"], timeout=5)
+        try:
+            client.sendall((json.dumps({"token": token, "command": command}) + "\n").encode())
+            return json.loads(client.recv(4096).split(b"\n")[0])
+        finally:
+            client.close()
+
+    def test_a_second_client_attaches_from_another_process_and_shares_the_owner(self):
+        started = runtime.start("local", wait=10, shutdown_timeout=2)
+        self.assertTrue(started["ok"])
+        state = runtime._read_state("local")
+        if WINDOWS:
+            self.assertRegex(state["socket"], r"^tcp:127\.0\.0\.1:[1-9][0-9]*$")
+        root = Path(__file__).resolve().parents[2]
+        done = subprocess.run([sys.executable, "strata.py", "runtime", "status", "--profile", "local",
+                               "--output", "json"], cwd=root, env=os.environ.copy(), capture_output=True,
+                              text=True, timeout=30)
+        self.assertEqual(done.returncode, 0, done.stderr)
+        other = json.loads(done.stdout)
+        self.assertEqual((other["state"], other["owner_id"], other["instance_id"], other["host_pid"]),
+                         ("ready", started["owner_id"], started["instance_id"], started["host_pid"]))
+        self.assertEqual(runtime.status("local")["owner_id"], started["owner_id"])   # attaching took nothing over
+        self.assertEqual(runtime.stop("local", timeout=12)["ok"], True)
+        self.assertEqual(runtime.status("local")["state"], "stopped")
+
+    def test_a_wrong_or_stale_token_cannot_stop_the_owner(self):
+        started = runtime.start("local", wait=10, shutdown_timeout=2)
+        self.assertTrue(started["ok"])
+        state = runtime._read_state("local")
+        wrong_last = "0" if state["token"][-1] != "0" else "1"
+        for bad in ("0" * 64, state["token"][:-1] + wrong_last, "", "x" * 10):
+            with self.subTest(token=bad[:6]):
+                answer = self._send(state, bad, "stop")
+                self.assertEqual((answer["ok"], answer["error"]), (False, "unauthorized"))
+        self.assertEqual(runtime.status("local")["instance_id"], started["instance_id"])
+        self.assertTrue(self._send(state, state["token"], "status")["ok"])     # the real token is what authorises
+
+    def test_a_recorded_pid_of_an_unrelated_live_process_is_never_signalled(self):
+        victim = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"])
+        try:
+            state_path, marker = self._stale_host(victim.pid)
+            with self.assertRaisesRegex(profiles.ProfileError, "still present"):
+                runtime.start("local", wait=0, shutdown_timeout=2)
+            with self.assertRaises(profiles.ProfileError):            # nothing answers on the recorded address
+                runtime.stop("local", timeout=2)
+            self.assertIsNone(victim.poll(), "an unrelated process was stopped")
+            self.assertTrue(state_path.exists() and marker.exists())
+        finally:
+            victim.kill()
+            victim.wait(timeout=10)
+
+    @unittest.skipUnless(WINDOWS, "a creation time is recorded with the PID on Windows only")
+    def test_a_reused_pid_with_another_creation_time_is_not_the_recorded_child(self):
+        victim = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"])
+        try:
+            real = runtime.winproc.process_started(victim.pid)
+            self.assertTrue(runtime._pid_exists(victim.pid, real))
+            self.assertFalse(runtime._pid_exists(victim.pid, real + 12345))
+            self._stale_host(victim.pid, started=real + 12345)           # the number was reused by another process
+            started = runtime.start("local", wait=10, shutdown_timeout=2)
+            self.assertTrue(started["ok"])                               # recovered: it was not the recorded child
+            self.assertIsNone(victim.poll(), "the process that reused the PID was stopped")
+        finally:
+            victim.kill()
+            victim.wait(timeout=10)
+
+    @unittest.skipUnless(WINDOWS, "os.kill(pid, 0) ends a process on Windows; liveness is a handle query")
+    def test_liveness_check_does_not_end_the_process_it_asks_about(self):
+        victim = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"])
+        try:
+            for _ in range(5):
+                self.assertTrue(runtime._pid_exists(victim.pid))
+            self.assertIsNone(victim.poll())
+        finally:
+            victim.kill()
+            victim.wait(timeout=10)
+        self.assertFalse(runtime._pid_exists(victim.pid))
+
+    @unittest.skipUnless(WINDOWS, "ACLs stand in for mode 0700 on Windows")
+    def test_runtime_directory_is_owner_only_and_a_shared_one_is_refused(self):
+        from strata_control import private_files
+        directory = runtime.runtime_dir()
+        self.assertTrue(private_files.is_private(directory))
+        shared = Path(self.temp.name) / "shared-runtime"
+        shared.mkdir()
+        done = subprocess.run(["icacls", str(shared), "/grant", "*S-1-1-0:(OI)(CI)R"], capture_output=True)
+        if done.returncode != 0:
+            self.skipTest("icacls could not widen the test directory")
+        with mock.patch.dict(os.environ, {"STRATA_RUNTIME_DIR": str(shared)}):
+            with self.assertRaisesRegex(profiles.ProfileError, "owner-only"):
+                runtime.runtime_dir()
+
+    def test_exclusive_lease_waits_for_shared_holders_and_shared_holders_coexist(self):
+        first = runtime.acquire_activity_lease(shared=True)
+        second = runtime.acquire_activity_lease(shared=True)
+        try:
+            with self.assertRaisesRegex(profiles.ProfileError, "engine lease"):
+                runtime.acquire_activity_lease(shared=False)
+        finally:
+            for descriptor in (first, second):
+                runtime._unlock_file(descriptor)
+                os.close(descriptor)
+        exclusive = runtime.acquire_activity_lease(shared=False)
+        try:
+            with self.assertRaisesRegex(profiles.ProfileError, "engine lease"):
+                runtime.acquire_activity_lease(shared=True)
+        finally:
+            runtime._unlock_file(exclusive)
+            os.close(exclusive)
+
 
 if __name__ == "__main__":
     unittest.main()