diff --git a/README.md b/README.md
index 0a64ae50..430efbfd 100644
--- a/README.md
+++ b/README.md
@@ -90,6 +90,8 @@ New-ItemProperty -Path HKLM:\SOFTWARE\OpenSSH -Name DefaultShell -Value 'C:\Prog
 
 Then pair as above with `python herdr-push pair`. For a user in Administrators, Windows' sshd reads keys from `C:\ProgramData\ssh\administrators_authorized_keys`, so run `pair` in a terminal opened as administrator too (or add the key there by hand).
 
+For notifications, the PC has to be awake and signed in: the watcher runs in your signed-in session, and herdr-push needs a Python 3 there (the app finds `python3`, `python` or `py -3`, and skips the Microsoft Store's stand-in). See [Notifications while the app is closed](#notifications-while-the-app-is-closed).
+
 Machines reachable only through another one can use it as a jump host ("Connect through"). For zero-touch or read-only machines, **Allow host scripts** (on by default) can be turned off so the app skips copying its helper scripts (`herdr-usage`, `herdr-logins`, `herdr-push`) there. From a machine's menu, **Import herdr machines** adds the machines its herdr already knows (`herdr machine list`).
 
 ## How it talks to herdr
@@ -218,6 +220,8 @@ Two ways in, both on every machine the app knows:
 
 Each state is shown once, however it arrives, and comes down again when it is over (answered on the desktop, the agent back at work). The switch next to a machine in Settings leaves it out.
 
+**On a Windows PC** the app sets herdr-push up the same way, with no administrator rights: a `Run` entry under your user starts it when you sign in, and a scheduled task of yours starts it right away in your desktop session. It reads the Herdr desktop app's herdr-core (where one runs), RAM, CPU, drives and the biggest processes, and tells the phone how long ago you last used the keyboard or mouse and whether the session is locked. The PC has to be awake and signed in: with nobody signed in, herdr-push's install says so (and its `status` shows `interactive: false`), and it starts with the next sign-in. The phone needs ntfy (or another UnifiedPush app), as for any machine. Classic herdr, Shepherdr and App change are not there.
+
 ### What reaches the phone
 
 **Settings → Notifications → What reaches the phone** has a switch for each kind: **Needs you** (an agent asks, a machine runs low), **Finished** (an agent ended its turn), **Went wrong** (a review or a check failed) and **Limits** (an AI provider at 80% or 95% of a limit or out of it; see Usage and limits). Finished is off unless you turn it on: a ping for every finished turn teaches you to ignore the ones that matter. What is switched off isn't shown, and turning it on later starts from then.
diff --git a/core/src/test/kotlin/com/makefunstuff/herdr/core/PushTest.kt b/core/src/test/kotlin/com/makefunstuff/herdr/core/PushTest.kt
index cdcdf2b7..cf8de857 100644
--- a/core/src/test/kotlin/com/makefunstuff/herdr/core/PushTest.kt
+++ b/core/src/test/kotlin/com/makefunstuff/herdr/core/PushTest.kt
@@ -217,7 +217,7 @@ class HerdrPushTest {
     @Test
     fun bundledScriptIsTheOneInHost() {
         assertEquals(script.readText(), HerdrPush.script)
-        assertEquals("12", HerdrPush.version)
+        assertEquals("13", HerdrPush.version)
     }
 
     @Test
diff --git a/host/herdr-push/README.md b/host/herdr-push/README.md
index 7b70f792..427a75fc 100644
--- a/host/herdr-push/README.md
+++ b/host/herdr-push/README.md
@@ -36,6 +36,21 @@ A machine that is offline is set up the next time the app connects to it. Machin
 
 The app also **looks at its machines itself** while it is closed (every 15 minutes by default; Settings → Notifications → The app also looks), with an exact alarm. That covers a machine without python3, or a push that got lost. Pushes arrive within seconds; the app's own looks only as often as you set.
 
+## On Windows
+
+herdr-push runs on a Windows PC with Git Bash under OpenSSH Server (see the README's "A Windows machine"), with Python 3.6 or newer. The app installs it as it does elsewhere; by hand, `python herdr-push install` does the same.
+
+It starts without administrator rights, in two ways:
+
+- A `Run` value under `HKCU\Software\Microsoft\Windows\CurrentVersion\Run`, `"<pythonw.exe>" "<script>" run`, starts it when you sign in.
+- A scheduled task of yours named `herdr-push`, with no time limit, which `install` and `ensure` run to start it now. It starts in your desktop session, because a process started from the SSH session ends with it.
+
+`uninstall` removes both. `status` says `"autostart": "windows-run"` and `"interactive": true` when the watcher can run in a signed-in session, `false` when nobody is signed in: the PC has to be awake and signed in for pushes to come.
+
+`presence` run over SSH on Windows reads as away and locked (an SSH session has no input desktop); the watcher, running in the signed-in session, sends the real value.
+
+What works there: agents in herdr-core (read through `herdr-core connect`, only where `herdr-core server status` says a core runs; classic herdr is not on Windows), RAM, CPU, each fixed drive (as `/c`, `/d`…) and the top processes, and presence: how long ago the PC had keyboard or mouse input, and whether the session is locked. Shepherdr and App change are not there.
+
 ## What it puts on the machine
 
 | Path | What |
@@ -46,6 +61,7 @@ The app also **looks at its machines itself** while it is closed (every 15 minut
 | `~/.config/systemd/user/herdr-push.service` | Linux with systemd: a user service, `Restart=always` |
 | `~/Library/LaunchAgents/dev.herdr.push.plist` | macOS: a launchd agent |
 | an `@reboot` line in your crontab | elsewhere, when there is `crontab` |
+| the `Run` value and the `herdr-push` scheduled task of your user | Windows (see above) |
 
 On Linux, a user service stops when your last login ends unless lingering is on. herdr-push turns it on (`loginctl enable-linger`) where the system lets you; where it doesn't, the app's Settings say "it stops when you log out", and an admin can run `loginctl enable-linger <you>`. Without any of these, the app starts herdr-push again whenever it connects.
 
@@ -53,7 +69,7 @@ On Linux, a user service stops when your last login ends unless lingering is on.
 
 ```sh
 P=~/.local/share/herdr-push/herdr-push
-python3 $P status          # running?, start-up entry, phones, last error
+python3 $P status          # running?, start-up entry, phones, last error (on Windows also interactive: whether a user is signed in)
 python3 $P once --dry-run  # one look: what it would send now (the first look only learns how things stand)
 python3 $P test            # a test notification to every phone
 python3 $P journal         # what it kept for the brief (--since <unix seconds>)
@@ -103,7 +119,7 @@ Anyone who sees the code within its five minutes could pair with it, as with any
 
 It touches one line in `~/.ssh/authorized_keys` while it waits (then the phone's key is there instead), and `pair-<token>.json` and `pair.lock` in `~/.local/state/herdr-push/`. It needs neither the watcher running nor a phone subscribed.
 
-On Windows (OpenSSH Server with Git Bash as its `DefaultShell`), `pair` is the one command that runs: `python herdr-push pair`. It reads the host keys and `sshd_config` in `C:\ProgramData\ssh`, writes the one-time key's command with forward-slash paths for Bash, and, for a user in Administrators while the stock `Match Group administrators` block is there, uses `administrators_authorized_keys`, which only a terminal opened as administrator can change.
+On Windows (OpenSSH Server with Git Bash as its `DefaultShell`), `python herdr-push pair` works too. It reads the host keys and `sshd_config` in `C:\ProgramData\ssh`, writes the one-time key's command with forward-slash paths for Bash, and, for a user in Administrators while the stock `Match Group administrators` block is there, uses `administrators_authorized_keys`, which only a terminal opened as administrator can change.
 
 It can't help when sshd reads keys from elsewhere (`AuthorizedKeysFile` in `/etc/ssh/sshd_config`; the phone then says the machine still refuses its key), or when the phone reaches the machine only through another one (a jump host): add those by hand.
 
diff --git a/host/herdr-push/herdr-push b/host/herdr-push/herdr-push
index 47b82bbb..8b789cff 100755
--- a/host/herdr-push/herdr-push
+++ b/host/herdr-push/herdr-push
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-# version: 12
+# version: 13
 """herdr-push: tells your phone when a herdr agent needs you.
 
 It runs on the machine with herdr, watches its agents (`herdr api snapshot`, and the Herdr desktop
@@ -33,7 +33,7 @@ that wasn't sealed with it. Python 3 alone, no packages.
   herdr-push test [<id>]         send a test notification now
   herdr-push status              how it stands, as JSON
   herdr-push ensure              start the watcher when it isn't running
-  herdr-push install             start it now and with the machine (systemd, launchd or cron)
+  herdr-push install             start it now and with the machine (systemd, launchd, cron or, on Windows, your sign-in)
   herdr-push uninstall [--purge] stop it and remove the start-up entry (--purge: and its settings)
   herdr-push once [--dry-run]    one look, printing what it would send
   herdr-push journal [--session NAME] [--since SECONDS]
@@ -42,7 +42,9 @@ that wasn't sealed with it. Python 3 alone, no packages.
   herdr-push pair [--host NAME[,NAME…]] [--port N] [--session NAME] [--name NAME]
                                  show a QR code a phone scans to add this machine (one-time, 5 minutes)
 
-On Windows (OpenSSH with Git Bash as its shell) only `pair` runs: notifications need Linux or macOS.
+On Windows (OpenSSH with Git Bash as its shell) it runs in your own desktop session, started when you sign in (a
+Run entry) and, from `install` or `ensure`, through a scheduled task of yours: never as a child of the SSH session.
+No elevation is asked.
 
 See host/herdr-push/README.md in the herdr-android repo.
 """
@@ -52,12 +54,16 @@ import hashlib
 import hmac
 import json
 import os
+import queue
 import re
 import shutil
 import signal
 import socket
+import struct
 import subprocess
 import sys
+import tempfile
+import threading
 import time
 import urllib.error
 import urllib.parse
@@ -66,10 +72,10 @@ import zlib
 
 try:
     import fcntl
-except ImportError:  # Windows: only `pair` runs there
+except ImportError:  # Windows: msvcrt's locks stand in for it
     fcntl = None
 
-VERSION = "12"
+VERSION = "13"
 NAME = "herdr-push"
 
 HOME = os.path.expanduser("~")
@@ -84,6 +90,13 @@ LOG = os.path.join(STATE_DIR, NAME + ".log")
 # Touched by anything that knows you are here (a shell prompt, an editor hook): counts as typing in herdr.
 PRESENT = os.path.join(STATE_DIR, "present")
 WINDOWS = os.name == "nt"
+NO_WINDOW = 0x08000000 if WINDOWS else 0  # CREATE_NO_WINDOW for the commands run: a watcher without a console opens none
+# The one byte of the lock file that a running watcher locks. Windows locks are mandatory, so it is not one that holds the pid.
+LOCK_BYTE = 1 << 20
+RUN_KEY = r"Software\Microsoft\Windows\CurrentVersion\Run"  # where Windows starts a user's programs at sign-in
+_KERNEL32 = None  # kernel32 once loaded, see kernel32()
+_USER32 = None  # user32 once loaded, see user32()
+_WTSAPI32 = None  # wtsapi32 once loaded, see wtsapi32()
 # Where sshd keeps its config and host keys: Win32-OpenSSH's under ProgramData.
 SSHD_DIR = os.path.join(os.environ.get("PROGRAMDATA") or "C:\\ProgramData", "ssh") if WINDOWS else "/etc/ssh"
 
@@ -183,7 +196,15 @@ def write_json(path, value, private=False):
     fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600 if private else 0o644)
     with os.fdopen(fd, "w") as f:
         json.dump(value, f, indent=1)
-    os.replace(tmp, path)
+    for attempt in range(5):
+        try:
+            os.replace(tmp, path)
+            return
+        except PermissionError:
+            # Windows refuses the swap while another process has the target open that instant: try again shortly.
+            if not WINDOWS or attempt == 4:
+                raise
+            time.sleep(0.05)
 
 
 def subscribers():
@@ -229,7 +250,7 @@ class Source:
 
     def run(self, args, timeout=20):
         r = subprocess.run(args, env=self.env(), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
-                           stderr=subprocess.PIPE, timeout=timeout)
+                           stderr=subprocess.PIPE, timeout=timeout, creationflags=NO_WINDOW)
         if r.returncode != 0:
             raise RuntimeError((r.stderr.decode(errors="replace").strip().splitlines() or ["exit %d" % r.returncode])[-1])
         return r.stdout.decode(errors="replace")
@@ -330,6 +351,89 @@ class HerdrCore:
         return self.call("terminal.read", {"terminal": address, "lines": 0}).get("text", "")
 
 
+def herdr_core_exe():
+    """Windows: where herdr-core.exe is, on the PATH or where its installer puts it; None when it isn't there."""
+    exe = shutil.which("herdr-core")
+    if not exe:
+        exe = os.path.join(os.environ.get("LOCALAPPDATA") or "", "Programs", "herdr-core", "herdr-core.exe")
+    return exe if os.path.isfile(exe) else None
+
+
+class HcPipe:
+    """herdr-core's API through `herdr-core connect` (JSON-RPC lines over its stdin and stdout), for Windows, where the
+    core has no Unix socket. Same calls as HerdrCore. It never starts a core: the caller checks one runs first."""
+
+    def __init__(self, exe):
+        self.exe = exe
+        self.proc = None
+        self.lines = None
+        self.n = 0
+
+    def call(self, method, params=None):
+        if self.proc is None:
+            self.proc = subprocess.Popen([self.exe, "connect"], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                                         stderr=subprocess.DEVNULL, creationflags=NO_WINDOW)
+            self.lines = queue.Queue()
+            # A pipe has no read timeout: a thread reads it and the calls wait on the queue.
+            threading.Thread(target=self._read, args=(self.proc.stdout, self.lines), daemon=True).start()
+            self._rpc("hello", {"generation": 1})
+        return self._rpc(method, params or {})
+
+    @staticmethod
+    def _read(f, lines):
+        try:
+            for line in f:
+                lines.put(line)
+        except (OSError, ValueError):
+            pass
+        lines.put(None)
+
+    def _rpc(self, method, params):
+        self.n += 1
+        try:
+            self.proc.stdin.write((json.dumps({"jsonrpc": "2.0", "id": self.n, "method": method, "params": params}) + "\n").encode())
+            self.proc.stdin.flush()
+            # Short, as HerdrCore's: a core that doesn't answer mustn't hold up classic herdr's notifications.
+            end = time.time() + 3
+            while True:
+                try:
+                    line = self.lines.get(timeout=max(end - time.time(), 0.01))
+                except queue.Empty:
+                    raise RuntimeError("herdr-core did not answer")
+                if line is None:
+                    raise RuntimeError("herdr-core closed the connection")
+                out = json.loads(line.decode(errors="replace"))
+                if out.get("id") == self.n:
+                    break
+        except Exception:
+            # What is left of this connection is of no use: the next call starts a new one.
+            self.close()
+            raise
+        if out.get("error"):
+            raise RuntimeError(out["error"].get("message") or "herdr-core: error")
+        return out.get("result") or {}
+
+    def close(self):
+        p, self.proc = self.proc, None
+        if p is None:
+            return
+        # Killed first: closing stdout waits for the reader thread, which waits for the child. A child that
+        # outlives the kill keeps its stdout open rather than hang the watcher.
+        try:
+            p.kill()
+            p.wait(timeout=3)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+        for c in (p.stdin, p.stdout) if p.poll() is not None else (p.stdin,):
+            try:
+                c.close()
+            except (OSError, ValueError):
+                pass
+
+    def screen(self, address):
+        return self.call("terminal.read", {"terminal": address, "lines": 0}).get("text", "")
+
+
 def hc_agent_name(t):
     app = ((t.get("status") or {}).get("app") or "").strip().lower()
     if app:
@@ -537,8 +641,190 @@ def live_event(st, agents, now, first):
 
 # ---- health (RAM, CPU, disks) --------------------------------------------------------------------
 
+def kernel32():
+    """Windows: kernel32 with the signatures of the calls health uses (64-bit safe), loaded once."""
+    global _KERNEL32
+    if _KERNEL32 is None:
+        import ctypes
+        from ctypes import wintypes
+        k = ctypes.WinDLL("kernel32", use_last_error=True)
+        k.GlobalMemoryStatusEx.argtypes = [ctypes.c_void_p]
+        k.GlobalMemoryStatusEx.restype = wintypes.BOOL
+        k.GetSystemTimes.argtypes = [ctypes.POINTER(wintypes.FILETIME)] * 3
+        k.GetSystemTimes.restype = wintypes.BOOL
+        k.GetLogicalDrives.argtypes = []
+        k.GetLogicalDrives.restype = wintypes.DWORD
+        k.GetDriveTypeW.argtypes = [wintypes.LPCWSTR]
+        k.GetDriveTypeW.restype = wintypes.UINT
+        k.GetDiskFreeSpaceExW.argtypes = [wintypes.LPCWSTR] + [ctypes.POINTER(ctypes.c_ulonglong)] * 3
+        k.GetDiskFreeSpaceExW.restype = wintypes.BOOL
+        k.SetThreadErrorMode.argtypes = [wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)]
+        k.SetThreadErrorMode.restype = wintypes.BOOL
+        k.GetTickCount64.argtypes = []
+        k.GetTickCount64.restype = ctypes.c_ulonglong
+        k.GetCurrentProcessId.argtypes = []
+        k.GetCurrentProcessId.restype = wintypes.DWORD
+        k.ProcessIdToSessionId.argtypes = [wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)]
+        k.ProcessIdToSessionId.restype = wintypes.BOOL
+        _KERNEL32 = k
+    return _KERNEL32
+
+
+def user32():
+    """Windows: user32 with the signatures of the calls presence uses (64-bit safe), loaded once."""
+    global _USER32
+    if _USER32 is None:
+        import ctypes
+        from ctypes import wintypes
+        u = ctypes.WinDLL("user32", use_last_error=True)
+        u.GetLastInputInfo.argtypes = [ctypes.c_void_p]
+        u.GetLastInputInfo.restype = wintypes.BOOL
+        u.OpenInputDesktop.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
+        u.OpenInputDesktop.restype = ctypes.c_void_p
+        u.CloseDesktop.argtypes = [ctypes.c_void_p]
+        u.CloseDesktop.restype = wintypes.BOOL
+        _USER32 = u
+    return _USER32
+
+
+def wtsapi32():
+    """Windows: wtsapi32 with the signatures of the calls presence uses (64-bit safe), loaded once."""
+    global _WTSAPI32
+    if _WTSAPI32 is None:
+        import ctypes
+        from ctypes import wintypes
+        w = ctypes.WinDLL("wtsapi32", use_last_error=True)
+        w.WTSQuerySessionInformationW.argtypes = [ctypes.c_void_p, wintypes.DWORD, ctypes.c_int,
+                                                  ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(wintypes.DWORD)]
+        w.WTSQuerySessionInformationW.restype = wintypes.BOOL
+        w.WTSEnumerateProcessesW.argtypes = [ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD,
+                                             ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(wintypes.DWORD)]
+        w.WTSEnumerateProcessesW.restype = wintypes.BOOL
+        w.WTSFreeMemory.argtypes = [ctypes.c_void_p]
+        w.WTSFreeMemory.restype = None
+        _WTSAPI32 = w
+    return _WTSAPI32
+
+
+def windows_swap():
+    """(total, in use) bytes of the page files, from NtQuerySystemInformation(SystemPageFileInformation); None if not told."""
+    import ctypes
+    ntdll = ctypes.WinDLL("ntdll")
+    ntdll.NtQuerySystemInformation.argtypes = [ctypes.c_ulong, ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(ctypes.c_ulong)]
+    ntdll.NtQuerySystemInformation.restype = ctypes.c_long
+    size = 4096
+    for _ in range(4):
+        buf = ctypes.create_string_buffer(size)
+        need = ctypes.c_ulong()
+        status = ntdll.NtQuerySystemInformation(18, buf, size, ctypes.byref(need))  # SystemPageFileInformation
+        if status & 0xFFFFFFFF == 0xC0000004:  # STATUS_INFO_LENGTH_MISMATCH
+            size = max(need.value, size * 2)
+            continue
+        if status != 0:
+            return None
+        # A chain of entries: next entry's offset, then size, in use and peak in 4 KiB pages.
+        total = used = offset = 0
+        while True:
+            nxt, pages, in_use = struct.unpack_from("<III", buf, offset)
+            total, used = total + pages * 4096, used + in_use * 4096
+            if not nxt:
+                return total, used
+            offset += nxt
+    return None
+
+
+def windows_memory():
+    """read_memory on Windows: GlobalMemoryStatusEx for the RAM, the page files for the swap."""
+    import ctypes
+
+    class MemoryStatus(ctypes.Structure):
+        _fields_ = [("length", ctypes.c_ulong), ("load", ctypes.c_ulong), ("total_phys", ctypes.c_ulonglong),
+                    ("avail_phys", ctypes.c_ulonglong), ("total_page", ctypes.c_ulonglong),
+                    ("avail_page", ctypes.c_ulonglong), ("total_virtual", ctypes.c_ulonglong),
+                    ("avail_virtual", ctypes.c_ulonglong), ("avail_extended", ctypes.c_ulonglong)]
+
+    m = MemoryStatus()
+    m.length = ctypes.sizeof(m)
+    if not kernel32().GlobalMemoryStatusEx(ctypes.byref(m)):
+        return None
+    try:
+        swap = windows_swap()
+    except Exception:  # noqa: BLE001
+        swap = None
+    swap_total, swap_used = swap or (0, 0)
+    return {"total": m.total_phys // 1024, "available": min(m.avail_phys, m.total_phys) // 1024,
+            "swap_total": swap_total // 1024, "swap_free": max(swap_total - swap_used, 0) // 1024}
+
+
+def windows_cpu():
+    """read_cpu on Windows, from GetSystemTimes: the kernel time already includes the idle time."""
+    from ctypes import wintypes
+    idle, kernel, user = wintypes.FILETIME(), wintypes.FILETIME(), wintypes.FILETIME()
+    if not kernel32().GetSystemTimes(idle, kernel, user):
+        return None
+
+    def ticks(t):
+        return (t.dwHighDateTime << 32) | t.dwLowDateTime
+    return (ticks(kernel) + ticks(user), ticks(idle))
+
+
+def windows_disks():
+    """read_disks on Windows: each fixed drive (not a card reader, a network drive or a CD) as /c, /d. A drive that
+    can't be read is left out."""
+    import ctypes
+    from ctypes import wintypes
+    k = kernel32()
+    # No "insert a disk" dialog for a drive that is not ready: the call just fails.
+    old = wintypes.DWORD()
+    k.SetThreadErrorMode(0x0001, ctypes.byref(old))  # SEM_FAILCRITICALERRORS
+    disks = []
+    try:
+        mask = k.GetLogicalDrives()
+        for i in range(26):
+            letter = chr(ord("a") + i)
+            root = letter.upper() + ":\\"
+            if not mask & (1 << i) or k.GetDriveTypeW(root) != 3:  # DRIVE_FIXED
+                continue
+            avail, total, free = ctypes.c_ulonglong(), ctypes.c_ulonglong(), ctypes.c_ulonglong()
+            if not k.GetDiskFreeSpaceExW(root, ctypes.byref(avail), ctypes.byref(total), ctypes.byref(free)) or total.value <= 0:
+                continue
+            disks.append({"mount": "/" + letter, "device": letter.upper() + ":", "total": total.value // 1024,
+                          "used": (total.value - free.value) // 1024, "avail": avail.value // 1024})
+    finally:
+        k.SetThreadErrorMode(old, None)
+    return disks
+
+
+def windows_proc_name(name, cmdline):
+    """proc_name for a Windows process: its image name without .exe, or the nicer name proc_name gives."""
+    exe = name[:-4] if name.lower().endswith(".exe") else name
+    label = proc_name((name.replace(" ", "") + " " + cmdline).replace("\\", "/"))
+    return exe if label == name.replace(" ", "") else label
+
+
+def windows_top_memory(n):
+    """top_memory on Windows: the working set of each process, from CIM through PowerShell."""
+    script = ("[Console]::OutputEncoding = [Text.Encoding]::UTF8; Get-CimInstance Win32_Process | "
+              "Sort-Object WorkingSetSize -Descending | Select-Object -First %d | ForEach-Object { "
+              "\"$($_.WorkingSetSize)`t$($_.Name)`t$($_.CommandLine)\" -replace '[\\r\\n]+', ' ' }" % n)
+    try:
+        r = subprocess.run(["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", script], stdin=subprocess.DEVNULL,
+                           stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=30, creationflags=NO_WINDOW)
+    except Exception:  # noqa: BLE001
+        return []
+    rows = []
+    for line in r.stdout.decode("utf-8", "replace").splitlines():
+        parts = line.split("\t", 2)
+        if len(parts) == 3 and parts[0].strip().isdigit():
+            rows.append([windows_proc_name(parts[1], parts[2]), int(parts[0]) // 1024])
+    return rows[:n]
+
+
 def read_memory():
-    """{"total", "available", "swap_total", "swap_free"} in KiB, from /proc/meminfo; None elsewhere."""
+    """{"total", "available", "swap_total", "swap_free"} in KiB, from /proc/meminfo (GlobalMemoryStatusEx on Windows);
+    None elsewhere."""
+    if WINDOWS:
+        return windows_memory()
     try:
         with open(MEMINFO) as f:
             m = {}
@@ -557,7 +843,9 @@ def read_memory():
 
 
 def read_cpu():
-    """(total, idle) CPU ticks so far, from /proc/stat; None elsewhere."""
+    """(total, idle) CPU ticks so far, from /proc/stat (GetSystemTimes on Windows); None elsewhere."""
+    if WINDOWS:
+        return windows_cpu()
     try:
         with open("/proc/stat") as f:
             x = [int(v) for v in f.readline().split()[1:9]]
@@ -568,6 +856,8 @@ def read_cpu():
 
 def read_disks():
     """The real filesystems from `df -kPT` (or `-kP`), as the app picks them: [{"mount", "total", "used", "avail"}]."""
+    if WINDOWS:
+        return windows_disks()
     out = ""
     if DF:
         with open(DF) as f:
@@ -627,6 +917,8 @@ def proc_name(args):
 
 def top_memory(n=3):
     """The [n] processes using the most memory: [name, KiB], biggest first."""
+    if WINDOWS:
+        return windows_top_memory(n)
     try:
         r = subprocess.run(["ps", "-axo", "rss=,args="], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                            stderr=subprocess.DEVNULL, timeout=15)
@@ -710,7 +1002,7 @@ def usage_look(st, now):
     st["at"] = now
     try:
         r = subprocess.run([sys.executable, USAGE_SCRIPT, "limits"], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
-                           stderr=subprocess.DEVNULL, timeout=120)
+                           stderr=subprocess.DEVNULL, timeout=120, creationflags=NO_WINDOW)
         lines = [l for l in r.stdout.decode("utf-8", "replace").splitlines() if l.strip().startswith("{")]
         out = json.loads(lines[-1]) if lines else {}
     except Exception as e:  # noqa: BLE001
@@ -834,13 +1126,39 @@ def herdr_client(argv):
     return True
 
 
+def windows_last_input(now=None):
+    """Windows: when this session last had keyboard or mouse input, or None if it can't tell. There is no input to read
+    from an SSH session, so this only means something where the watcher runs in the interactive one."""
+    import ctypes
+    from ctypes import wintypes
+
+    class LASTINPUTINFO(ctypes.Structure):
+        _fields_ = [("cbSize", wintypes.UINT), ("dwTime", wintypes.DWORD)]
+
+    info = LASTINPUTINFO(ctypes.sizeof(LASTINPUTINFO), 0)
+    try:
+        if not user32().GetLastInputInfo(ctypes.byref(info)):
+            return None
+        # dwTime is a 32-bit tick count that wraps every 49.7 days, so take the difference modulo 2**32.
+        idle_ms = ((kernel32().GetTickCount64() & 0xFFFFFFFF) - info.dwTime) % 2 ** 32
+    except Exception:  # noqa: BLE001 - no user32 or no desktop: it can't tell
+        return None
+    return (time.time() if now is None else now) - idle_ms / 1000.0
+
+
 def typed_at(ps=None):
-    """When someone last typed in a herdr client here (its terminal's last input), or touched the presence file; None."""
+    """When someone last typed in a herdr client here (its terminal's last input), or touched the presence file; None.
+    On Windows it is the last keyboard or mouse input of the session instead (see windows_last_input)."""
     times = []
     try:
         times.append(os.stat(PRESENT).st_mtime)
     except OSError:
         pass
+    if WINDOWS:
+        last = windows_last_input()
+        if last is not None:
+            times.append(last)
+        return max(times) if times else None
     ps = _out(["ps", "-A", "-o", "tty=,args="]) if ps is None else ps
     for line in (ps or "").splitlines():
         parts = line.split()
@@ -858,6 +1176,8 @@ def screen_unlocked():
     forced = os.environ.get("HERDR_PUSH_SCREEN")  # tests: locked, unlocked or none
     if forced:
         return {"locked": False, "unlocked": True}.get(forced)
+    if WINDOWS:
+        return windows_unlocked()
     if sys.platform == "darwin":
         return _mac_unlocked()
     sessions = _out(["loginctl", "list-sessions", "--no-legend"])
@@ -881,6 +1201,110 @@ def screen_unlocked():
     return desktop
 
 
+def windows_session_id():
+    """Windows: the session id of this process, or None if it can't tell."""
+    import ctypes
+    from ctypes import wintypes
+
+    k = kernel32()
+    mine = wintypes.DWORD()
+    return mine.value if k.ProcessIdToSessionId(k.GetCurrentProcessId(), ctypes.byref(mine)) else None
+
+
+def windows_sessions(exe):
+    """Windows: the session id of each running process whose image is called [exe] (lower case), and of this one:
+    ([ids], mine), mine None if it can't tell. Asks the terminal service for the process list, which gives every process
+    its session (ProcessIdToSessionId is refused for SYSTEM ones like LogonUI.exe), in a few milliseconds and with no
+    process spawned. None instead if the list can't be read."""
+    import ctypes
+    from ctypes import wintypes
+
+    class WTS_PROCESS_INFOW(ctypes.Structure):
+        _fields_ = [("SessionId", wintypes.DWORD), ("ProcessId", wintypes.DWORD), ("pProcessName", wintypes.LPWSTR),
+                    ("pUserSid", ctypes.c_void_p)]
+
+    w = wtsapi32()
+    first = ctypes.c_void_p()
+    count = wintypes.DWORD()
+    # Level 0 (WTS_PROCESS_INFOW), version 1, on this server (a null handle).
+    if not w.WTSEnumerateProcessesW(None, 0, 1, ctypes.byref(first), ctypes.byref(count)):
+        return None
+    try:
+        rows = ctypes.cast(first, ctypes.POINTER(WTS_PROCESS_INFOW))
+        ids = [rows[i].SessionId for i in range(count.value) if (rows[i].pProcessName or "").lower() == exe]
+    finally:
+        w.WTSFreeMemory(first)
+    return ids, windows_session_id()
+
+
+def windows_lock_screen():
+    """Windows: whether LogonUI.exe, the lock screen, runs in this session. It runs once per session, so the lock screen of
+    another signed-in user doesn't count."""
+    found = windows_sessions("logonui.exe")
+    if found is None:
+        return False
+    ids, mine = found
+    return any(mine is None or i == mine for i in ids)
+
+
+def windows_interactive():
+    """Windows: whether someone is signed in at the PC, which is when explorer.exe runs in a session other than 0 (services
+    and SSH sessions live in session 0, a desktop session never does)."""
+    found = windows_sessions("explorer.exe")
+    return found is not None and any(i != 0 for i in found[0])
+
+
+def windows_session_locked():
+    """Windows: whether the terminal service says this session is locked: True or False, or None if it doesn't say (the
+    call fails, the flag is unknown, or this is session 0, which has no desktop to lock)."""
+    import ctypes
+    from ctypes import wintypes
+
+    class LEVEL1(ctypes.Structure):  # the start of WTSINFOEX_LEVEL1_W, the part that is read
+        _fields_ = [("SessionId", wintypes.ULONG), ("SessionState", ctypes.c_int), ("SessionFlags", ctypes.c_long)]
+
+    class LEVELS(ctypes.Union):  # the union is 8-byte aligned (it has a LARGE_INTEGER), so Level1 sits at offset 8
+        _fields_ = [("Level1", LEVEL1), ("align", ctypes.c_longlong)]
+
+    class WTSINFOEXW(ctypes.Structure):
+        _fields_ = [("Level", wintypes.DWORD), ("Data", LEVELS)]
+
+    mine = windows_session_id()
+    if not mine:  # None, or 0: an SSH session or a service
+        return None
+    w = wtsapi32()
+    buf = ctypes.c_void_p()
+    size = wintypes.DWORD()
+    if not w.WTSQuerySessionInformationW(None, mine, 25, ctypes.byref(buf), ctypes.byref(size)):  # WTSSessionInfoEx
+        return None
+    try:
+        info = ctypes.cast(buf, ctypes.POINTER(WTSINFOEXW)).contents
+        flags = info.Data.Level1.SessionFlags if info.Level == 1 else None
+    finally:
+        w.WTSFreeMemory(buf)
+    return {0: True, 1: False}.get(flags)  # WTS_SESSIONSTATE_LOCK is 0, _UNLOCK is 1, _UNKNOWN is -1
+
+
+def windows_unlocked():
+    """Windows: whether this session's screen is unlocked. The terminal service's lock flag for the session says it; where
+    it doesn't (it can't tell, or this is an SSH session), it is locked while the lock screen runs or the input desktop
+    can't be opened (it is the secure one then). Only means something in the interactive session: an SSH session has no
+    input desktop of its own, so it reads as locked there."""
+    try:
+        locked = windows_session_locked()
+        if locked is not None:
+            return not locked
+        if windows_lock_screen():
+            return False
+        desktop = user32().OpenInputDesktop(0, False, 0x100)  # DESKTOP_SWITCHDESKTOP
+        if not desktop:
+            return False
+        user32().CloseDesktop(desktop)
+        return True
+    except Exception:  # noqa: BLE001 - it can't tell
+        return None
+
+
 def _mac_unlocked():
     out = _out(["ioreg", "-n", "Root", "-d1", "-a"])
     if not out:
@@ -1014,11 +1438,21 @@ _classic_last = {}
 
 def core_look(src, st, now):
     """herdr-core's agents where a core runs (else None), with [src] set to read their screens."""
-    path = hc_socket()
+    path = None if WINDOWS else hc_socket()
     try:
-        if not os.path.exists(path):
-            raise RuntimeError("no herdr-core socket")
-        src.core = HerdrCore(path)
+        if WINDOWS:
+            exe = herdr_core_exe()
+            if not exe:
+                raise RuntimeError("no herdr-core")
+            # `server status` exits 0 only while a core runs, so a machine without one is never given one.
+            if subprocess.call([exe, "server", "status", "--json"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
+                               stderr=subprocess.DEVNULL, timeout=5, creationflags=NO_WINDOW) != 0:
+                raise RuntimeError("no herdr-core running")
+            src.core = HcPipe(exe)
+        else:
+            if not os.path.exists(path):
+                raise RuntimeError("no herdr-core socket")
+            src.core = HerdrCore(path)
         snap = hc_snapshot(src.core.call("terminal.list").get("terminals") or [])
         _core_last[src.key] = (snap, now)
         st.pop("core_problem", None)
@@ -1060,88 +1494,91 @@ def tick(state, n, dry_run=False, now=None):
         return seen_here[0]
 
     srcs = sources(subs)
-    for key, (src, group) in srcs.items():
-        st = state.setdefault("sessions", {}).setdefault(key, {})
-        live = any(s.get("live") for s in group)
-        watch_panes = watched(group, "watch_panes")
-        events = []
-        agents = None
-        # What the journal saw last; its first look only learns how things stand.
-        jst = st.setdefault("journal", {})
-        noted = []
-        core = core_look(src, st, now) if src.session == "default" else None
-        # herdr-core's agents last told, on a machine without classic herdr: they go when the core does.
-        told_core = any(p.startswith(HC_PREFIX) for p in st.get("agents") or {})
-        if src.herdr or core is not None or told_core:
+    try:
+        for key, (src, group) in srcs.items():
+            st = state.setdefault("sessions", {}).setdefault(key, {})
+            live = any(s.get("live") for s in group)
+            watch_panes = watched(group, "watch_panes")
+            events = []
+            agents = None
+            # What the journal saw last; its first look only learns how things stand.
+            jst = st.setdefault("journal", {})
+            noted = []
+            core = core_look(src, st, now) if src.session == "default" else None
+            # herdr-core's agents last told, on a machine without classic herdr: they go when the core does.
+            told_core = any(p.startswith(HC_PREFIX) for p in st.get("agents") or {})
+            if src.herdr or core is not None or told_core:
+                try:
+                    snap = {"workspaces": [], "tabs": [], "agents": []}
+                    if src.herdr:
+                        try:
+                            snap = src.snapshot()
+                            _classic_last[key] = snap
+                            st.pop("classic_problem", None)
+                        except Exception:  # noqa: BLE001 - herdr-core's agents are still told, classic herdr's kept as they were
+                            # Without classic herdr's last snapshot, one that had agents keeps them by not looking now.
+                            told_classic = any(not p.startswith(HC_PREFIX) for p in st.get("agents") or {})
+                            if key not in _classic_last and (told_classic or (core is None and not told_core)):
+                                raise
+                            log_once(st, "classic_problem", "herdr session %s:" % key, sys.exc_info()[1])
+                            snap = _classic_last.get(key, snap)
+                    if core is not None:
+                        snap = merged(snap, core)
+                        # A core seen for the first time (herdr-push updated to read it): its agents
+                        # as they stand are learnt, not told, as on a first look.
+                        if not st.get("core_known"):
+                            for a in core.get("agents") or []:
+                                if "agents" in st:
+                                    st["agents"].setdefault(a["pane_id"], a.get("agent_status", "unknown"))
+                                if "agents" in jst:
+                                    jst["agents"].setdefault(a["pane_id"], "%s#%s" % (a.get("agent_status", "unknown"), a.get("state_change_seq", "")))
+                        st["core_known"] = True
+                    first = "agents" not in st
+                    told = agent_events(src, snap, st.setdefault("agents", {}), first)
+                    events += told
+                    events += watch_agent_events(snap, st.setdefault("watch_agents", {}), watch_panes,
+                                                 {e["agent"].get("pane_id") for e in told})
+                    learn = "agents" not in jst or first
+                    noted += journal_agents(snap, jst.setdefault("agents", {}), learn, now)
+                    if live:
+                        agents = live_agents(snap, st, now)
+                    st.pop("problem", None)
+                except Exception as e:  # noqa: BLE001
+                    if st.get("problem") != str(e):
+                        log("herdr session %s:" % key, e)
+                    st["problem"] = str(e)
             try:
-                snap = {"workspaces": [], "tabs": [], "agents": []}
-                if src.herdr:
-                    try:
-                        snap = src.snapshot()
-                        _classic_last[key] = snap
-                        st.pop("classic_problem", None)
-                    except Exception:  # noqa: BLE001 - herdr-core's agents are still told, classic herdr's kept as they were
-                        # Without classic herdr's last snapshot, one that had agents keeps them by not looking now.
-                        told_classic = any(not p.startswith(HC_PREFIX) for p in st.get("agents") or {})
-                        if key not in _classic_last and (told_classic or (core is None and not told_core)):
-                            raise
-                        log_once(st, "classic_problem", "herdr session %s:" % key, sys.exc_info()[1])
-                        snap = _classic_last.get(key, snap)
-                if core is not None:
-                    snap = merged(snap, core)
-                    # A core seen for the first time (herdr-push updated to read it): its agents
-                    # as they stand are learnt, not told, as on a first look.
-                    if not st.get("core_known"):
-                        for a in core.get("agents") or []:
-                            if "agents" in st:
-                                st["agents"].setdefault(a["pane_id"], a.get("agent_status", "unknown"))
-                            if "agents" in jst:
-                                jst["agents"].setdefault(a["pane_id"], "%s#%s" % (a.get("agent_status", "unknown"), a.get("state_change_seq", "")))
-                    st["core_known"] = True
-                first = "agents" not in st
-                told = agent_events(src, snap, st.setdefault("agents", {}), first)
-                events += told
-                events += watch_agent_events(snap, st.setdefault("watch_agents", {}), watch_panes,
-                                             {e["agent"].get("pane_id") for e in told})
-                learn = "agents" not in jst or first
-                noted += journal_agents(snap, jst.setdefault("agents", {}), learn, now)
-                if live:
-                    agents = live_agents(snap, st, now)
-                st.pop("problem", None)
-            except Exception as e:  # noqa: BLE001
-                if st.get("problem") != str(e):
-                    log("herdr session %s:" % key, e)
-                st["problem"] = str(e)
-        try:
-            if noted:
-                journal_add(key, noted, now)
-            elif "at" not in jst:
-                journal_start(key, now)
-            # Up to when the journal is complete (the watcher may be stopped for a while).
-            jst["at"] = int(now)
-        except Exception as e:  # noqa: BLE001 - the journal is a nicety next to the pushes
-            log("journal:", e)
-        if live:
-            ev = live_event(st, agents, now, first="live_sent" not in st)
-            if ev is not None:
-                events.append(ev)
-        else:
-            for k in ("live_sent", "live_agents", "live_agents_since"):
-                st.pop(k, None)
-        if not events:
-            continue
-        sent[key] = events
-        for sub in group:
-            record = state.setdefault("phones", {}).setdefault(sub["id"], {})
-            if gone(record, sub):
-                continue
-            mine = [e for e in events if (e.get("type") != "live" or sub.get("live")) and sub["id"] in e.get("for", [sub["id"]])]
-            if dry_run or not mine:
+                if noted:
+                    journal_add(key, noted, now)
+                elif "at" not in jst:
+                    journal_start(key, now)
+                # Up to when the journal is complete (the watcher may be stopped for a while).
+                jst["at"] = int(now)
+            except Exception as e:  # noqa: BLE001 - the journal is a nicety next to the pushes
+                log("journal:", e)
+            if live:
+                ev = live_event(st, agents, now, first="live_sent" not in st)
+                if ev is not None:
+                    events.append(ev)
+            else:
+                for k in ("live_sent", "live_agents", "live_agents_since"):
+                    st.pop(k, None)
+            if not events:
                 continue
-            send(sub, mine, record, here)
-    for src, _ in srcs.values():
-        if src.core:
-            src.core.close()
+            sent[key] = events
+            for sub in group:
+                record = state.setdefault("phones", {}).setdefault(sub["id"], {})
+                if gone(record, sub):
+                    continue
+                mine = [e for e in events if (e.get("type") != "live" or sub.get("live")) and sub["id"] in e.get("for", [sub["id"]])]
+                if dry_run or not mine:
+                    continue
+                send(sub, mine, record, here)
+    finally:
+        # Also when a look fails: a herdr-core connection that stays open is a process (Windows) that stays on.
+        for src, _ in srcs.values():
+            if src.core:
+                src.core.close()
     if subs:
         ev = health_look(state.setdefault("health", {}), now)
         if ev is not None:
@@ -1164,12 +1601,27 @@ def tick(state, n, dry_run=False, now=None):
     return sent
 
 
+def lock_byte(f, mode):
+    """Windows: locks or unlocks (msvcrt.LK_NBLCK, LK_UNLCK) the byte of [f] that tells a watcher runs. It is far past the
+    pid, which the locked byte would hide from the other processes. Raises OSError when another process holds it."""
+    import msvcrt
+    f.seek(LOCK_BYTE)
+    msvcrt.locking(f.fileno(), mode, 1)
+
+
 def run():
     os.makedirs(STATE_DIR, exist_ok=True)
+    if WINDOWS and sys.stderr is None:
+        # pythonw.exe has no console, so no stderr either: what log() says goes to the log file.
+        sys.stdout = sys.stderr = open(LOG, "a", buffering=1, encoding="utf-8")
     # Not "w": that would wipe the running watcher's pid before finding out it runs.
     lock = open(LOCK, "a+")
     try:
-        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if WINDOWS:
+            import msvcrt
+            lock_byte(lock, msvcrt.LK_NBLCK)
+        else:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
     except OSError:
         log("already running")
         return 0
@@ -1204,12 +1656,19 @@ def running_pid():
         return None
     with f:
         try:
-            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            if WINDOWS:
+                import msvcrt
+                lock_byte(f, msvcrt.LK_NBLCK)
+            else:
+                fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
         except OSError:
             f.seek(0)
             pid = f.read().strip()
             return int(pid) if pid.isdigit() else -1
-        fcntl.flock(f, fcntl.LOCK_UN)
+        if WINDOWS:
+            lock_byte(f, msvcrt.LK_UNLCK)
+        else:
+            fcntl.flock(f, fcntl.LOCK_UN)
         return None
 
 
@@ -2010,6 +2469,14 @@ def cmd_test(pid=None):
 
 
 def autostart():
+    if WINDOWS:
+        import winreg
+        try:
+            with winreg.OpenKey(winreg.HKEY_CURRENT_USER, RUN_KEY) as key:
+                winreg.QueryValueEx(key, NAME)
+            return "windows-run"
+        except OSError:
+            return "none"
     if sys.platform == "darwin":
         return "launchd" if os.path.exists(launchd_plist()) else "none"
     if os.path.exists(systemd_unit()):
@@ -2032,8 +2499,11 @@ def cmd_status():
                        "ok": r.get("ok"), "error": r.get("error"), "gone": gone(r, s)})
     problems = {k: v["problem"] for k, v in state.get("sessions", {}).items() if v.get("problem")}
     pid = running_pid()
-    out({"ok": True, "version": VERSION, "running": pid is not None, "pid": pid, "autostart": autostart(),
-         "phones": phones, "problems": problems, "path": os.path.abspath(__file__)})
+    info = {"ok": True, "version": VERSION, "running": pid is not None, "pid": pid, "autostart": autostart(),
+            "phones": phones, "problems": problems, "path": os.path.abspath(__file__)}
+    if WINDOWS:
+        info["interactive"] = windows_interactive()  # false: no one is signed in, so there is nowhere to run it
+    out(info)
     return 0
 
 
@@ -2050,6 +2520,18 @@ def start_background():
 
 
 def cmd_ensure():
+    if WINDOWS:
+        # Never a child of this (SSH) session: the task starts it in the signed-in user's own session.
+        if running_pid() is None and windows_interactive():
+            # Make the task (and the Run value, if sign-in autostart is installed) again from this script: after a
+            # Python upgrade or a move they point at a program that is gone, and /run would start nothing. An
+            # `uninstall`ed machine gets no Run value back, as Linux's ensure leaves no unit or cron line.
+            windows_task_create()
+            if autostart() == "windows-run":
+                windows_run_value()
+            windows_task("/run")
+            wait_for_watcher(5)
+        return cmd_status()
     if running_pid() is None:
         kind = autostart()
         if kind == "systemd":
@@ -2077,7 +2559,111 @@ def has_systemd():
         return False
 
 
+def wait_for_watcher(seconds):
+    """Waits up to [seconds] for the watcher to hold the lock; whether it does."""
+    for _ in range(int(seconds * 10)):
+        if running_pid() is not None:
+            return True
+        time.sleep(0.1)
+    return running_pid() is not None
+
+
+def windows_command():
+    """What starts the watcher on Windows: (pythonw.exe, which opens no console window, next to this Python; its arguments)."""
+    py = os.path.join(os.path.dirname(sys.executable), "pythonw.exe")
+    if not os.path.exists(py):
+        py = sys.executable
+    return py, '"%s" run' % os.path.abspath(__file__)
+
+
+def windows_task(*args):
+    """Runs schtasks on this user's task of the watcher: (whether it worked, what it said)."""
+    try:
+        r = subprocess.run(["schtasks"] + list(args) + ["/tn", NAME], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+                           stderr=subprocess.STDOUT, timeout=30, creationflags=NO_WINDOW)
+    except (OSError, subprocess.TimeoutExpired) as e:
+        return False, str(e)
+    return r.returncode == 0, r.stdout.decode(errors="replace").strip()
+
+
+TASK_XML = """<?xml version="1.0" encoding="UTF-16"?>
+<Task version="1.2" xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
+  <Principals>
+    <Principal id="Author">
+      <LogonType>InteractiveToken</LogonType>
+      <RunLevel>LeastPrivilege</RunLevel>
+    </Principal>
+  </Principals>
+  <Settings>
+    <MultipleInstancesPolicy>IgnoreNew</MultipleInstancesPolicy>
+    <DisallowStartIfOnBatteries>false</DisallowStartIfOnBatteries>
+    <StopIfGoingOnBatteries>false</StopIfGoingOnBatteries>
+    <AllowStartOnDemand>true</AllowStartOnDemand>
+    <Enabled>true</Enabled>
+    <Hidden>false</Hidden>
+    <ExecutionTimeLimit>PT0S</ExecutionTimeLimit>
+  </Settings>
+  <Actions Context="Author">
+    <Exec>
+      <Command>%s</Command>
+      <Arguments>%s</Arguments>
+    </Exec>
+  </Actions>
+</Task>
+"""
+
+
+def windows_task_create():
+    """The task that starts the watcher in the user's interactive session (InteractiveToken: only while they are signed
+    in, no password, no elevation). It is what `schtasks /run` starts from an SSH session, which can't open a window
+    there. No trigger (the Run value starts it at sign-in), and none of the limits a task has by default: it would stop
+    the watcher after 3 days, or on battery."""
+    from xml.sax.saxutils import escape
+    py, args = windows_command()
+    fd, path = tempfile.mkstemp(suffix=".xml")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-16") as f:  # schtasks reads a UTF-16 file
+            f.write(TASK_XML % (escape(py), escape(args)))
+        return windows_task("/create", "/xml", path, "/f")
+    finally:
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+def windows_run_value():
+    """Writes the Run value that starts the watcher at sign-in."""
+    import winreg
+    with winreg.CreateKey(winreg.HKEY_CURRENT_USER, RUN_KEY) as key:
+        winreg.SetValueEx(key, NAME, 0, winreg.REG_SZ, '"%s" %s' % windows_command())
+
+
+def windows_install():
+    windows_run_value()  # first: sign-in autostart doesn't depend on the task
+    ok, err = windows_task_create()
+    note = ""
+    if not ok:
+        note = "couldn't create the task (%s): it starts with the next sign-in" % " ".join(err.split())
+    elif windows_interactive():
+        if running_pid() is not None:
+            stop()  # one of an older install: the task starts the one that is here now
+        windows_task("/run")
+        if not wait_for_watcher(5):
+            # The task may still count the watcher just stopped as running, and then ignores this start: once more.
+            windows_task("/run")
+            wait_for_watcher(5)
+    else:
+        note = "no one is signed in at the PC: it starts with the next sign-in"
+    pid = running_pid()
+    out({"ok": pid is not None or bool(note), "msg": note or ("running" if pid is not None else "the task didn't start it"),
+         "autostart": autostart(), "pid": pid, "version": VERSION})
+    return 0 if pid is not None or note else 1
+
+
 def cmd_install():
+    if WINDOWS:
+        return windows_install()
     me = os.path.abspath(__file__)
     note = ""
     if sys.platform == "darwin":
@@ -2142,7 +2728,25 @@ def stop():
             time.sleep(0.1)
 
 
+def windows_uninstall():
+    import winreg
+    try:
+        with winreg.OpenKey(winreg.HKEY_CURRENT_USER, RUN_KEY, 0, winreg.KEY_SET_VALUE) as key:
+            winreg.DeleteValue(key, NAME)
+    except OSError:
+        pass
+    windows_task("/delete", "/f")
+
+
 def cmd_uninstall(purge):
+    if WINDOWS:
+        windows_uninstall()
+        stop()
+        if purge:
+            shutil.rmtree(CONFIG_DIR, ignore_errors=True)
+            shutil.rmtree(STATE_DIR, ignore_errors=True)
+        out({"ok": True, "msg": "removed"})
+        return 0
     if sys.platform == "darwin" and os.path.exists(launchd_plist()):
         subprocess.run(["launchctl", "unload", "-w", launchd_plist()], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
         os.remove(launchd_plist())
@@ -2220,9 +2824,6 @@ LAUNCHD = """<?xml version="1.0" encoding="UTF-8"?>
 def main(argv):
     cmd = argv[1] if len(argv) > 1 else "help"
     rest = argv[2:]
-    if WINDOWS and cmd not in ("pair", "pair-accept", "version", "_qr", "_seal", "help", "-h", "--help"):
-        print("herdr-push %s: Windows runs only `herdr-push pair`; notifications need Linux or macOS." % cmd, file=sys.stderr)
-        return 2
     if WINDOWS and hasattr(sys.stdout, "reconfigure"):
         # A pipe there is in the ANSI code page, which has no block characters for the QR code.
         sys.stdout.reconfigure(encoding="utf-8")