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

refactor(hermes_cli/update): unify git probe/ok helpers in update_cmd_git, dict-dispatch parked-branch reasons, merge print ladders

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:21:29 -07:00
parent 426bac85bf
commit f6f8c1cc2d
1 changed files with 124 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/update_cmd_git.py
+124 -247
View File
@@ -9,19 +9,28 @@ import logging
from contextlib import suppress
import subprocess
import sys
from datetime import datetime
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
# Log-record parity with the origin module.
logger = logging.getLogger("hermes_cli.update_cmd")
_ORPHAN_RESCUE_REFS_TO_KEEP = 10
_ORPHAN_RESCUE_REF_MAX_AGE_DAYS = 30
_GIT_TEXT_KW = dict(capture_output=True, text=True, encoding="utf-8", errors="replace")
_BAR = "=" * 68
def _git_ok(git_cmd, args, cwd, **kw) -> bool:
"""True when ``_git_run`` exits 0; any exception counts as failure."""
from hermes_cli.update_cmd import _git_run
try:
return _git_run(git_cmd, args, cwd, **kw).returncode == 0
except Exception:
return False
def _prune_orphan_rescue_refs(
git_cmd,
@@ -39,27 +48,20 @@ def _prune_orphan_rescue_refs(
"""
from hermes_cli.update_cmd import _git_run
with suppress(OSError):
prefix = f"refs/hermes-update-backups/orphan-{branch}-"
list_result = _git_run(
git_cmd,
["for-each-ref", "--format=%(refname)", "--sort=refname",
f"refs/hermes-update-backups/orphan-{branch}-*"],
cwd,
git_cmd, ["for-each-ref", "--format=%(refname)", "--sort=refname", f"{prefix}*"], cwd,
)
if list_result.returncode != 0:
return
refs = [line.strip() for line in list_result.stdout.splitlines() if line.strip()]
stale = set(refs[:-keep] if keep > 0 else refs)
if max_age_days > 0:
from datetime import timedelta, timezone
cutoff = datetime.now(timezone.utc) - timedelta(days=max_age_days)
prefix = f"refs/hermes-update-backups/orphan-{branch}-"
for ref in refs:
stamp = ref[len(prefix):][:15] # "YYYYMMDD-HHMMSS"
try:
ref_time = datetime.strptime(stamp, "%Y%m%d-%H%M%S").replace(
tzinfo=timezone.utc
)
ref_time = datetime.strptime(stamp, "%Y%m%d-%H%M%S").replace(tzinfo=timezone.utc)
except ValueError:
continue
if ref_time < cutoff:
@@ -77,21 +79,13 @@ def _branch_head_label(git_cmd=None, cwd=None) -> str | None:
try:
cmd = list(git_cmd) if git_cmd else ["git"]
root = cwd if cwd is not None else _m().PROJECT_ROOT
branch = subprocess.run(
cmd + ["rev-parse", "--abbrev-ref", "HEAD"],
cwd=root, capture_output=True,
text=True, encoding="utf-8", errors="replace",
)
sha = subprocess.run(
cmd + ["rev-parse", "--short", "HEAD"],
cwd=root, capture_output=True,
text=True, encoding="utf-8", errors="replace",
)
branch_name = branch.stdout.strip()
sha_text = sha.stdout.strip()
if branch.returncode != 0 or sha.returncode != 0 or not sha_text:
return None
if not branch_name:
def _rev_parse(*args):
return subprocess.run(cmd + ["rev-parse", *args], cwd=root, **_GIT_TEXT_KW)
branch, sha = _rev_parse("--abbrev-ref", "HEAD"), _rev_parse("--short", "HEAD")
branch_name, sha_text = branch.stdout.strip(), sha.stdout.strip()
if branch.returncode != 0 or sha.returncode != 0 or not sha_text or not branch_name:
return None
label = "detached" if branch_name == "HEAD" else branch_name
return f"{label} @ {sha_text}"
@@ -120,11 +114,8 @@ def _assess_parked_branch_switch(
from hermes_cli.update_cmd import _git_run
try:
from hermes_cli.config import load_config
_update_cfg = (load_config() or {}).get("updates", {})
if isinstance(_update_cfg, dict) and not bool(
_update_cfg.get("auto_switch_parked_branch", True)
):
if isinstance(_update_cfg, dict) and not bool(_update_cfg.get("auto_switch_parked_branch", True)):
return False, "disabled"
except Exception as exc:
# Config read failure must not disable the safety checks; fall through with default.
@@ -146,6 +137,12 @@ def _assess_parked_branch_switch(
return True, ""
_PARKED_SKIP_WHY = {
"dirty": "the working tree has uncommitted changes",
"disabled": "updates.auto_switch_parked_branch is set to false in config.yaml",
}
def _print_parked_branch_skip_warning(
git_cmd: list[str],
cwd: Path,
@@ -156,36 +153,24 @@ def _print_parked_branch_skip_warning(
"""LOUD block: why the update was skipped on a parked branch, behind-count, fix commands."""
from hermes_cli.update_cmd import _git_run
behind = None
try:
with suppress(Exception):
behind_result = _git_run(git_cmd, ["rev-list", f"HEAD..origin/{target_branch}", "--count"], cwd)
if behind_result.returncode == 0 and behind_result.stdout.strip():
behind = int(behind_result.stdout.strip())
except Exception:
behind = None
if reason == "dirty":
why = "the working tree has uncommitted changes"
elif reason == "disabled":
why = "updates.auto_switch_parked_branch is set to false in config.yaml"
else:
why = f"the branch state could not be verified against origin/{target_branch}"
bar = "=" * 68
print()
print(bar)
print(f"⚠ CODE UPDATE SKIPPED — checkout is parked on '{current_branch}'")
why = _PARKED_SKIP_WHY.get(
reason, f"the branch state could not be verified against origin/{target_branch}"
)
print(f"\n{_BAR}\n⚠ CODE UPDATE SKIPPED — checkout is parked on '{current_branch}'")
print(f" Not auto-switching to {target_branch}: {why}.")
if behind is not None and behind > 0:
print(
f" This checkout is {behind} commit(s) BEHIND "
f"origin/{target_branch} — the code you are running is stale."
)
print()
print(" To resolve, inspect the branch and switch back yourself:")
print(f" git -C {cwd} status")
print(f" git -C {cwd} checkout {target_branch} && hermes update")
print(" (commit or stash your work on the branch first if you want to keep it)")
print(bar)
print(f" This checkout is {behind} commit(s) BEHIND origin/{target_branch} — the code you are running is stale.")
print(
f"\n To resolve, inspect the branch and switch back yourself:\n"
f" git -C {cwd} status\n"
f" git -C {cwd} checkout {target_branch} && hermes update\n"
f" (commit or stash your work on the branch first if you want to keep it)\n{_BAR}"
)
def _print_parked_branch_kept_notice(
@@ -196,21 +181,14 @@ def _print_parked_branch_kept_notice(
Non-interactive callers can't resolve a skip, so we proceed — but the unmerged work
(still safe on its branch) must be impossible to miss.
"""
bar = "=" * 68
print()
print(bar)
print(
f"\n{_BAR}\n"
f"⚠ Checkout was parked on '{current_branch}' with "
f"{unmerged_count} commit(s) not merged into origin/{target_branch}."
)
print(
f"{unmerged_count} commit(s) not merged into origin/{target_branch}.\n"
f" Switching to {target_branch} so the update can proceed — your "
f"commit(s) are safe on '{current_branch}'."
f"commit(s) are safe on '{current_branch}'.\n\n"
f" To pick the work back up later:\n git checkout {current_branch}\n{_BAR}"
)
print()
print(" To pick the work back up later:")
print(f" git checkout {current_branch}")
print(bar)
OFFICIAL_REPO_URLS = {
@@ -219,11 +197,7 @@ OFFICIAL_REPO_URLS = {
"https://github.com/NousResearch/hermes-agent",
"git@github.com:NousResearch/hermes-agent",
}
OFFICIAL_REPO_URL = "https://github.com/NousResearch/hermes-agent.git"
SKIP_UPSTREAM_PROMPT_FILE = ".skip_upstream_prompt"
@@ -241,36 +215,22 @@ def _is_fork(origin_url: Optional[str]) -> bool:
"""Check if the origin remote points to a fork (not the official repo)."""
if not origin_url:
return False
normalized = origin_url.rstrip("/")
if normalized.endswith(".git"):
normalized = normalized[:-4]
for official in OFFICIAL_REPO_URLS:
official_normalized = official.rstrip("/")
if official_normalized.endswith(".git"):
official_normalized = official_normalized[:-4]
if normalized == official_normalized:
return False
return True
def _norm(url: str) -> str:
url = url.rstrip("/")
return url[:-4] if url.endswith(".git") else url
return _norm(origin_url) not in {_norm(official) for official in OFFICIAL_REPO_URLS}
def _has_upstream_remote(git_cmd: list[str], cwd: Path) -> bool:
"""Check if an 'upstream' remote already exists."""
from hermes_cli.update_cmd import _git_run
try:
result = _git_run(git_cmd, ["remote", "get-url", "upstream"], cwd)
return result.returncode == 0
except Exception:
return False
return _git_ok(git_cmd, ["remote", "get-url", "upstream"], cwd)
def _add_upstream_remote(git_cmd: list[str], cwd: Path) -> bool:
"""Add the official repo as the 'upstream' remote. Returns True on success."""
from hermes_cli.update_cmd import _git_run
try:
result = _git_run(git_cmd, ["remote", "add", "upstream", OFFICIAL_REPO_URL], cwd)
return result.returncode == 0
except Exception:
return False
return _git_ok(git_cmd, ["remote", "add", "upstream", OFFICIAL_REPO_URL], cwd)
def _count_commits_between(git_cmd: list[str], cwd: Path, base: str, head: str) -> int:
@@ -300,12 +260,7 @@ def _mark_skip_upstream_prompt():
def _sync_fork_with_upstream(git_cmd: list[str], cwd: Path) -> bool:
"""Push updated main to origin (sync fork); True on success."""
from hermes_cli.update_cmd import _git_run
try:
result = _git_run(git_cmd, ["push", "origin", "main", "--force-with-lease"], cwd, network=True)
return result.returncode == 0
except Exception:
return False
return _git_ok(git_cmd, ["push", "origin", "main", "--force-with-lease"], cwd, network=True)
def _sync_with_upstream_if_needed(
@@ -329,69 +284,45 @@ def _sync_with_upstream_if_needed(
_no_prompt_git_kwargs,
_should_skip_upstream_prompt,
)
has_upstream = _has_upstream_remote(git_cmd, cwd)
if not has_upstream:
if not _has_upstream_remote(git_cmd, cwd):
if _should_skip_upstream_prompt():
return False
print()
print("ℹ Your fork is not tracking the official Hermes repository.")
print(" This means you may miss updates from NousResearch/hermes-agent.")
print()
if assume_yes or (
input_fn is None and not (sys.stdin.isatty() and sys.stdout.isatty())
):
print(
"\nℹ Your fork is not tracking the official Hermes repository.\n"
" This means you may miss updates from NousResearch/hermes-agent.\n"
)
if assume_yes or (input_fn is None and not (sys.stdin.isatty() and sys.stdout.isatty())):
# --yes means "don't block", not "mutate my remotes"; don't persist the decline.
print(" Skipping upstream setup (non-interactive run).")
print(
" Add it later with: git remote add upstream https://github.com/NousResearch/hermes-agent.git"
)
print(" Add it later with: git remote add upstream https://github.com/NousResearch/hermes-agent.git")
return False
if input_fn is not None:
response = (
input_fn("Add official repo as 'upstream' remote? [y/N]", "n")
.strip()
.lower()
)
response = input_fn("Add official repo as 'upstream' remote? [y/N]", "n").strip().lower()
else:
try:
response = (
input("Add official repo as 'upstream' remote? [Y/n]: ")
.strip()
.lower()
)
response = input("Add official repo as 'upstream' remote? [Y/n]: ").strip().lower()
except (EOFError, KeyboardInterrupt, UnicodeDecodeError):
print()
response = "n"
if response in {"", "y", "yes"}:
print("→ Adding upstream remote...")
if _add_upstream_remote(git_cmd, cwd):
print(" ✓ Added upstream: https://github.com/NousResearch/hermes-agent.git")
has_upstream = True
else:
print(" ✗ Failed to add upstream remote. Skipping upstream sync.")
return False
else:
print(
" Skipped. Run 'git remote add upstream https://github.com/NousResearch/hermes-agent.git' to add later."
)
if response not in {"", "y", "yes"}:
print(" Skipped. Run 'git remote add upstream https://github.com/NousResearch/hermes-agent.git' to add later.")
_mark_skip_upstream_prompt()
return False
print("→ Adding upstream remote...")
if not _add_upstream_remote(git_cmd, cwd):
print(" ✗ Failed to add upstream remote. Skipping upstream sync.")
return False
print(" ✓ Added upstream: https://github.com/NousResearch/hermes-agent.git")
# Only upstream/main: a bare fetch drags in thousands of auto-generated branches.
print()
print("→ Fetching upstream...")
print("\n→ Fetching upstream...")
try:
subprocess.run(
git_cmd + ["fetch", "upstream", "main", "--quiet"],
cwd=cwd,
capture_output=True,
check=True,
**_no_prompt_git_kwargs(),
cwd=cwd, capture_output=True, check=True, **_no_prompt_git_kwargs(),
)
except subprocess.CalledProcessError:
print(" ✗ Failed to fetch upstream. Skipping upstream sync.")
@@ -405,40 +336,35 @@ def _sync_with_upstream_if_needed(
return False
if origin_ahead > 0:
print()
print(f"ℹ Your fork has {origin_ahead} commit(s) not on upstream.")
print(" Skipping upstream sync to preserve your changes.")
print(" If you want to merge upstream changes, run:")
print(" git pull upstream main")
print(
f"\nℹ Your fork has {origin_ahead} commit(s) not on upstream.\n"
" Skipping upstream sync to preserve your changes.\n"
" If you want to merge upstream changes, run:\n git pull upstream main"
)
return True
if upstream_ahead == 0:
print(" ✓ Fork is up to date with upstream")
return True
print()
print(f"→ Fork is {upstream_ahead} commit(s) behind upstream")
print("→ Pulling from upstream...")
print(f"\n→ Fork is {upstream_ahead} commit(s) behind upstream\n→ Pulling from upstream...")
try:
subprocess.run(
git_cmd + ["pull", "--ff-only", "upstream", "main"],
cwd=cwd,
check=True,
**_no_prompt_git_kwargs(),
cwd=cwd, check=True, **_no_prompt_git_kwargs(),
)
except subprocess.CalledProcessError:
print(" ✗ Failed to pull from upstream. You may need to resolve conflicts manually.")
return False
print(" ✓ Updated from upstream")
print("→ Syncing fork...")
print(" ✓ Updated from upstream\n→ Syncing fork...")
if _sync_fork_with_upstream(git_cmd, cwd):
print(" ✓ Fork synced with upstream")
else:
print(" ℹ Got updates from upstream but couldn't push to fork (no write access?)")
print(" Your local repo is updated, but your fork on GitHub may be behind.")
print(
" ℹ Got updates from upstream but couldn't push to fork (no write access?)\n"
" Your local repo is updated, but your fork on GitHub may be behind."
)
return True
@@ -450,30 +376,20 @@ def _classify_fetch_failure(stderr: str) -> str:
"""
def _has_http_code(*codes: str) -> bool:
return any(
f"HTTP {code}" in stderr or f"returned error: {code}" in stderr
for code in codes
)
return any(f"HTTP {code}" in stderr or f"returned error: {code}" in stderr for code in codes)
if _has_http_code("429") or "rate limit" in stderr.lower():
return (
"✗ GitHub is rate limiting requests or having an outage (HTTP 429)"
" — try again in 5 minutes."
)
return "✗ GitHub is rate limiting requests or having an outage (HTTP 429) — try again in 5 minutes."
if _has_http_code("500", "502", "503", "504"):
return (
"✗ GitHub appears to be having an outage — try again in a few"
" minutes (https://www.githubstatus.com)."
)
return "✗ GitHub appears to be having an outage — try again in a few minutes (https://www.githubstatus.com)."
if "Could not resolve host" in stderr or "unable to access" in stderr:
return "✗ Network error — cannot reach the remote repository."
if "could not read Username" in stderr or "terminal prompts disabled" in stderr:
# Anonymous fetch got HTTP 401: GitHub does this during outages (and for
# renamed/private repos) — not a user credentials problem.
return (
"✗ GitHub rejected the anonymous fetch (asked for a login) — this"
" usually means a GitHub outage; try again in a few minutes"
" (https://www.githubstatus.com). If it persists, check"
"✗ GitHub rejected the anonymous fetch (asked for a login) — this usually means a GitHub outage;"
" try again in a few minutes (https://www.githubstatus.com). If it persists, check"
" `git remote -v` points at a public repo."
)
if "Authentication failed" in stderr:
@@ -489,6 +405,15 @@ def _print_fetch_failure(stderr: str) -> None:
print(f" {stderr.splitlines()[0]}")
def _probe_fork_bomb(argv: list) -> Optional[bool]:
"""Run ``<argv> --version``; True/False = guard message seen/absent, None = probe itself failed."""
try:
result = subprocess.run(argv + ["--version"], timeout=15, **_GIT_TEXT_KW)
except Exception:
return None
return "fork bomb" in ((result.stdout or "") + (result.stderr or "")).lower()
def _git_is_trampoline(git_cmd: list) -> bool:
"""Whether *git_cmd* is a broken Git-for-Windows trampoline shim.
@@ -496,17 +421,7 @@ def _git_is_trampoline(git_cmd: list) -> bool:
it every call dies with the launcher's guard message (a PATH problem, not network).
Never raises; unknown states report False so a probe failure can't block an update.
"""
try:
result = subprocess.run(
git_cmd + ["--version"],
capture_output=True,
text=True, encoding="utf-8", errors="replace",
timeout=15,
)
except Exception:
return False
output = ((result.stdout or "") + (result.stderr or "")).lower()
return "fork bomb" in output
return _probe_fork_bomb(git_cmd) is True
def _portable_git_candidates() -> list:
@@ -529,21 +444,9 @@ def _locate_real_git() -> Optional[Path]:
Path(r"C:\Program Files (x86)\Git\mingw64\libexec\git-core\git.exe"),
] + _portable_git_candidates()
for candidate in candidates:
if not candidate.exists():
continue
try:
result = subprocess.run(
[str(candidate), "--version"],
capture_output=True,
text=True, encoding="utf-8", errors="replace",
timeout=15,
)
except Exception:
continue
output = ((result.stdout or "") + (result.stderr or "")).lower()
if "fork bomb" in output:
continue
return candidate
# A failed probe (None) disqualifies the candidate just like a guard hit.
if candidate.exists() and _probe_fork_bomb([str(candidate)]) is False:
return candidate
return None
@@ -552,9 +455,7 @@ def _ensure_non_trampoline_git(git_cmd: list) -> list:
keep working; if none is found leave the command untouched (fetch-failure handler falls
back to ZIP). No-op off Windows and when git is healthy."""
from hermes_cli.update_cmd import _locate_real_git
if sys.platform != "win32":
return git_cmd
if not _git_is_trampoline(git_cmd):
if sys.platform != "win32" or not _git_is_trampoline(git_cmd):
return git_cmd
real_git = _locate_real_git()
if real_git is None:
@@ -576,16 +477,11 @@ def _discard_lockfile_churn(git_cmd, repo_root):
diff = _git_run(git_cmd, ["diff", "--name-only"], repo_root)
if diff.returncode != 0:
return
dirty_package_dirs = {
Path(line.strip()).parent
for line in diff.stdout.splitlines()
if line.strip().endswith("package.json")
}
changed = [line.strip() for line in diff.stdout.splitlines()]
dirty_package_dirs = {Path(p).parent for p in changed if p.endswith("package.json")}
dirty = [
line.strip()
for line in diff.stdout.splitlines()
if line.strip().endswith("package-lock.json")
and Path(line.strip()).parent not in dirty_package_dirs
p for p in changed
if p.endswith("package-lock.json") and Path(p).parent not in dirty_package_dirs
]
if not dirty:
return
@@ -607,13 +503,11 @@ def _normalize_managed_eol(git_cmd, repo_root):
# -c, not config: evaluate the tree as it WOULD look pinned, persisting nothing.
probe = git_cmd + ["-c", "core.autocrlf=false"]
def _dirty(*extra):
out = subprocess.run(
probe + ["diff", "-z", "--name-only", *extra],
cwd=repo_root,
capture_output=True,
text=True, encoding="utf-8", errors="replace",
)
def _probe_run(*args, **kw):
return subprocess.run(probe + list(args), cwd=repo_root, **_GIT_TEXT_KW, **kw)
def _dirty():
out = _probe_run("diff", "-z", "--name-only")
if out.returncode != 0:
return None
return {p for p in out.stdout.split("\0") if p}
@@ -621,24 +515,15 @@ def _normalize_managed_eol(git_cmd, repo_root):
def _real_dirty():
# Files with a *content* change ignoring CRLF. ``--name-only --ignore-cr-at-eol``
# still LISTS CR-only files; ``--numstat`` honors the filter (no record for them).
out = subprocess.run(
probe + ["-c", "core.quotepath=false",
"diff", "--numstat", "--ignore-cr-at-eol"],
cwd=repo_root,
capture_output=True,
text=True, encoding="utf-8", errors="replace",
)
out = _probe_run("-c", "core.quotepath=false", "diff", "--numstat", "--ignore-cr-at-eol")
if out.returncode != 0:
return None
paths = set()
for line in out.stdout.splitlines():
if not line.strip():
continue
# "<added>\t<deleted>\t<path>"; rename detection off, so exactly one path.
parts = line.split("\t", 2)
if len(parts) == 3 and parts[2]:
paths.add(parts[2])
return paths
# "<added>\t<deleted>\t<path>"; rename detection off, so exactly one path.
return {
parts[2]
for parts in (line.split("\t", 2) for line in out.stdout.splitlines() if line.strip())
if len(parts) == 3 and parts[2]
}
def _eol_only():
all_dirty, real_dirty = _dirty(), _real_dirty()
@@ -657,14 +542,9 @@ def _normalize_managed_eol(git_cmd, repo_root):
return
if eol_only:
# Pathspec via stdin: thousands of paths exceed the Windows argv limit.
subprocess.run(
probe
+ ["checkout", "--pathspec-from-file=-", "--pathspec-file-nul", "--"],
cwd=repo_root,
input="\0".join(sorted(eol_only)),
capture_output=True,
text=True, encoding="utf-8", errors="replace",
check=False,
_probe_run(
"checkout", "--pathspec-from-file=-", "--pathspec-file-nul", "--",
input="\0".join(sorted(eol_only)), check=False,
)
if _eol_only():
# Still dirty: pinning would only surface churn we failed to clear.
@@ -672,8 +552,5 @@ def _normalize_managed_eol(git_cmd, repo_root):
print(f"→ Normalized line-ending churn ({len(eol_only)} file(s))")
subprocess.run(
git_cmd + ["config", "core.autocrlf", "false"],
cwd=repo_root,
capture_output=True,
check=False,
git_cmd + ["config", "core.autocrlf", "false"], cwd=repo_root, capture_output=True, check=False,
)