hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(cli): split source_check.check_for_updates into cache, channel, fetch helpers

Browse Source 
check_for_updates had grown to ~CC 75 over 128 lines, mixing install gating,
checkout probing, cache freshness, channel resolution, remote-tip fetch and
result shaping. It now reads as orchestration over cohesive helpers:
_unsupported_reason, _read_checkout (_Checkout facts), _cached_status /
_write_cache, _resolve_channel, _check_branch (_branch_remote,
_heal_deleted_branch, _behind_count). Every function is below CC 30
(check_for_updates 26).

Behaviour, public signature and return shape are unchanged; module-level
seams tests patch (_git_stdout, _git_ok, _branch_tip, resolve_source_target,
time, ...) are still read from this module at call time.
This commit is contained in:
ethernet
2026-09-24 11:38:52 -04:00
parent d18382441f
commit 0cfd42e562
1 changed files with 178 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

272
hermes_cli/source_check.py
View File

@@ -9,6 +9,7 @@ import logging
import os
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
from urllib.parse import quote
@@ -183,6 +184,159 @@ def _commits(payload: dict | None) -> list[dict]:
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 _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 _resolve_channel(result: dict, channel: str, co: _Checkout):
"""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)
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)
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 _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)
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,
@@ -193,124 +347,54 @@ def check_for_updates(*, install_root: Path | None = None, home: Path | None = N
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 detect_install_method, get_project_root, require_readable_config_before_write
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 install_id, resolve_update_channel
from hermes_cli.release_channels import validate_name
from hermes_cli.update_contract import COMMIT_BUILD_UPDATE_MESSAGE
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()
stamp = read_install_stamp(root)
result = {"supported": False, "hermesRoot": str(root), "behind": None, "commits": []}
if stamp.get("source") == "commit-build":
return {**result, "reason": "commit-build", "message": COMMIT_BUILD_UPDATE_MESSAGE}
if stamp.get("payload") in {"bundled", "light", "runtime"} or (install_root is None and detect_install_method(root) in {"docker", "apt"}):
return {**result, "reason": "not-a-git-checkout"}
if not embedded and not (root / ".git").exists():
return {**result, "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 {**result, "reason": "update-root-steward-owned-git-tree",
"message": "This installation is managed by its install method; use its updater.", "advice": "git pull"}
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)
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)
desktop_config = _quiet(lambda: json.loads(branch_config_path.read_text(encoding="utf-8-sig"))) if branch_config_path else None
configured_branch = desktop_config.get("branch") if isinstance(desktop_config, dict) else None
if isinstance(configured_branch, str):
configured_branch = configured_branch.strip() or None
else:
configured_branch = None
selected_branch = branch or configured_branch or (current_branch if current_branch and current_branch != "HEAD" else "main")
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))
result.update(supported=True, currentSha=head, currentBranch=current_branch, dirty=dirty)
co = _read_checkout(root, 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": head, "origin": origin, "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": 1}
cache_file = Path(cache_path) if cache_path is not None else home / "source-checks" / f"{install_id(root)}.json"
cached = _quiet(lambda: json.loads(cache_file.read_text(encoding="utf-8-sig")))
now = time.time()
if (not force and isinstance(cached, dict) and cached.get("identity") == identity
and isinstance(cached.get("status"), dict) and cached["status"].get("supported") is True):
status = cached.get("status", {})
ttl = _UPDATE_CHECK_FAILURE_CACHE_SECONDS if status.get("error") else _UPDATE_CHECK_CACHE_SECONDS
ts = cached.get("ts")
if isinstance(ts, (float, int)) and 0 <= now - ts < ttl:
return {**status, "dirty": dirty, "currentBranch": current_branch}
cached = None if force else _cached_status(cache_file, identity, now)
if cached is not None:
return {**cached, "dirty": co.dirty, "currentBranch": co.current_branch}
result["fetchedAt"] = int(now * 1000)
source_target = None
if not _is_full_sha(head):
if not _is_full_sha(co.head):
result.update(error="head-unavailable", message="Could not read the installed revision.")
elif branch is None:
try:
source_target = resolve_source_target(channel, [git] if not embedded else None, root,
repository=repository or OFFICIAL_REPOSITORY)
except (OSError, ValueError, subprocess.SubprocessError) as exc:
result.update(error="release-unavailable", message=f"Could not resolve the {channel} source channel: {exc}")
else:
if source_target.commit:
target = source_target.commit
result.pop("branch", None)
result.update(channel=channel, targetSha=target, updateAvailable=head != target,
behind=0 if head == target else UPDATE_AVAILABLE_NO_COUNT,
sourceVersion=source_target.version, buildId=source_target.build_id)
if source_target.retired:
result["retirement"] = {"destination": source_target.channel, "sourceOnly": True}
else:
# The record supplies a default, not permission to leave the user's branch.
selected_branch = configured_branch or (
current_branch if current_branch and current_branch != "HEAD" else source_target.branch)
source_target = _resolve_channel(result, channel, co)
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):
result["branch"] = selected_branch
official_ssh = (repository and repository.lower() == OFFICIAL_REPOSITORY.lower()
and 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.
remote = (f"https://github.com/{OFFICIAL_REPOSITORY}.git"
if embedded or (official_ssh and selected_branch != "main") else "origin")
target, missing, failure = _branch_tip(repository, selected_branch, root, git, remote)
if missing and selected_branch != "main":
result["branch"] = "main"
if branch_config_path and not branch and configured_branch == selected_branch:
# Do not overwrite a concurrent choice made while the network probe ran.
current_config = _quiet(lambda: json.loads(branch_config_path.read_text(encoding="utf-8-sig")))
if current_config == desktop_config:
from utils import atomic_json_write
atomic_json_write(branch_config_path, {**desktop_config, "branch": "main"})
target, _, failure = _branch_tip(repository, "main", root, git, remote if 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.")
else:
behind = UPDATE_AVAILABLE_NO_COUNT
if head == target or (not embedded and _git_ok(
["merge-base", "--is-ancestor", target, head], cwd=root, git=git)):
behind = 0
elif repository:
payload = _github_compare(head, target, repository)
ahead = (payload or {}).get("ahead_by")
if isinstance(ahead, int) and not isinstance(ahead, bool) and ahead >= 0:
behind = ahead
result["commits"] = _quiet(lambda: _commits(payload), []) if behind else []
result.update(targetSha=target, behind=behind, updateAvailable=behind != 0)
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)
# 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)
return result