-
Notifications
You must be signed in to change notification settings - Fork 3k
fix(state): restore Hermes cron scripts before enabling restored jobs #7880
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Closed
Closed
Changes from all commits
Commits
Show all changes
34 commits
Select commit
Hold shift + click to select a range
6462e22
fix(state): restore Hermes cron scripts before enabling restored jobs
laitingsheng 7da56a0
test(state): keep the staged restore test body linear
laitingsheng 617ea5d
fix(state): remove the restore archive copy when publishing fails
laitingsheng 66e47a6
test(state): assert the failed publication attempt in staged restore
laitingsheng dfde003
Merge remote-tracking branch 'origin/main' into fix/hermes-cron-scrip…
laitingsheng a5a1de4
merge: sync Hermes cron restore with main
apurvvkumaria 507a354
merge: sync Hermes cron restore with main
apurvvkumaria bef478a
fix(shields): lock restored Hermes cron scripts
apurvvkumaria 191d7d1
Merge remote-tracking branch 'origin/main' into codex/pr7880-hermes-l…
apurvvkumaria cef29e3
fix(state): roll back failed staged restores
apurvvkumaria 447a759
test(state): keep rollback fixtures linear
apurvvkumaria 446f157
merge(main): sync PR #7880 with current main
apurvvkumaria ae5963b
Merge remote-tracking branch 'origin/main' into codex/pr7880-hermes-l…
apurvvkumaria 25e00ce
fix(state): drain Hermes scheduled work during restore
apurvvkumaria 9ec10b3
test(state): linearize Hermes restore fixtures
apurvvkumaria e198654
test(hermes): cover restore guard image contracts
apurvvkumaria 88f6b0c
merge(main): sync Hermes state restore coverage
apurvvkumaria b52ec4f
Merge branch 'main' into fix/hermes-cron-script-state-restore
senthilr-nv 7824b3f
merge: resolve conflicts with main
github-actions[bot] a5a698f
merge: resolve conflicts with main
github-actions[bot] 6949d3e
fix(state): close the staged restore command builder
laitingsheng 664b91f
merge(main): refresh #7880 branch
laitingsheng 9affeda
fix(state): hold Hermes restore drain ownership atomically
laitingsheng 2056cc6
merge(main): refresh #7880 branch
laitingsheng 2cd6d4e
merge: resolve conflicts with main
github-actions[bot] bc0391d
merge(main): refresh #7880 branch
laitingsheng 6a74ed6
fix(hermes): pin the current restore cron guard digest
laitingsheng af6abcc
merge(main): refresh #7880 branch
laitingsheng 8f029ed
merge(main): refresh #7880 branch
laitingsheng f7cba25
Merge branch 'main' into fix/hermes-cron-script-state-restore
cv a7544ee
fix(hermes): reject cron scripts the gateway cannot reach
laitingsheng 554aa38
merge(main): refresh #7880 branch
laitingsheng 61b91a3
merge(main): refresh #7880 branch
laitingsheng 1e263e6
merge: resolve conflicts with main
github-actions[bot] File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,279 @@ | ||
| #!/opt/hermes/.venv/bin/python -I | ||
| # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
| """Quiesce Hermes cron dispatch while NemoClaw restores scheduled work state.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import argparse | ||
| import grp | ||
| import json | ||
| import os | ||
| import pwd | ||
| import secrets | ||
| import stat | ||
| import sys | ||
| import time | ||
| from pathlib import Path | ||
| from typing import Any | ||
|
|
||
| _OWNER_PREFIX = "nemoclaw-state-restore:" | ||
| _POLL_INTERVAL_SECONDS = 0.1 | ||
| _OWNERSHIP_FILE = ".nemoclaw-restore-drain" | ||
| _GATEWAY_USER = "gateway" | ||
|
|
||
|
|
||
| def _configure_home(raw_home: str) -> Path: | ||
| home = Path(raw_home) | ||
| if not home.is_absolute(): | ||
| raise ValueError("Hermes restore guard requires an absolute --home path") | ||
| os.environ["HERMES_HOME"] = str(home) | ||
| return home | ||
|
|
||
|
|
||
| def _gateway_modules() -> tuple[Any, Any]: | ||
| from gateway import drain_control, status | ||
|
|
||
| return drain_control, status | ||
|
|
||
|
|
||
| def _runtime_is_safely_drained(status: Any, pid: int) -> bool: | ||
| runtime = status.read_runtime_status() | ||
| return bool( | ||
| isinstance(runtime, dict) | ||
| and runtime.get("pid") == pid | ||
| and runtime.get("gateway_state") == "draining" | ||
| and status.parse_active_agents(runtime.get("active_agents")) == 0 | ||
| ) | ||
|
|
||
|
|
||
| def _owned_marker_present(drain_control: Any, home: Path, token: str) -> bool: | ||
| marker = drain_control.read_drain_request(home=home) | ||
| return bool(isinstance(marker, dict) and marker.get("principal") == token) | ||
|
|
||
|
|
||
| def _release_owned_marker(drain_control: Any, home: Path, token: str) -> None: | ||
| if not _owned_marker_present(drain_control, home, token): | ||
| return | ||
| if not drain_control.clear_drain_request(home=home): | ||
| raise RuntimeError("Hermes restore guard could not clear its drain marker") | ||
|
|
||
|
|
||
| def _ownership_path(home: Path) -> Path: | ||
| return home / _OWNERSHIP_FILE | ||
|
|
||
|
|
||
| def _claim_ownership(home: Path, token: str) -> bool: | ||
| try: | ||
| descriptor = os.open( | ||
| _ownership_path(home), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600 | ||
| ) | ||
| except FileExistsError: | ||
| return False | ||
| with os.fdopen(descriptor, "w", encoding="utf-8") as handle: | ||
| handle.write(token) | ||
| return True | ||
|
|
||
|
|
||
| def _release_ownership(home: Path, token: str) -> None: | ||
| path = _ownership_path(home) | ||
| try: | ||
| recorded = path.read_text(encoding="utf-8") | ||
| except FileNotFoundError: | ||
| return | ||
| if recorded == token: | ||
| path.unlink(missing_ok=True) | ||
|
|
||
|
|
||
| def begin_drain(home: Path, timeout_seconds: float) -> str: | ||
| drain_control, status = _gateway_modules() | ||
| token = f"{_OWNER_PREFIX}{secrets.token_hex(16)}" | ||
| if not _claim_ownership(home, token): | ||
| raise RuntimeError("Another NemoClaw restore already owns the Hermes drain") | ||
|
|
||
| owned = False | ||
| try: | ||
| if drain_control.drain_requested(home=home): | ||
| result = "preserved" | ||
| else: | ||
| drain_control.write_drain_request(principal=token, home=home) | ||
| owned = True | ||
| result = token | ||
|
|
||
| deadline = time.monotonic() + timeout_seconds | ||
| while time.monotonic() < deadline: | ||
| live_pid = status.get_running_pid() | ||
| if live_pid is None or _runtime_is_safely_drained(status, live_pid): | ||
| if not owned: | ||
| _release_ownership(home, token) | ||
| return result | ||
| time.sleep(_POLL_INTERVAL_SECONDS) | ||
| except BaseException: | ||
| if owned: | ||
| _release_owned_marker(drain_control, home, token) | ||
| _release_ownership(home, token) | ||
| raise | ||
|
|
||
| if owned: | ||
| _release_owned_marker(drain_control, home, token) | ||
| _release_ownership(home, token) | ||
| raise TimeoutError( | ||
| f"Hermes gateway did not drain active messaging, API, and cron work within {timeout_seconds:g}s" | ||
| ) | ||
|
|
||
|
|
||
| def assert_safely_drained(home: Path) -> None: | ||
| drain_control, status = _gateway_modules() | ||
| pid = status.get_running_pid() | ||
| if pid is None: | ||
| return | ||
| if not drain_control.drain_requested(home=home) or not _runtime_is_safely_drained( | ||
| status, pid | ||
| ): | ||
| raise RuntimeError("Hermes gateway is not safely drained for scheduled-work restore") | ||
|
|
||
|
|
||
| def _gateway_identity() -> tuple[int, set[int]] | None: | ||
| try: | ||
| entry = pwd.getpwnam(_GATEWAY_USER) | ||
| except KeyError: | ||
| return None | ||
| memberships = { | ||
| group.gr_gid for group in grp.getgrall() if _GATEWAY_USER in group.gr_mem | ||
| } | ||
| memberships.add(entry.pw_gid) | ||
| return entry.pw_uid, memberships | ||
|
|
||
|
|
||
| def _accessible_by_gateway(path: Path, access_mode: int, bits: tuple[int, int, int]) -> bool: | ||
| identity = _gateway_identity() | ||
| if identity is None: | ||
| return os.access(path, access_mode) | ||
| uid, gids = identity | ||
| if uid == os.geteuid(): | ||
| return os.access(path, access_mode) | ||
| owner_bit, group_bit, other_bit = bits | ||
| info = path.stat() | ||
| if info.st_uid == uid: | ||
| return bool(info.st_mode & owner_bit) | ||
| if info.st_gid in gids: | ||
| return bool(info.st_mode & group_bit) | ||
| return bool(info.st_mode & other_bit) | ||
|
|
||
|
|
||
| def _readable_by_gateway(script_path: Path) -> bool: | ||
| return _accessible_by_gateway( | ||
| script_path, os.R_OK, (stat.S_IRUSR, stat.S_IRGRP, stat.S_IROTH) | ||
| ) | ||
|
|
||
|
|
||
| def _searchable_by_gateway(directory: Path) -> bool: | ||
| return _accessible_by_gateway( | ||
| directory, os.X_OK, (stat.S_IXUSR, stat.S_IXGRP, stat.S_IXOTH) | ||
| ) | ||
|
|
||
|
|
||
| def _enclosing_directories(scripts_dir: Path, script_parent: Path) -> list[Path]: | ||
| directories = [script_parent] | ||
| while directories[-1] != scripts_dir: | ||
| directories.append(directories[-1].parent) | ||
| directories.reverse() | ||
| return directories | ||
|
|
||
|
|
||
| def _load_jobs(jobs_file: Path) -> list[Any]: | ||
| if not jobs_file.exists(): | ||
| return [] | ||
| data = json.loads(jobs_file.read_text(encoding="utf-8-sig")) | ||
| jobs = data.get("jobs", []) if isinstance(data, dict) else data | ||
| if not isinstance(jobs, list): | ||
| raise ValueError("Hermes cron database must contain a jobs list") | ||
| return jobs | ||
|
|
||
|
|
||
| def validate_enabled_scripts(home: Path) -> None: | ||
| scripts_dir = (home / "scripts").resolve() | ||
| for index, job in enumerate(_load_jobs(home / "cron" / "jobs.json")): | ||
| if not isinstance(job, dict): | ||
| raise ValueError(f"Hermes cron job at index {index} is not an object") | ||
| if not job.get("enabled", True) or job.get("state") == "paused": | ||
| continue | ||
| script = job.get("script") | ||
| if script in {None, ""}: | ||
| if job.get("no_agent"): | ||
| raise ValueError( | ||
| f"Enabled no-agent Hermes cron job at index {index} has no script" | ||
| ) | ||
| continue | ||
| if not isinstance(script, str): | ||
| raise ValueError(f"Enabled Hermes cron job at index {index} has a non-string script") | ||
| raw_path = Path(script).expanduser() | ||
| script_path = ( | ||
| raw_path.resolve() | ||
| if raw_path.is_absolute() | ||
| else (scripts_dir / raw_path).resolve() | ||
| ) | ||
| try: | ||
| script_path.relative_to(scripts_dir) | ||
| except ValueError as error: | ||
| raise ValueError( | ||
| f"Enabled Hermes cron job at index {index} resolves outside the scripts directory" | ||
| ) from error | ||
| if not script_path.is_file() or not _readable_by_gateway(script_path): | ||
| raise ValueError( | ||
| f"Enabled Hermes cron job at index {index} references a missing or unreadable script" | ||
| ) | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| for directory in _enclosing_directories(scripts_dir, script_path.parent): | ||
| if not _searchable_by_gateway(directory): | ||
| raise ValueError( | ||
| f"Enabled Hermes cron job at index {index} references a script the Hermes " | ||
| "gateway cannot reach through its directories" | ||
| ) | ||
|
|
||
|
|
||
| def validate_restore(home: Path) -> None: | ||
| assert_safely_drained(home) | ||
| validate_enabled_scripts(home) | ||
|
|
||
|
|
||
| def release_drain(home: Path, token: str) -> None: | ||
| if not token.startswith(_OWNER_PREFIX) or len(token) != len(_OWNER_PREFIX) + 32: | ||
| raise ValueError("Invalid Hermes restore drain ownership token") | ||
| drain_control, _status = _gateway_modules() | ||
| _release_owned_marker(drain_control, home, token) | ||
| _release_ownership(home, token) | ||
|
|
||
|
|
||
| def _parser() -> argparse.ArgumentParser: | ||
| parser = argparse.ArgumentParser() | ||
| parser.add_argument("action", choices=("begin", "assert-safe", "validate", "release")) | ||
| parser.add_argument("--home", required=True) | ||
| parser.add_argument("--timeout", type=float, default=60.0) | ||
| parser.add_argument("--token") | ||
| return parser | ||
|
|
||
|
|
||
| def main() -> int: | ||
| args = _parser().parse_args() | ||
| try: | ||
| home = _configure_home(args.home) | ||
| if args.action == "begin": | ||
| if args.timeout <= 0: | ||
| raise ValueError("Hermes restore drain timeout must be positive") | ||
| print(begin_drain(home, args.timeout)) | ||
| elif args.action == "assert-safe": | ||
| assert_safely_drained(home) | ||
| elif args.action == "validate": | ||
| validate_restore(home) | ||
| else: | ||
| if not args.token: | ||
| raise ValueError("Hermes restore drain release requires --token") | ||
| release_drain(home, args.token) | ||
| return 0 | ||
| except Exception as error: | ||
| print(f"Hermes restore guard failed: {error}", file=sys.stderr) | ||
| return 1 | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| raise SystemExit(main()) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.