diff --git a/hermes_cli/update_cmd_git.py b/hermes_cli/update_cmd_git.py index b4bee112ad..03c71ec74f 100644 --- a/hermes_cli/update_cmd_git.py +++ b/hermes_cli/update_cmd_git.py @@ -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 `` --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 - # "\t\t"; 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 + # "\t\t"; 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, )