mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
6560a973e0
A suspended or restricted account's token (still printed by `gh auth token`) is refused with 403, not 401, so the update check never retried anonymously and the Updates screen said "We couldn't reach the update server" while the same endpoint answered 200 anonymously (#134335). A 403 that is a spent rate limit (x-ratelimit-remaining: 0) still fails as before.
616 lines
30 KiB
Python
616 lines
30 KiB
Python
"""Passive source update decisions for one exact installation and profile.
|
|
|
|
No fetch, lock repair, or Git writes. Desktop, CLI, and dashboard share this owner.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import quote
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from hermes_constants import get_hermes_home
|
|
from hermes_cli.source_releases import OFFICIAL_REPOSITORY, _GITHUB_ORIGIN, resolve_source_target
|
|
|
|
logger = logging.getLogger(__name__)
|
|
UPDATE_AVAILABLE_NO_COUNT = -1
|
|
_UPDATE_CHECK_CACHE_SECONDS = 24 * 3600
|
|
_UPDATE_CHECK_FAILURE_CACHE_SECONDS = 3600
|
|
_RUNTIME_LOOKUP_SECONDS = 8.0
|
|
|
|
|
|
def _quiet(fn, default=None):
|
|
try:
|
|
return fn()
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def source_git_env() -> dict[str, str]:
|
|
"""Keep read-only Git probes in their explicit cwd, not an inherited worktree.
|
|
|
|
No probe may lazy-fetch from a partial clone's promisor remote: asking about
|
|
an upstream tip the clone never fetched would download its history, and the
|
|
probe timeout kills only git itself, orphaning the fetch (see NO_LAZY_FETCH_ENV).
|
|
"""
|
|
from hermes_cli._subprocess_compat import NO_LAZY_FETCH_ENV, noninteractive_git_env
|
|
|
|
env = noninteractive_git_env()
|
|
# Pathspec-mode overrides change how every probe's path arguments match
|
|
# (GIT_LITERAL_PATHSPECS turns a ":(literal)" magic prefix into filename text).
|
|
for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_COMMON_DIR", "GIT_INDEX_FILE",
|
|
"GIT_OBJECT_DIRECTORY", "GIT_ALTERNATE_OBJECT_DIRECTORIES", "GIT_SHALLOW_FILE", "GIT_NAMESPACE",
|
|
"GIT_LITERAL_PATHSPECS", "GIT_GLOB_PATHSPECS", "GIT_NOGLOB_PATHSPECS", "GIT_ICASE_PATHSPECS"):
|
|
env.pop(key, None)
|
|
env["GIT_OPTIONAL_LOCKS"] = "0"
|
|
env.update(NO_LAZY_FETCH_ENV)
|
|
return env
|
|
|
|
|
|
_GIT_TEXT_KW = {"text": True, "encoding": "utf-8", "errors": "replace"}
|
|
|
|
|
|
def _git_run(args: list[str], *, cwd: Optional[Path] = None, timeout: int = 5, text: bool = True,
|
|
git: str = "git"):
|
|
"""Read Git state without prompts, optional index writes, or inherited targeting."""
|
|
from hermes_cli._subprocess_compat import windows_hide_flags
|
|
|
|
kwargs: dict = {"creationflags": windows_hide_flags(), "env": source_git_env(), "stdin": subprocess.DEVNULL}
|
|
try:
|
|
return subprocess.run(
|
|
[git, *args], capture_output=True, timeout=timeout, cwd=str(cwd) if cwd is not None else None,
|
|
**(_GIT_TEXT_KW if text else {}), **kwargs, check=False)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _git_stdout(args: list[str], *, cwd: Path, timeout: int = 5, git: str = "git") -> Optional[str]:
|
|
result = _git_run(args, cwd=cwd, timeout=timeout, git=git)
|
|
if result is None or result.returncode != 0:
|
|
return None
|
|
return (result.stdout or "").strip()
|
|
|
|
|
|
def _git_ok(args: list[str], **kw) -> bool:
|
|
"""True when ``git <args>`` ran and exited 0 (output discarded)."""
|
|
result = _git_run(args, text=False, **kw)
|
|
return result is not None and result.returncode == 0
|
|
|
|
|
|
def _git_count(args: list[str], *, cwd: Path) -> Optional[int]:
|
|
"""``int`` of a successful ``git rev-list --count``-style command, else None."""
|
|
result = _git_run(args, cwd=cwd)
|
|
if result is not None and result.returncode == 0:
|
|
return _quiet(lambda: int(result.stdout.strip()))
|
|
return None
|
|
|
|
|
|
def _is_full_sha(value: Optional[str]) -> bool:
|
|
return isinstance(value, str) and len(value) == 40 and all(c in "0123456789abcdefABCDEF" for c in value)
|
|
|
|
|
|
def _github_compare(current_rev: str, target_rev: str, repository: str = OFFICIAL_REPOSITORY) -> Optional[dict]:
|
|
# Do not memoize this separately: force must bypass failed AND successful network results.
|
|
if not (_is_full_sha(current_rev) and _is_full_sha(target_rev)):
|
|
return None
|
|
payload = _quiet(lambda: json.loads(_request(
|
|
f"https://api.github.com/repos/{repository}/compare/{current_rev}...{target_rev}")))
|
|
return payload if isinstance(payload, dict) else None
|
|
|
|
|
|
def _github_compare_behind(current_rev: str, target_rev: str, repository: str = OFFICIAL_REPOSITORY) -> Optional[int]:
|
|
payload = _github_compare(current_rev, target_rev, repository)
|
|
ahead = payload.get("ahead_by") if payload else None
|
|
return ahead if isinstance(ahead, int) and not isinstance(ahead, bool) and ahead >= 0 else None
|
|
|
|
|
|
def _request(url: str, accept: str = "application/vnd.github+json") -> str:
|
|
"""GET an api.github.com resource with the credential ladder in hermes_cli.github_api.
|
|
|
|
A token GitHub rejects drops this request to anonymous rather than failing the
|
|
check on a stale credential: 401, or a 403 that is not a spent rate limit (a
|
|
suspended or restricted account's token, still printed by ``gh auth token``, #134335).
|
|
"""
|
|
from hermes_cli.github_api import github_token
|
|
|
|
token = github_token()
|
|
try:
|
|
return _request_with(url, accept, token)
|
|
except urllib.error.HTTPError as exc:
|
|
spent = exc.headers is not None and exc.headers.get("x-ratelimit-remaining") == "0"
|
|
if token is None or exc.code not in (401, 403) or spent:
|
|
exc.hermes_authenticated = token is not None # which quota a 403/429 spent
|
|
raise
|
|
logger.debug("GitHub rejected the configured token; retrying anonymously")
|
|
try:
|
|
return _request_with(url, accept, None)
|
|
except urllib.error.HTTPError as exc:
|
|
exc.hermes_authenticated = False
|
|
raise
|
|
|
|
|
|
def _request_with(url: str, accept: str, token: str | None) -> str:
|
|
headers = {"Accept": accept, "User-Agent": "hermes-update-check"}
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
req = urllib.request.Request(url, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
return response.read(2 * 1024 * 1024).decode("utf-8-sig").strip()
|
|
|
|
|
|
def _branch_tip(repository: str | None, branch: str, root: Path, git: str,
|
|
remote: str = "origin") -> tuple[str | None, bool, str | None]:
|
|
"""``(sha, missing, failure)``: ``missing`` only on a confirmed empty advertisement;
|
|
``failure`` names why no tip could be read, for the user-facing message."""
|
|
# A successful empty ref advertisement alone proves a branch was deleted.
|
|
# GitHub 404 can also mean a private repository: it must not heal a branch.
|
|
failure = None
|
|
if repository:
|
|
from hermes_cli.github_api import describe_github_failure, github_token
|
|
try:
|
|
sha = _request(f"https://api.github.com/repos/{repository}/commits/{quote(branch, safe='')}",
|
|
"application/vnd.github.sha")
|
|
except Exception as exc:
|
|
sha = None
|
|
failure = describe_github_failure(exc, authenticated=github_token() is not None)
|
|
if _is_full_sha(sha):
|
|
return sha, False, None
|
|
if failure is None:
|
|
failure = "api.github.com returned no commit for the branch."
|
|
if branch == "main" and remote == "origin":
|
|
return None, False, failure
|
|
result = _git_run(["ls-remote", "--exit-code", "--heads", remote, f"refs/heads/{branch}"],
|
|
cwd=root, git=git, timeout=10)
|
|
if result is None:
|
|
return None, False, failure or f"`git ls-remote {remote}` could not run."
|
|
sha = result.stdout.split()[0] if result.returncode == 0 and result.stdout else None
|
|
if _is_full_sha(sha):
|
|
return sha, False, None
|
|
if result.returncode == 2:
|
|
return None, True, None
|
|
detail = (result.stderr or "").strip().splitlines()
|
|
return None, False, failure or (f"`git ls-remote {remote}` failed: {detail[-1]}" if detail
|
|
else f"`git ls-remote {remote}` returned no tip.")
|
|
|
|
|
|
def _commits(payload: dict | None) -> list[dict]:
|
|
from datetime import datetime
|
|
rows = []
|
|
entries = (payload or {}).get("commits", [])
|
|
for entry in entries if isinstance(entries, list) else []:
|
|
if not isinstance(entry, dict) or not isinstance(entry.get("sha"), str):
|
|
continue
|
|
commit = entry.get("commit") or {}
|
|
when = (commit.get("committer") or {}).get("date") or ""
|
|
at = _quiet(lambda: int(datetime.fromisoformat(when).timestamp() * 1000), 0)
|
|
rows.append({"sha": entry["sha"], "summary": str(commit.get("message", "")).split("\n", 1)[0],
|
|
"author": str((commit.get("author") or {}).get("name", "")), "at": at})
|
|
return rows[::-1]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Checkout:
|
|
"""Read-only facts about the checkout under test, gathered once per check."""
|
|
root: Path
|
|
git: str
|
|
embedded: Optional[str]
|
|
head: Optional[str]
|
|
current_branch: Optional[str]
|
|
origin: str
|
|
repository: Optional[str]
|
|
dirty: bool
|
|
|
|
|
|
def _read_json(path: Path):
|
|
return _quiet(lambda: json.loads(path.read_text(encoding="utf-8-sig")))
|
|
|
|
|
|
def _unsupported_reason(stamp: dict, root: Path, *, explicit_root: bool, embedded: Optional[str]) -> Optional[dict]:
|
|
"""Fields explaining why this install cannot self-update from Git, or None when it can."""
|
|
from hermes_cli.config import detect_install_method
|
|
from hermes_cli.update_contract import COMMIT_BUILD_UPDATE_MESSAGE
|
|
|
|
if stamp.get("source") == "commit-build":
|
|
return {"reason": "commit-build", "message": COMMIT_BUILD_UPDATE_MESSAGE}
|
|
if stamp.get("payload") in {"bundled", "light", "runtime"} or (
|
|
not explicit_root and detect_install_method(root) in {"docker", "apt"}):
|
|
return {"reason": "not-a-git-checkout"}
|
|
if not embedded and not (root / ".git").exists():
|
|
return {"reason": "not-a-git-checkout", "message": "This install has no git checkout to update."}
|
|
if stamp.get("updateMechanism") not in (None, "self") and not embedded:
|
|
return {"reason": "update-root-steward-owned-git-tree",
|
|
"message": "This installation is managed by its install method; use its updater.", "advice": "git pull"}
|
|
return None
|
|
|
|
|
|
def _installed_git(git: str) -> str:
|
|
"""``git`` as given, unless it is a bare ``git`` PATH lacks: then the git ``hermes update`` runs.
|
|
|
|
A Windows source install's only git is often the copy install.ps1 staged in PM's store, which
|
|
is on PATH only inside ``hermes update`` (``expose_pm_git``). Desktop and the dashboard pass a
|
|
bare ``git`` there, it dies with WinError 2, and the check reports ``head-unavailable``, which
|
|
Desktop shows as "couldn't reach the update server" while ``hermes update`` works.
|
|
"""
|
|
if git != "git" or shutil.which(git):
|
|
return git
|
|
from hermes_cli.update_pause_record import _git_executable
|
|
|
|
return _git_executable()
|
|
|
|
|
|
def _read_checkout(root: Path, git: str, embedded: Optional[str]) -> _Checkout:
|
|
# An embedded revision has no checkout to ask: it always tracks the official repository.
|
|
head = embedded or _git_stdout(["rev-parse", "HEAD"], cwd=root, git=git)
|
|
current_branch = None if embedded else _git_stdout(["rev-parse", "--abbrev-ref", "HEAD"], cwd=root, git=git)
|
|
origin = "" if embedded else (_git_stdout(["remote", "get-url", "origin"], cwd=root, git=git) or "")
|
|
match = _GITHUB_ORIGIN.fullmatch(origin)
|
|
repository = OFFICIAL_REPOSITORY if embedded else (match[1] if match else None)
|
|
dirty = False if embedded else bool(_git_stdout(["status", "--porcelain"], cwd=root, git=git))
|
|
return _Checkout(root, git, embedded, head, current_branch, origin, repository, dirty)
|
|
|
|
|
|
def _configured_branch(desktop_config) -> Optional[str]:
|
|
value = desktop_config.get("branch") if isinstance(desktop_config, dict) else None
|
|
return (value.strip() or None) if isinstance(value, str) else None
|
|
|
|
|
|
def _checked_out_branch(current_branch: Optional[str], fallback: Optional[str]) -> Optional[str]:
|
|
"""The checkout's branch, or ``fallback`` when detached or unreadable."""
|
|
return current_branch if current_branch and current_branch != "HEAD" else fallback
|
|
|
|
|
|
def _cached_status(cache_file: Path, identity: dict, now: float) -> Optional[dict]:
|
|
"""A still-fresh supported status cached for exactly this identity, else None.
|
|
|
|
Failures expire sooner so a transient network error does not hide updates for a day.
|
|
"""
|
|
cached = _read_json(cache_file)
|
|
if not (isinstance(cached, dict) and cached.get("identity") == identity
|
|
and isinstance(cached.get("status"), dict) and cached["status"].get("supported") is True):
|
|
return None
|
|
status = cached.get("status", {})
|
|
ttl = _UPDATE_CHECK_FAILURE_CACHE_SECONDS if status.get("error") else _UPDATE_CHECK_CACHE_SECONDS
|
|
ts = cached.get("ts")
|
|
return status if isinstance(ts, (float, int)) and 0 <= now - ts < ttl else None
|
|
|
|
|
|
def _write_cache(cache_file: Path, identity: dict, now: float, result: dict) -> None:
|
|
try:
|
|
from utils import atomic_json_write
|
|
cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
atomic_json_write(cache_file, {"identity": identity, "ts": now, "status": result})
|
|
except OSError as exc:
|
|
logger.debug("Could not cache source check: %s", exc)
|
|
|
|
|
|
def _read_runtime_verdict(repository: str, sha: str) -> Optional[bool]:
|
|
"""Read the tree and at most four schema blobs at once, within one lookup deadline."""
|
|
from contextvars import copy_context
|
|
from fnmatch import fnmatchcase
|
|
from itertools import islice
|
|
from queue import Empty, Queue
|
|
from threading import Thread
|
|
|
|
from hermes_cli.update_downgrade_guard import STATE_MODULES, defines_runtime
|
|
|
|
deadline = time.monotonic() + _RUNTIME_LOOKUP_SECONDS
|
|
replies: Queue[Optional[str]] = Queue()
|
|
|
|
def launch(url: str, *, api: bool = False) -> None:
|
|
def read() -> None:
|
|
replies.put(_quiet(lambda: _request(url) if api else _request_with(url, "*/*", None)))
|
|
|
|
# A socket timeout alone does not bound a response that keeps trickling bytes. Daemon
|
|
# workers let --version return at the deadline without waiting for stalled downloads.
|
|
Thread(target=copy_context().run, args=(read,), daemon=True).start()
|
|
|
|
def receive() -> Optional[str]:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise Empty
|
|
return replies.get(timeout=remaining)
|
|
|
|
launch(f"https://api.github.com/repos/{repository}/git/trees/{sha}", api=True)
|
|
try:
|
|
tree = _quiet(lambda: json.loads(receive() or ""))
|
|
entries = tree.get("tree") if isinstance(tree, dict) and not tree.get("truncated") else None
|
|
if not isinstance(entries, list):
|
|
return None
|
|
paths = [e["path"] for e in entries if isinstance(e, dict) and e.get("type") == "blob"
|
|
and isinstance(e.get("path"), str) and fnmatchcase(e["path"], STATE_MODULES)]
|
|
if not paths:
|
|
return None # not a recognised Hermes tree: no claim either way
|
|
urls = iter(f"https://raw.githubusercontent.com/{repository}/{sha}/{quote(path)}" for path in paths)
|
|
pending = 0
|
|
for url in islice(urls, 4):
|
|
launch(url)
|
|
pending += 1
|
|
while pending:
|
|
text = receive()
|
|
if text is None:
|
|
return None # an unread module may hold the marker
|
|
if defines_runtime([text]):
|
|
return False
|
|
pending -= 1
|
|
url = next(urls, None)
|
|
if url is not None:
|
|
launch(url)
|
|
pending += 1
|
|
return True
|
|
except Empty:
|
|
return None
|
|
|
|
|
|
def _github_predates_runtime(repository: str, sha: str, cache_file: Path) -> Optional[bool]:
|
|
"""Read an unfetched release without Git writes. Definite verdicts are immutable per SHA;
|
|
unknown verdicts expire so passive cache hits do not repeatedly contact an unreachable GitHub.
|
|
Schema blobs use raw.githubusercontent.com rather than spending the anonymous API quota."""
|
|
known = _read_json(cache_file)
|
|
known = known if isinstance(known, dict) else {}
|
|
cached = known.get(sha)
|
|
if isinstance(cached, bool):
|
|
return cached
|
|
now = time.time()
|
|
retry_at = cached.get("retryAt") if isinstance(cached, dict) else None
|
|
if isinstance(retry_at, (int, float)) and now < retry_at:
|
|
return None
|
|
verdict = _read_runtime_verdict(repository, sha)
|
|
try:
|
|
from utils import atomic_json_write
|
|
cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
atomic_json_write(cache_file, {**known, sha: verdict if verdict is not None else
|
|
{"retryAt": time.time() + _UPDATE_CHECK_FAILURE_CACHE_SECONDS}})
|
|
except OSError as exc:
|
|
logger.debug("Could not cache the release schema verdict: %s", exc)
|
|
return verdict
|
|
|
|
|
|
def _update_refuses(co: _Checkout, target: str, home: Path) -> bool:
|
|
"""The updater's downgrade refusal, asked read-only. An embedded checkout has no git to ask."""
|
|
if co.embedded:
|
|
return False
|
|
from hermes_cli.update_downgrade_guard import would_refuse
|
|
|
|
def git(*args: str) -> int:
|
|
result = _git_run(list(args), cwd=co.root, git=co.git, text=False, timeout=20)
|
|
return -1 if result is None else result.returncode
|
|
|
|
def remote(sha: str) -> Optional[bool]:
|
|
return _github_predates_runtime(co.repository or OFFICIAL_REPOSITORY, sha,
|
|
home / "source-checks" / "release-runtime.json")
|
|
return would_refuse(co.root, target, home, git=git, remote=remote)
|
|
|
|
|
|
def _hold_refused_release(result: dict, co: _Checkout, home: Path) -> None:
|
|
"""Hold an offered release `hermes update` refuses (it predates the gateway runtime these session
|
|
stores use). Covers releases off this checkout's history too, which the ancestry-only rollback
|
|
test cannot see. Apply only after caching the release facts: stores can gain runtime history,
|
|
or the only ledgered profile can be removed, between checks."""
|
|
target = result.get("targetSha")
|
|
if result.get("updateAvailable") and target and _update_refuses(co, target, home):
|
|
result.update(updateAvailable=False, behind=0, held="downgrade", rollback=False)
|
|
|
|
|
|
def _resolve_channel(result: dict, channel: str, co: _Checkout, *, forward_only: bool = False):
|
|
"""Resolve a release channel's target into ``result``; the SourceTarget, or None on error.
|
|
|
|
A target with a pinned commit is final; one without names a branch to follow instead.
|
|
"""
|
|
try:
|
|
source_target = resolve_source_target(channel, [co.git] if not co.embedded else None, co.root,
|
|
repository=co.repository or OFFICIAL_REPOSITORY,
|
|
forward_only=forward_only)
|
|
except (OSError, ValueError, subprocess.SubprocessError) as exc:
|
|
result.update(error="release-unavailable", message=f"Could not resolve the {channel} source channel: {exc}")
|
|
return None
|
|
if source_target.commit:
|
|
target = source_target.commit
|
|
result.pop("branch", None)
|
|
result.update(channel=channel, targetSha=target, updateAvailable=co.head != target,
|
|
behind=0 if co.head == target else UPDATE_AVAILABLE_NO_COUNT,
|
|
sourceVersion=source_target.version, buildId=source_target.build_id)
|
|
# Forward-only pins the target to this checkout's own HEAD, newer than the release.
|
|
# Always present: Desktop names the release only on an explicit False.
|
|
result["aheadOfRelease"] = bool(getattr(source_target, "ahead", False))
|
|
# A release HEAD already contains moves the checkout back: Desktop names it a rollback,
|
|
# not a new update. Local ancestry only; an unknown relation stays an update.
|
|
result["rollback"] = bool(co.head and co.head != target and not co.embedded and _git_ok(
|
|
["merge-base", "--is-ancestor", target, co.head], cwd=co.root, git=co.git))
|
|
if source_target.retired:
|
|
result["retirement"] = {"destination": source_target.channel, "sourceOnly": True}
|
|
return source_target
|
|
|
|
|
|
def _branch_remote(co: _Checkout, selected_branch: str) -> str:
|
|
official_ssh = (co.repository and co.repository.lower() == OFFICIAL_REPOSITORY.lower()
|
|
and co.origin.lower().startswith(("git@", "ssh://")))
|
|
# The public official repo does not require the user's SSH credentials.
|
|
# Forks must keep their own origin, including its authentication.
|
|
return (f"https://github.com/{OFFICIAL_REPOSITORY}.git"
|
|
if co.embedded or (official_ssh and selected_branch != "main") else "origin")
|
|
|
|
|
|
def _heal_deleted_branch(branch_config_path: Path, desktop_config: dict) -> None:
|
|
"""Point the Desktop branch setting back at main after its branch was deleted upstream."""
|
|
# Do not overwrite a concurrent choice made while the network probe ran.
|
|
if _read_json(branch_config_path) == desktop_config:
|
|
from utils import atomic_json_write
|
|
atomic_json_write(branch_config_path, {**desktop_config, "branch": "main"})
|
|
|
|
|
|
def _unhealable_reason(co: _Checkout, branch: str) -> Optional[str]:
|
|
"""Why a branch the remote does not advertise must keep its pin; None when healing loses nothing.
|
|
|
|
An empty ref advertisement cannot tell "merged and deleted upstream" from "never pushed":
|
|
only a branch that was once published (remote-tracking ref or configured upstream, which
|
|
survives ``fetch --prune``) and whose commits are all in main may be re-pinned to main.
|
|
"""
|
|
if co.embedded:
|
|
return None
|
|
local = f"refs/heads/{branch}"
|
|
if not _git_ok(["rev-parse", "--verify", "--quiet", local], cwd=co.root, git=co.git):
|
|
return None # Nothing in this checkout to abandon.
|
|
if not (_git_ok(["rev-parse", "--verify", "--quiet", f"refs/remotes/origin/{branch}"], cwd=co.root, git=co.git)
|
|
or _git_ok(["config", "--get", f"branch.{branch}.merge"], cwd=co.root, git=co.git)):
|
|
return "never-pushed"
|
|
for base in ("refs/remotes/origin/main", "refs/heads/main"):
|
|
# `git cherry` also treats rebase-merged commits (same patch, new SHA) as merged.
|
|
cherry = _git_run(["cherry", base, local], cwd=co.root, git=co.git, timeout=10)
|
|
if cherry is not None and cherry.returncode == 0:
|
|
return "unmerged" if any(line.startswith("+") for line in cherry.stdout.splitlines()) else None
|
|
return "unmerged" # Unknown merge state keeps the branch.
|
|
|
|
|
|
def _behind_count(co: _Checkout, target: str) -> tuple[int, list[dict]]:
|
|
"""``(behind, commits)`` for ``target``: local ancestry first, then the GitHub compare API."""
|
|
if co.head == target or (not co.embedded and _git_ok(
|
|
["merge-base", "--is-ancestor", target, co.head], cwd=co.root, git=co.git)):
|
|
return 0, []
|
|
if co.repository:
|
|
payload = _github_compare(co.head, target, co.repository)
|
|
ahead = (payload or {}).get("ahead_by")
|
|
if isinstance(ahead, int) and not isinstance(ahead, bool) and ahead >= 0:
|
|
return ahead, (_quiet(lambda: _commits(payload), []) if ahead else [])
|
|
return UPDATE_AVAILABLE_NO_COUNT, []
|
|
|
|
|
|
def _check_branch(result: dict, co: _Checkout, selected_branch: str, *,
|
|
heal: Optional[tuple[Path, dict]]) -> None:
|
|
"""Compare the checkout with ``selected_branch``'s remote tip, falling back to main if it was deleted."""
|
|
result["branch"] = selected_branch
|
|
remote = _branch_remote(co, selected_branch)
|
|
target, missing, failure = _branch_tip(co.repository, selected_branch, co.root, co.git, remote)
|
|
reason = _unhealable_reason(co, selected_branch) if missing and selected_branch != "main" else None
|
|
if reason:
|
|
detail = ("has never been pushed" if reason == "never-pushed"
|
|
else "is gone from the remote but has commits that are not in main")
|
|
result.update(error="branch-local-only", localOnly=True,
|
|
message=f"Branch '{selected_branch}' {detail}; keeping it instead of switching to main.")
|
|
return
|
|
if missing and selected_branch != "main":
|
|
result["branch"] = "main"
|
|
if heal:
|
|
_heal_deleted_branch(*heal)
|
|
target, _, failure = _branch_tip(co.repository, "main", co.root, co.git, remote if co.embedded else "origin")
|
|
if target is None:
|
|
result.update(error="fetch-failed",
|
|
message=f"Could not resolve the remote branch tip: {failure}" if failure
|
|
else "Could not resolve the remote branch tip.")
|
|
return
|
|
behind, commits = _behind_count(co, target)
|
|
result["commits"] = commits
|
|
result.update(targetSha=target, behind=behind, updateAvailable=behind != 0)
|
|
|
|
|
|
def check_for_updates(*, install_root: Path | None = None, home: Path | None = None,
|
|
branch: str | None = None, channel: str | None = None,
|
|
cache_path: Path | None = None, branch_config_path: Path | None = None,
|
|
force: bool = False,
|
|
passive: bool = False, git: str = "git") -> dict:
|
|
"""Return a presentation-ready status. Omitted branch follows the current checkout.
|
|
|
|
Only the default (running installation) may use HERMES_REVISION. An explicit
|
|
target must never inherit the host process's embedded revision or stamp.
|
|
"""
|
|
from hermes_cli.config import get_project_root, require_readable_config_before_write
|
|
from hermes_cli.steward import read_install_stamp
|
|
from hermes_cli.update_channel import channel_record, install_id, resolve_update_channel, rides_default_channel
|
|
from hermes_cli.release_channels import validate_name
|
|
|
|
embedded = (os.environ.get("HERMES_REVISION") or None) if install_root is None else None
|
|
root = Path(install_root if install_root is not None else get_project_root()).resolve()
|
|
home = Path(home if home is not None else get_hermes_home()).resolve()
|
|
result = {"supported": False, "hermesRoot": str(root), "behind": None, "commits": []}
|
|
unsupported = _unsupported_reason(read_install_stamp(root), root,
|
|
explicit_root=install_root is not None, embedded=embedded)
|
|
if unsupported:
|
|
return {**result, **unsupported}
|
|
config = require_readable_config_before_write(home / "config.yaml")
|
|
if passive and (config.get("updates") or {}).get("check") is False:
|
|
return {**result, "reason": "disabled"}
|
|
channel = resolve_update_channel(config, root) if channel is None else validate_name(channel)
|
|
forward_only = rides_default_channel(channel_record(config, root), channel, root)
|
|
co = _read_checkout(root, _installed_git(git), embedded)
|
|
desktop_config = _read_json(branch_config_path) if branch_config_path else None
|
|
configured_branch = _configured_branch(desktop_config)
|
|
selected_branch = branch or configured_branch or _checked_out_branch(co.current_branch, "main")
|
|
result.update(supported=True, currentSha=co.head, currentBranch=co.current_branch, dirty=co.dirty)
|
|
if channel != "main":
|
|
result["channel"] = channel
|
|
else:
|
|
result["branch"] = selected_branch
|
|
identity = {"root": str(root), "home": str(home), "head": co.head, "origin": co.origin, "branch": selected_branch,
|
|
"channel": channel, "embedded": embedded, "branchOverride": branch is not None, "channelProtocol": 3,
|
|
"forwardOnly": forward_only}
|
|
cache_file = Path(cache_path) if cache_path is not None else home / "source-checks" / f"{install_id(root)}.json"
|
|
now = time.time()
|
|
cached = None if force else _cached_status(cache_file, identity, now)
|
|
if cached is not None:
|
|
cached = {**cached, "dirty": co.dirty, "currentBranch": co.current_branch}
|
|
if cached.get("channel"):
|
|
_hold_refused_release(cached, co, home)
|
|
return cached
|
|
result["fetchedAt"] = int(now * 1000)
|
|
source_target = None
|
|
if not _is_full_sha(co.head):
|
|
result.update(error="head-unavailable", message="Could not read the installed revision.")
|
|
elif branch is None:
|
|
source_target = _resolve_channel(result, channel, co, forward_only=forward_only)
|
|
if source_target is not None and not source_target.commit:
|
|
# The record supplies a default, not permission to leave the user's branch.
|
|
selected_branch = configured_branch or _checked_out_branch(co.current_branch, source_target.branch)
|
|
if "error" not in result and (source_target is None or source_target.branch is not None):
|
|
# Only a Desktop-configured branch the caller did not override is healed.
|
|
heal = branch_config_path and not branch and configured_branch == selected_branch
|
|
_check_branch(result, co, selected_branch,
|
|
heal=(branch_config_path, desktop_config) if heal else None)
|
|
_write_cache(cache_file, identity, now, result)
|
|
if result.get("channel"):
|
|
_hold_refused_release(result, co, home)
|
|
return result
|
|
|
|
|
|
def main() -> None:
|
|
import argparse
|
|
import contextlib
|
|
import sys
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--install-root", type=Path, required=True)
|
|
parser.add_argument("--home", type=Path, required=True)
|
|
parser.add_argument("--git", default="git")
|
|
parser.add_argument("--branch")
|
|
from hermes_cli.release_channels import validate_name
|
|
parser.add_argument("--channel", type=validate_name)
|
|
parser.add_argument("--cache-path", type=Path)
|
|
parser.add_argument("--branch-config-path", type=Path)
|
|
parser.add_argument("--force", action="store_true")
|
|
parser.add_argument("--set-channel", type=validate_name,
|
|
help="Persist this install's channel (the Desktop selector), then report as usual.")
|
|
args = parser.parse_args()
|
|
with contextlib.redirect_stdout(sys.stderr):
|
|
if args.set_channel:
|
|
from hermes_constants import set_hermes_home_override
|
|
from hermes_cli.update_channel import set_install_channel
|
|
|
|
# The record belongs in the --home profile's config.yaml, as the check reads it.
|
|
set_hermes_home_override(args.home)
|
|
set_install_channel(args.set_channel, args.install_root)
|
|
args.force = True
|
|
result = check_for_updates(**{k: v for k, v in vars(args).items() if k != "set_channel"})
|
|
if isinstance(result, dict):
|
|
# Older runtimes reject --set-channel; Desktop offers its selector only on this flag.
|
|
result["channelSelectable"] = True
|
|
print(json.dumps(result))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|