"""Is a newer version published? Notify, never install. Omarchy checks itself for updates (`omarchy-update-available` looks at its own checkout and its package) but nothing checks plugins, so a user can sit on an old Nexthop indefinitely without ever being told. This closes that gap the smallest way it can be closed. **This module never updates anything.** It answers one question — is the installed checkout behind its origin — and the answer becomes a quiet glyph in the panel naming the command the user can run. Updating stays where Omarchy put it: `omarchy plugin update`, which shows the diff and asks. A plugin that fetched and ran new code would be self-modifying code inside a system that deliberately gates updates behind human review, and it would reopen a security review that took four rounds to clear. How it stays cheap and read-only: * `git ls-remote` asks the remote for its HEAD without writing a single byte into the user's checkout — no fetch, no new objects, no refs touched. It takes well under a second and needs no credentials. * Whether we are *behind* rather than merely *different* is then decided from objects we already have, so a developer checkout that is ahead of origin is never nagged. * Egress is to the repository the user installed from and nowhere else. It carries nothing about them or their network. It is still egress, so it is disclosed and `updateCheck` turns it off. The one piece of untrusted input here is the commit id the remote hands back, and it goes on to be an argument to another `git` call — so it is validated against the exact 40-hex shape before it reaches a subprocess, the same doctrine `vet_target()` applies to URLs we did not choose. """ import os import re import shutil import subprocess import threading from pathlib import Path # A git object id and nothing else. This is the guard that matters: the value # arrives from the network and is then passed as an argument to git. RE_SHA = re.compile(r"^[0-9a-f]{40}$") # A release is a rare event, and the check exists to catch the user who would # otherwise never look. Daily is generous. CHECK_INTERVAL_S = 24 * 3600 # Not at startup: the daemon restarts with the shell, and a check on every # restart would be noise for no benefit. Nothing is lost by waiting. FIRST_CHECK_DELAY_S = 300 # The remote may be slow, unreachable, or a captive portal that answers # everything. None of those may stall the daemon. GIT_TIMEOUT_S = 20 # `ls-remote` prints one short line per ref; this is far past HEAD alone. MAX_LS_REMOTE_BYTES = 64 * 1024 def verdict(local: str, remote: str, have_remote: bool, head_before_remote: bool, remote_before_head: bool) -> str: """Where the checkout stands relative to origin. Pure, so it is testable. * `current` — the same commit. * `behind` — origin is strictly ahead of us: an update. * `ahead` — we are strictly ahead of origin: a dev checkout. * `diverged` — neither contains the other. * `unknown` — we could not tell, and say so rather than guessing. Both ancestry directions are needed, and the reason is the likeliest case of all: `omarchy plugin update` fetches before it shows its diff, so a user who looked and said "not now" already *holds* origin's commit while still being behind it. Deciding "behind" from "we have never seen that object" alone would show that user nothing. """ if not local or not remote: return "unknown" if local == remote: return "current" if not have_remote: # We do not hold origin's commit at all, so it is newer than anything # we know about. return "behind" if head_before_remote: return "behind" if remote_before_head: return "ahead" return "diverged" class UpdateWatch: """Asks origin, on a slow cadence, whether this checkout is behind. The result lives in memory only. A daemon restart forgets it and waits `FIRST_CHECK_DELAY_S` before asking again, which is why no fifth state file was added for this. """ def __init__(self, repo: Path = None, enabled: bool = True, spawn=None): # Derived from this file, not from the working directory: the daemon # can be started from anywhere. self.repo = Path(repo) if repo else Path(__file__).resolve().parent.parent self.enabled = enabled self.state = "unknown" self.checked_ts = None self._next = None # How a check is run. Off the loop by default: `ls-remote` may sit # at its 20 s timeout on exactly the flaky network where the # outage watch matters, and the loop must not wait for it. Tests # pass a synchronous spawn so the verdict lands within the tick. self._spawn = spawn or self._in_thread self._lock = threading.Lock() self._inflight = False self._pending = None # a verdict awaiting the next tick @staticmethod def _in_thread(fn): threading.Thread(target=fn, name="update-check", daemon=True).start() def _git(self, *args, capture: bool = True): """One git call. Fixed argv, no shell, bounded, always timed out.""" env = dict(os.environ) # A credential prompt on a private or moved remote would otherwise # block until the timeout every single time. env["GIT_TERMINAL_PROMPT"] = "0" env.pop("GIT_ASKPASS", None) env.pop("SSH_ASKPASS", None) # `git -C ` does not require to BE a repository: git # walks up until it finds one. A copy-install with no .git of its # own, inside a home that is itself a checkout (dotfiles), would # otherwise be compared against — and contact the remote of — the # wrong repository every day. The ceiling stops the walk at our # parent, so a missing .git answers "unknown", never someone else. env["GIT_CEILING_DIRECTORIES"] = str(self.repo.parent) try: proc = subprocess.run( ["git", "-C", str(self.repo), # No credential helper and no askpass program may run for # this: a private or moved remote gets "unknown", not a # keyring prompt once a day. GIT_TERMINAL_PROMPT above # covers only git's own tty prompt. "-c", "credential.helper=", "-c", "core.askPass=", *args], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE if capture else subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=GIT_TIMEOUT_S, env=env, ) except (OSError, subprocess.TimeoutExpired): return None, "" out = "" if capture and proc.stdout: out = proc.stdout[:MAX_LS_REMOTE_BYTES].decode( "utf-8", "replace").strip() return proc.returncode, out def _local_head(self): code, out = self._git("rev-parse", "HEAD") return out if code == 0 and RE_SHA.match(out) else None def _remote_head(self): """Origin's HEAD, or None. Writes nothing into the checkout.""" code, out = self._git("ls-remote", "origin", "HEAD") if code != 0 or not out: return None sha = out.split()[0] if out.split() else "" # The guard: this value came off the network and is about to become a # git argument. Anything but a bare object id is refused outright. return sha if RE_SHA.match(sha) else None def check(self) -> str: """Run one check now and return the verdict. Read-only throughout.""" if not shutil.which("git"): return "unknown" local = self._local_head() if not local: return "unknown" # not a git checkout, or no commits remote = self._remote_head() if not remote: return "unknown" # offline, no origin, or a junk answer if local == remote: return "current" # Do we already hold origin's commit? Only objects we have are # consulted from here on, so no fetch is ever needed. code, _ = self._git("cat-file", "-e", remote + "^{commit}", capture=False) have = code == 0 before, after = False, False if have: # Exit 1 means "not an ancestor", which is an answer, not an error. code, _ = self._git("merge-base", "--is-ancestor", local, remote, capture=False) before = code == 0 code, _ = self._git("merge-base", "--is-ancestor", remote, local, capture=False) after = code == 0 return verdict(local, remote, have, before, after) def _run_check(self): state = self.check() with self._lock: self._pending = state self._inflight = False def _collect(self, now: float): with self._lock: state, self._pending = self._pending, None if state is not None: self.state = state self.checked_ts = round(now) def tick(self, now: float): """Called from the daemon loop; starts a check when one is due and adopts the verdict of one that has finished. Never blocks.""" if not self.enabled: # Turning the setting off clears any standing notice, so the # glyph disappears rather than lingering with a stale answer. # A check already running is left to finish and its answer # dropped here, unread. self.state, self.checked_ts, self._next = "unknown", None, None with self._lock: self._pending = None return self._collect(now) if self._next is None: self._next = now + FIRST_CHECK_DELAY_S return if now < self._next: return self._next = now + CHECK_INTERVAL_S with self._lock: if self._inflight: return self._inflight = True self._spawn(self._run_check) # A synchronous spawn has already finished; adopt it now rather # than a tick later. self._collect(now) def snapshot(self) -> dict: """What the panel reads. None while nothing has been established.""" if not self.enabled or self.state == "unknown": return None return { "state": self.state, "available": self.state == "behind", "checked_ts": self.checked_ts, }