diff --git a/CHANGELOG.md b/CHANGELOG.md index 84aab3e..c905c26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,30 @@ # Changelog +## 2.1.1 + +Read-only diagnostic safety patch. + +### Fixes & Improvements + +1. **Strict diagnostic mode.** `run-once --diagnostic` inspects a stable private SQLite + snapshot without constructing the pipeline, changing HEGI state or queues, creating + dead letters, or calling Telegram, LLM, Memory Forest, archive, or run logging paths. +2. **Public CLI isolation.** Plugin registration starts the embedded worker only inside a + Hermes gateway process, and writable state initialization occurs later inside the + worker lock, so `hermes hegi ... --diagnostic` remains read-only before its handler. + A gateway blocked by a migration-held worker lock retries initialization once per + second instead of leaving approval state uninitialized for a full polling interval. +3. **Fail-closed snapshots.** Streaming snapshots preserve bounded memory, use no-atime + reads on Linux/WSL, pin config and state through directory descriptors, reject pathname + rebinding and unsafe artifacts, and distinguish concurrent changes from invalid state. +4. **Schema verification.** Diagnostics require schema version 8, required table/column + definitions, indexes (including partial idempotency indexes), and foreign keys. +5. **Migration quiescence and post-check.** The migration script stops the selected + gateway and holds both the gateway runtime lock and worker lock before its rollback + snapshot, launches foreground-capable restarts detached, verifies gateway lock + reacquisition, and runs the diagnostic while retaining the worker lock. Documentation + now distinguishes this mode from the legacy stateful `--dry-run` pipeline rehearsal. + ## 2.1.0 First standalone HEGI release. diff --git a/MIGRATION.md b/MIGRATION.md index 92d8c80..b73a1da 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -139,19 +139,29 @@ scripts/migrate-existing-install.sh --apply \ The script performs these changes: -1. Creates `/hegi/migration-backups//` with mode `0700`. -2. Copies the HEGI config. -3. Uses SQLite `.backup` for the state database, when present. -4. Copies the legacy `plugins/hegi-telegram` directory, when present. -5. Records whether `hegi.service` was enabled. -6. Seals the backup with a version marker and SHA-256 manifest covering every artifact. -7. Installs `icerain-cmd/hermes-hegi` through the Hermes plugin manager with `--force`. -8. Enables `hegi-telegram` without a built-in tool override grant. -9. Runs the additive HEGI state migration, creating another config/state checkpoint. -10. Disables the legacy service unless `--keep-old-service` was supplied. -11. Restarts the Hermes gateway. -12. Runs `hermes hegi doctor`. -13. Runs `hermes hegi run-once --dry-run`. +1. Acquires the worker lock, stops the selected profile's Hermes gateway, and holds its + runtime lock, quiescing both worker and legacy hook writes before the rollback snapshot. +2. Creates `/hegi/migration-backups//` with mode `0700`. +3. Copies the HEGI config. +4. Uses SQLite `.backup` for the state database, when present. +5. Copies the legacy `plugins/hegi-telegram` directory, when present. +6. Records whether `hegi.service` was enabled. +7. Seals the backup with a version marker and SHA-256 manifest covering every artifact. +8. Installs `icerain-cmd/hermes-hegi` through the Hermes plugin manager with `--force`. +9. Enables `hegi-telegram` without a built-in tool override grant. +10. Runs the additive HEGI state migration, creating another config/state checkpoint. +11. Disables the legacy service unless `--keep-old-service` was supplied. +12. Releases the gateway runtime lock, launches gateway restart detached (including on + foreground-only WSL installations), and verifies that the gateway reacquired it. +13. Runs `hermes hegi doctor`. +14. Runs the no-state-write post-check `hermes hegi run-once --diagnostic` while the + migration still owns the worker lock. + +If backup creation fails before plugin installation or schema migration begins, the +script makes a best-effort restart of the original gateway. Once an install or schema +mutation has started, a failure deliberately leaves the gateway stopped so an operator +can inspect the completed rollback backup instead of loading a potentially partial +deployment. Save the printed backup directory and rollback command. If a later step fails, the backup from an earlier step still exists. Do not delete it while diagnosing the partial @@ -166,7 +176,7 @@ hermes plugins list --enabled --json hermes gateway status hermes hegi doctor --json hermes hegi status -hermes hegi run-once --dry-run +hermes hegi run-once --diagnostic ``` Confirm: @@ -178,7 +188,7 @@ Confirm: - the config still points to the intended source databases and local archive; - existing episode and delivery counts remain present; - no second standalone worker is active; and -- dead-letter count did not grow unexpectedly. +- dead-letter count did not change during the diagnostic. Inspect the script-created state backup too: @@ -188,8 +198,8 @@ sqlite3 /path/to/migration-backup/state.db "PRAGMA quick_check;" ## 7. Telegram report smoke test -Choose a known, non-sensitive quiet episode and check dry-run output before allowing a -real report: +After the read-only migration verification is complete, choose a known, non-sensitive +quiet episode for an explicitly stateful pipeline rehearsal before allowing a real report: ```bash hermes hegi run-once --dry-run diff --git a/README.md b/README.md index 4927ffd..586f6c3 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# HEGI 2.1.0: AI Research Secretary for Hermes +# HEGI 2.1.1: AI Research Secretary for Hermes [한국어 문서](docs/README.ko.md) @@ -7,7 +7,7 @@ structured minutes, action items, revision-safe archives, Telegram reports, and Forest recommendations. It is distributed as `hermes-hegi`, imported as `hegi`, and registered with Hermes under the plugin key `hegi-telegram`. -HEGI 2.1.0 is a standalone plugin. It does not patch Hermes, copy code into the Hermes +HEGI 2.1.1 is a standalone plugin. It does not patch Hermes, copy code into the Hermes installation, or depend on private Hermes modules. The package uses the public plugin context for CLI registration, `pre_gateway_dispatch`, LLM access, and tool dispatch. It calls the Telegram Bot API through its own adapter. @@ -133,14 +133,14 @@ Install a tagged source tree or a downloaded release wheel into the same Python environment as Hermes, then enable the discovered entry point: ```bash -python -m pip install "git+https://github.com/icerain-cmd/hermes-hegi.git@v2.1.0" +python -m pip install "git+https://github.com/icerain-cmd/hermes-hegi.git@v2.1.1" hermes plugins enable hegi-telegram --no-allow-tool-override ``` or: ```bash -python -m pip install ./hermes_hegi-2.1.0-py3-none-any.whl +python -m pip install ./hermes_hegi-2.1.1-py3-none-any.whl hermes plugins enable hegi-telegram --no-allow-tool-override ``` @@ -190,13 +190,24 @@ hermes gateway restart hermes hegi doctor hermes hegi doctor --json hermes hegi status +hermes hegi run-once --diagnostic hermes hegi run-once --dry-run ``` -`run-once` defaults to dry-run behavior. It may use the configured LLM and Memory Forest -search tools and may update HEGI's local state, but it does not write an archive, send a -Telegram report, consume the source range, approve a Draft, or commit a memory. Use -`--send` only when you intend to archive and deliver a real report: +`run-once --diagnostic` streams a stable raw SQLite snapshot into private temporary +storage, checks configuration, the supported schema version and required invariants, +integrity, and aggregate queue +counts there, and verifies that the operating database and sidecars did not change. It +never constructs the pipeline or calls the LLM, Memory Forest, Telegram, archive, or run +logger. On Linux and WSL it opens SQLite artifacts with `O_NOATIME`; configuration and env +files are read, so their filesystem access timestamps are outside this logical no-write +guarantee. + +The existing `run-once --dry-run` remains a stateful pipeline simulation for backward +compatibility. It may use the configured LLM and Memory Forest search tools and may +update HEGI's local state, retries, dead letters, cursors, buffers, and run log. It does +not write an archive, send a Telegram report, consume the source range, approve a Draft, +or commit a memory. Use `--send` only when you intend to archive and deliver a real report: ```bash hermes hegi run-once --send @@ -213,7 +224,8 @@ The CLI surface is: | `hermes hegi setup [--dry-run|--apply]` | Discover and write HEGI configuration | | `hermes hegi doctor [--json]` | Check Hermes APIs, configuration, tools, token, paths, and schema | | `hermes hegi status` | Print episode, queue, notification, and dead-letter counts | -| `hermes hegi run-once [--dry-run|--send]` | Run one pipeline cycle; dry-run is the default | +| `hermes hegi run-once --diagnostic` | Inspect a stable private state snapshot with no HEGI data writes or external calls | +| `hermes hegi run-once [--dry-run|--send]` | Run a stateful pipeline cycle; dry-run is the default | | `hermes hegi migrate [--dry-run|--apply]` | Back up config and state, then apply additive schema changes | | `hermes hegi export-diagnostics [--output FILE]` | Write a redacted diagnostic JSON file with mode `0600` | | `hermes hegi uninstall --keep-data` | Disable HEGI while retaining config, state, and archives | @@ -233,11 +245,11 @@ hermes hegi migrate --dry-run hermes hegi migrate --apply hermes gateway restart hermes hegi doctor -hermes hegi run-once --dry-run +hermes hegi run-once --diagnostic ``` For a Python installation, install the intended wheel or tag with `python -m pip install ---upgrade ...`, then run the same migrate, restart, doctor, and dry-run sequence. Schema +--upgrade ...`, then run the same migrate, restart, doctor, and diagnostic sequence. Schema migrations are additive. Do not downgrade or restore `state.db` without a migration-specific rollback plan. diff --git a/docs/README.ko.md b/docs/README.ko.md index 20df244..3d75e47 100644 --- a/docs/README.ko.md +++ b/docs/README.ko.md @@ -1,4 +1,4 @@ -# HEGI 2.1.0: Hermes용 AI 연구비서 +# HEGI 2.1.1: Hermes용 AI 연구비서 [English README](../README.md) @@ -6,7 +6,7 @@ HEGI는 여러 Hermes 연구 에이전트의 대화를 회의 Episode, 구조화 revision-safe archive, Telegram 보고, Memory Forest 권고로 변환한다. Python 배포명은 `hermes-hegi`, import package는 `hegi`, Hermes 플러그인 키는 `hegi-telegram`이다. -HEGI 2.1.0은 독립 플러그인이다. Hermes를 패치하거나 Hermes 설치 디렉터리에 코드를 +HEGI 2.1.1은 독립 플러그인이다. Hermes를 패치하거나 Hermes 설치 디렉터리에 코드를 복사하지 않으며 Hermes private module에 의존하지 않는다. 이 패키지는 CLI 등록, `pre_gateway_dispatch`, LLM 접근, tool dispatch에 public plugin context를 사용한다. Telegram Bot API는 HEGI 자체 adapter로 호출한다. @@ -133,14 +133,14 @@ tag가 지정된 source tree나 다운로드한 release wheel을 Hermes와 같 environment에 설치한 다음 발견된 entry point를 활성화한다. ```bash -python -m pip install "git+https://github.com/icerain-cmd/hermes-hegi.git@v2.1.0" +python -m pip install "git+https://github.com/icerain-cmd/hermes-hegi.git@v2.1.1" hermes plugins enable hegi-telegram --no-allow-tool-override ``` 또는 다음을 실행한다. ```bash -python -m pip install ./hermes_hegi-2.1.0-py3-none-any.whl +python -m pip install ./hermes_hegi-2.1.1-py3-none-any.whl hermes plugins enable hegi-telegram --no-allow-tool-override ``` @@ -190,13 +190,21 @@ hermes gateway restart hermes hegi doctor hermes hegi doctor --json hermes hegi status +hermes hegi run-once --diagnostic hermes hegi run-once --dry-run ``` -`run-once`는 기본적으로 dry-run으로 동작한다. 설정된 LLM과 Memory Forest search tool을 -사용하고 HEGI local state를 갱신할 수 있지만 archive를 쓰거나 Telegram 보고를 보내지 -않고, source range를 consumed로 전환하지 않으며, Draft approve나 memory commit을 -수행하지 않는다. 실제 archive와 보고 전송을 의도할 때만 `--send`를 사용한다. +`run-once --diagnostic`은 운영 SQLite를 private 임시 공간으로 streaming snapshot한 뒤 +지원 schema, 무결성, aggregate queue count를 검사한다. HEGI state나 queue를 쓰지 않고 +pipeline, LLM, Memory Forest, Telegram, archive, run logger를 호출하지 않는다. Linux와 +WSL에서는 SQLite artifact를 `O_NOATIME`으로 연다. config와 env 파일은 읽으므로 그 +파일들의 filesystem access timestamp는 논리적 무쓰기 보장 범위에 포함하지 않는다. + +기존 `run-once --dry-run`은 호환성을 위해 남겨 둔 stateful pipeline simulation이다. +설정된 LLM과 Memory Forest search tool을 사용하고 retry, dead letter, cursor, buffer 등 +HEGI local state를 갱신할 수 있다. archive를 쓰거나 Telegram 보고를 보내지 않고, +source range를 consumed로 전환하지 않으며, Draft approve나 memory commit을 수행하지 +않는다. 실제 archive와 보고 전송을 의도할 때만 `--send`를 사용한다. ```bash hermes hegi run-once --send @@ -213,6 +221,7 @@ CLI surface는 다음과 같다. | `hermes hegi setup [--dry-run|--apply]` | HEGI config 탐지 및 작성 | | `hermes hegi doctor [--json]` | Hermes API, config, tool, token, path, schema 점검 | | `hermes hegi status` | episode, queue, notification, dead-letter count 출력 | +| `hermes hegi run-once --diagnostic` | HEGI data write와 external call 없이 private state snapshot 점검 | | `hermes hegi run-once [--dry-run|--send]` | pipeline cycle 한 번 실행. 기본값은 dry-run | | `hermes hegi migrate [--dry-run|--apply]` | config와 state를 백업하고 additive schema 변경 적용 | | `hermes hegi export-diagnostics [--output FILE]` | mode `0600`인 redacted diagnostic JSON 작성 | @@ -233,11 +242,11 @@ hermes hegi migrate --dry-run hermes hegi migrate --apply hermes gateway restart hermes hegi doctor -hermes hegi run-once --dry-run +hermes hegi run-once --diagnostic ``` Python 설치에서는 의도한 wheel 또는 tag를 `python -m pip install --upgrade ...`로 -설치한 다음 같은 migrate, restart, doctor, dry-run 순서를 실행한다. schema migration은 +설치한 다음 같은 migrate, restart, doctor, diagnostic 순서를 실행한다. schema migration은 additive 방식이다. migration별 rollback 계획 없이 `state.db`를 downgrade하거나 복원하지 않는다. diff --git a/docs/operations.md b/docs/operations.md index 5d1e057..3bff601 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -1,6 +1,6 @@ # Operations -This guide covers a new HEGI 2.1.0 installation. Operators moving an in-tree or standalone +This guide covers a new HEGI 2.1.1 installation. Operators moving an in-tree or standalone daemon deployment should use [MIGRATION.md](../MIGRATION.md) first. ## Prerequisites @@ -33,7 +33,7 @@ For a wheel installation, install into the Python environment that provides `her enable the package entry point: ```bash -python -m pip install ./hermes_hegi-2.1.0-py3-none-any.whl +python -m pip install ./hermes_hegi-2.1.1-py3-none-any.whl hermes plugins enable hegi-telegram --no-allow-tool-override ``` @@ -144,15 +144,35 @@ and must be disabled before the embedded worker handles production data. ## One-cycle checks -Run a dry cycle after setup, upgrade, gateway restart, or state migration: +Run the strict read-only diagnostic after setup, upgrade, gateway restart, or state +migration: + +```bash +hermes hegi run-once --diagnostic +``` + +The diagnostic streams a stable raw main/WAL snapshot into a private temporary directory. +It reads configuration, the supported schema version and required invariants, integrity, +and aggregate queue state +from that copy, then verifies that the operating database and its sidecars are unchanged. +On Linux and WSL, SQLite artifacts are opened with `O_NOATIME`; config and env access +timestamps are outside the logical no-write guarantee. It does not construct the +pipeline, call the LLM or Memory Forest, contact Telegram, write an archive, or append the +run log. If another worker changes SQLite files during the check, it exits with +`hegi-state-changing`; an unsafe or unreadable artifact instead exits with +`hegi-state-unsafe`. Stop a concurrent worker before retrying a changing state; do not +treat an unsafe artifact as transient. + +Use the older stateful simulation only when retry, cursor, buffer, episode, dead-letter, +and run-log updates are acceptable: ```bash hermes hegi run-once --dry-run ``` Dry-run may read source databases, call the configured LLM, search Memory Forest, and -write HEGI state needed for diagnostics. It does not write the meeting archive, deliver a -Telegram report, consume the source range, approve a Draft, or commit a memory. +write HEGI state. It does not write the meeting archive, deliver a Telegram report, +consume the source range, approve a Draft, or commit a memory. Send a real report only after the dry-run result and target chat are confirmed: @@ -271,7 +291,7 @@ hermes hegi migrate --dry-run hermes hegi migrate --apply hermes gateway restart hermes hegi doctor -hermes hegi run-once --dry-run +hermes hegi run-once --diagnostic ``` For a package installation, upgrade to an explicit tag or wheel and continue with the diff --git a/hegi/_version.py b/hegi/_version.py index 4bcfdcb..af41481 100644 --- a/hegi/_version.py +++ b/hegi/_version.py @@ -1,3 +1,3 @@ """Single source of truth for the HEGI package version.""" -__version__ = "2.1.0" +__version__ = "2.1.1" diff --git a/hegi/cli.py b/hegi/cli.py index af2ea28..8785b5a 100644 --- a/hegi/cli.py +++ b/hegi/cli.py @@ -4,22 +4,32 @@ import argparse import copy +import hashlib import json import os import sqlite3 +import stat import sys import tempfile import time from pathlib import Path from typing import Any +import yaml + from ._version import __version__ from .bootstrap import _atomic_yaml, _backup, setup from .compat import Check, doctor_checks -from .config import default_config_path, load_config, state_path_is_bound, validate_config +from .config import ( + _config_from_mapping, + default_config_path, + load_config, + state_path_is_bound, + validate_config, +) from .locking import run_with_process_lock from .pipeline import HegiPipeline -from .state import StateStore +from .state import SCHEMA, StateStore from .yamlio import load_yaml_mapping @@ -27,6 +37,14 @@ class CLIUsageError(ValueError): """A deliberately public, path-free CLI usage message.""" +class _DiagnosticBusy(RuntimeError): + """The operating SQLite files changed while a read-only snapshot was taken.""" + + +class _DiagnosticArtifactError(RuntimeError): + """An operating SQLite artifact cannot be inspected safely.""" + + def _json(value: Any) -> None: print(json.dumps(value, ensure_ascii=False, indent=2)) @@ -151,7 +169,626 @@ def cmd_retry_failed(args: argparse.Namespace) -> int: return 0 +_STATE_ARTIFACT_SUFFIXES = ("", "-wal", "-shm", "-journal") +_DIAGNOSTIC_SCHEMA_VERSION = 8 +_DIAGNOSTIC_REQUIRED_TABLES = frozenset( + { + "collector_cursor", + "processed_episodes", + "message_buffer", + "report_delivery", + "dead_letter", + "approval_events", + "approval_jobs", + "approval_transitions", + "notification_jobs", + "action_items", + } +) + + +def _artifact_metadata(metadata: os.stat_result) -> tuple[Any, ...]: + return ( + metadata.st_dev, + metadata.st_ino, + stat.S_IMODE(metadata.st_mode), + metadata.st_nlink, + metadata.st_size, + metadata.st_atime_ns, + metadata.st_mtime_ns, + metadata.st_ctime_ns, + ) + + +def _open_state_directory(state_path: Path) -> int | None: + parent = Path(os.path.abspath(state_path.expanduser())).parent + if os.name != "posix" or os.open not in os.supports_dir_fd: + try: + resolved = parent.resolve(strict=True) + except OSError as exc: + raise _DiagnosticArtifactError("state directory is unavailable") from exc + if parent != resolved: + raise _DiagnosticArtifactError("state directory cannot contain symbolic links") + return None + flags = os.O_RDONLY + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + if hasattr(os, "O_DIRECTORY"): + flags |= os.O_DIRECTORY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor: int | None = None + try: + descriptor = os.open(parent.anchor, flags) + for component in parent.parts[1:]: + next_descriptor = os.open(component, flags, dir_fd=descriptor) + opened = os.fstat(next_descriptor) + if not stat.S_ISDIR(opened.st_mode): + os.close(next_descriptor) + raise _DiagnosticArtifactError("state path contains a non-directory component") + os.close(descriptor) + descriptor = next_descriptor + return descriptor + except _DiagnosticArtifactError: + if descriptor is not None: + os.close(descriptor) + raise + except OSError as exc: + if descriptor is not None: + os.close(descriptor) + raise _DiagnosticArtifactError("state directory cannot be opened safely") from exc + + +def _verify_state_directory_binding(state_path: Path, directory_fd: int | None) -> None: + if directory_fd is None: + parent = Path(os.path.abspath(state_path.expanduser())).parent + try: + if parent != parent.resolve(strict=True): + raise _DiagnosticArtifactError("state directory cannot contain symbolic links") + except OSError as exc: + raise _DiagnosticArtifactError("state directory is unavailable") from exc + return + current_descriptor = _open_state_directory(state_path) + if current_descriptor is None: + raise _DiagnosticArtifactError("state directory cannot be rebound safely") + try: + pinned = os.fstat(directory_fd) + current = os.fstat(current_descriptor) + finally: + os.close(current_descriptor) + if pinned.st_dev != current.st_dev or pinned.st_ino != current.st_ino: + raise _DiagnosticBusy("state directory path changed during diagnostic read") + + +def _open_regular_artifact(path: Path, directory_fd: int | None) -> tuple[int, os.stat_result]: + try: + lexical = ( + os.stat(path.name, dir_fd=directory_fd, follow_symlinks=False) + if directory_fd is not None + else path.lstat() + ) + except FileNotFoundError: + raise + except OSError as exc: + raise _DiagnosticArtifactError("state artifact metadata is unavailable") from exc + if not stat.S_ISREG(lexical.st_mode) or lexical.st_nlink != 1: + raise _DiagnosticArtifactError("state artifacts must be regular single-link files") + flags = os.O_RDONLY + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + if hasattr(os, "O_NOATIME"): + flags |= os.O_NOATIME + try: + descriptor = ( + os.open(path.name, flags, dir_fd=directory_fd) + if directory_fd is not None + else os.open(path, flags) + ) + except FileNotFoundError: + raise + except OSError as exc: + raise _DiagnosticArtifactError("state artifact cannot be opened safely") from exc + try: + opened = os.fstat(descriptor) + except OSError as exc: + os.close(descriptor) + raise _DiagnosticArtifactError("state artifact metadata is unavailable") from exc + if ( + not stat.S_ISREG(opened.st_mode) + or opened.st_nlink != 1 + or opened.st_dev != lexical.st_dev + or opened.st_ino != lexical.st_ino + ): + os.close(descriptor) + raise _DiagnosticArtifactError("state artifact changed during validation") + return descriptor, opened + + +def _read_small_regular_artifact( + path: Path, + directory_fd: int | None, + *, + maximum_bytes: int = 1024 * 1024, +) -> tuple[bytes, tuple[Any, ...]]: + descriptor, before = _open_regular_artifact(path, directory_fd) + chunks: list[bytes] = [] + total = 0 + try: + while True: + chunk = os.read(descriptor, min(64 * 1024, maximum_bytes + 1 - total)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + if total > maximum_bytes: + raise _DiagnosticArtifactError("diagnostic config exceeds the safe size limit") + after = os.fstat(descriptor) + except _DiagnosticArtifactError: + raise + except OSError as exc: + raise _DiagnosticArtifactError("diagnostic config cannot be read safely") from exc + finally: + os.close(descriptor) + if _artifact_metadata(before) != _artifact_metadata(after) or total != after.st_size: + raise _DiagnosticBusy("diagnostic config changed while it was read") + contents = b"".join(chunks) + return contents, (*_artifact_metadata(after), hashlib.sha256(contents).hexdigest()) + + +def _load_diagnostic_config( + path: str | Path, +) -> tuple[Any, int | None, tuple[Any, ...] | None]: + config_path = Path(os.path.abspath(Path(path).expanduser())) + directory_fd = _open_state_directory(config_path.parent / "state.db") + if directory_fd is None: + return load_config(config_path), None, None + try: + raw_bytes, config_fingerprint = _read_small_regular_artifact( + config_path, directory_fd + ) + loaded = yaml.safe_load(raw_bytes.decode("utf-8")) + if loaded is None: + loaded = {} + if not isinstance(loaded, dict): + raise _DiagnosticArtifactError("diagnostic config root must be a mapping") + config = _config_from_mapping( + config_path, + loaded, + home=config_path.parent.parent, + ) + return config, directory_fd, config_fingerprint + except Exception: + os.close(directory_fd) + raise + + +def _verify_config_binding( + config_path: Path, + directory_fd: int | None, + expected: tuple[Any, ...] | None, +) -> None: + if directory_fd is None or expected is None: + return + try: + current = _stream_regular_artifact(config_path, directory_fd) + except FileNotFoundError as exc: + raise _DiagnosticArtifactError("diagnostic config is unavailable") from exc + if current != expected: + raise _DiagnosticBusy("diagnostic config changed during diagnostic read") + + +def _stream_regular_artifact( + path: Path, + directory_fd: int | None, + destination: Path | None = None, +) -> tuple[Any, ...]: + descriptor, before = _open_regular_artifact(path, directory_fd) + destination_descriptor: int | None = None + destination_complete = False + digest = hashlib.sha256() + total = 0 + try: + if destination is not None: + destination_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_CLOEXEC"): + destination_flags |= os.O_CLOEXEC + destination_descriptor = os.open(destination, destination_flags, 0o600) + while True: + chunk = os.read(descriptor, 1024 * 1024) + if not chunk: + break + total += len(chunk) + digest.update(chunk) + if destination_descriptor is not None: + pending = memoryview(chunk) + while pending: + written = os.write(destination_descriptor, pending) + if written <= 0: + raise OSError("diagnostic snapshot write did not make progress") + pending = pending[written:] + if destination_descriptor is not None: + os.fsync(destination_descriptor) + after = os.fstat(descriptor) + if _artifact_metadata(before) != _artifact_metadata(after) or total != after.st_size: + raise _DiagnosticBusy("state artifact changed during diagnostic read") + destination_complete = True + return (*_artifact_metadata(after), digest.hexdigest()) + except _DiagnosticBusy: + raise + except OSError as exc: + raise _DiagnosticArtifactError("diagnostic snapshot could not be created safely") from exc + finally: + os.close(descriptor) + if destination_descriptor is not None: + os.close(destination_descriptor) + if destination is not None and not destination_complete: + destination.unlink(missing_ok=True) + + +def _capture_state_artifacts( + state_path: Path, + directory_fd: int | None, +) -> dict[str, tuple[Any, ...]]: + fingerprints: dict[str, tuple[Any, ...]] = {} + for suffix in _STATE_ARTIFACT_SUFFIXES: + candidate = Path(f"{state_path}{suffix}") + try: + fingerprint = _stream_regular_artifact(candidate, directory_fd) + except FileNotFoundError: + continue + fingerprints[suffix] = fingerprint + if "" not in fingerprints: + raise _DiagnosticArtifactError("HEGI state database does not exist") + return fingerprints + + +def _copy_stable_state_snapshot( + state_path: Path, + directory_fd: int | None, + destination: Path, +) -> dict[str, tuple[Any, ...]]: + for _attempt in range(3): + destination.unlink(missing_ok=True) + Path(f"{destination}-wal").unlink(missing_ok=True) + try: + before = _capture_state_artifacts(state_path, directory_fd) + if "-journal" in before: + raise _DiagnosticBusy("state rollback journal is active") + copied: dict[str, tuple[Any, ...]] = {} + for suffix in before: + source = Path(f"{state_path}{suffix}") + target = ( + destination + if suffix == "" + else Path(f"{destination}-wal") + if suffix == "-wal" + else None + ) + copied[suffix] = _stream_regular_artifact(source, directory_fd, target) + after = _capture_state_artifacts(state_path, directory_fd) + except (FileNotFoundError, _DiagnosticBusy): + destination.unlink(missing_ok=True) + Path(f"{destination}-wal").unlink(missing_ok=True) + time.sleep(0.05) + continue + if before != copied or copied != after: + destination.unlink(missing_ok=True) + Path(f"{destination}-wal").unlink(missing_ok=True) + time.sleep(0.05) + continue + return before + raise _DiagnosticBusy("state changed while creating a read-only snapshot") + + +def _status_counts(connection: sqlite3.Connection, table: str) -> dict[str, int]: + return { + str(row[0]): int(row[1]) + for row in connection.execute( + f'SELECT status, COUNT(*) FROM "{table}" GROUP BY status ORDER BY status' + ) + } + + +def _normalized_schema_sql(value: Any) -> str: + return " ".join(str(value or "").split()).casefold() + + +def _normalized_column_default(value: Any) -> str: + return str(value or "").strip() + + +def _schema_signature( + connection: sqlite3.Connection, + tables: set[str], +) -> tuple[ + dict[str, dict[str, tuple[str, int, str, int]]], + dict[str, tuple[str, int, int, tuple[str, ...], str]], + dict[str, frozenset[tuple[str, str, str, str, str, str]]], + frozenset[str], +]: + columns: dict[str, dict[str, tuple[str, int, str, int]]] = {} + indexes: dict[str, tuple[str, int, int, tuple[str, ...], str]] = {} + foreign_keys: dict[str, frozenset[tuple[str, str, str, str, str, str]]] = {} + autoincrement_tables: set[str] = set() + for table in tables: + table_definition = connection.execute( + "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", + (table,), + ).fetchone() + if "autoincrement" in _normalized_schema_sql( + table_definition[0] if table_definition else "" + ): + autoincrement_tables.add(table) + columns[table] = { + str(row[1]): ( + str(row[2]).upper(), + int(row[3]), + _normalized_column_default(row[4]), + int(row[5]), + ) + for row in connection.execute(f'PRAGMA table_info("{table}")') + } + for row in connection.execute(f'PRAGMA index_list("{table}")'): + name = str(row[1]) + definition = connection.execute( + "SELECT sql FROM sqlite_master WHERE type='index' AND name=?", + (name,), + ).fetchone() + index_columns = tuple( + str(item[2]) + for item in connection.execute(f'PRAGMA index_info("{name}")') + ) + indexes[name] = ( + table, + int(row[2]), + int(row[4]) if len(row) > 4 else 0, + index_columns, + _normalized_schema_sql(definition[0] if definition else ""), + ) + foreign_keys[table] = frozenset( + ( + str(row[2]), + str(row[3]), + str(row[4]), + str(row[5]), + str(row[6]), + str(row[7]), + ) + for row in connection.execute(f'PRAGMA foreign_key_list("{table}")') + ) + return columns, indexes, foreign_keys, frozenset(autoincrement_tables) + + +def _required_schema_signature() -> tuple[ + dict[str, dict[str, tuple[str, int, str, int]]], + dict[str, tuple[str, int, int, tuple[str, ...], str]], + dict[str, frozenset[tuple[str, str, str, str, str, str]]], + frozenset[str], +]: + reference = sqlite3.connect(":memory:") + try: + reference.row_factory = sqlite3.Row + reference.executescript(SCHEMA) + StateStore._migrate_processed_episodes(reference) + StateStore._migrate_approval_jobs(reference) + StateStore._migrate_notification_jobs(reference) + return _schema_signature(reference, set(_DIAGNOSTIC_REQUIRED_TABLES)) + finally: + reference.close() + + +def _diagnostic_state_summary(snapshot: Path) -> dict[str, Any]: + connection = sqlite3.connect(f"{snapshot.resolve().as_uri()}?mode=ro", uri=True, timeout=5) + try: + connection.execute("PRAGMA query_only=ON") + quick_check = str(connection.execute("PRAGMA quick_check").fetchone()[0]) + tables = { + str(row[0]) + for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'") + } + missing = sorted(_DIAGNOSTIC_REQUIRED_TABLES - tables) + ( + expected_columns, + expected_indexes, + expected_foreign_keys, + expected_autoincrement_tables, + ) = _required_schema_signature() + existing_required_tables = set(_DIAGNOSTIC_REQUIRED_TABLES) - set(missing) + ( + actual_columns, + actual_indexes, + actual_foreign_keys, + actual_autoincrement_tables, + ) = _schema_signature(connection, existing_required_tables) + missing_column_count = sum( + len(set(expected_columns[table]) - set(actual_columns[table])) + for table in existing_required_tables + ) + incompatible_column_count = sum( + sum( + actual_columns[table].get(name) != definition + for name, definition in expected_columns[table].items() + if name in actual_columns[table] + ) + for table in existing_required_tables + ) + missing_index_count = sum( + actual_indexes.get(name) != definition + for name, definition in expected_indexes.items() + ) + missing_foreign_key_count = sum( + len(expected_foreign_keys[table] - actual_foreign_keys.get(table, frozenset())) + for table in existing_required_tables + ) + missing_table_semantic_count = len( + expected_autoincrement_tables - actual_autoincrement_tables + ) + foreign_key_violation_count = sum( + 1 for _row in connection.execute("PRAGMA foreign_key_check") + ) + schema_version = int(connection.execute("PRAGMA user_version").fetchone()[0]) + schema_ready = ( + not missing + and missing_column_count == 0 + and incompatible_column_count == 0 + and missing_index_count == 0 + and missing_foreign_key_count == 0 + and missing_table_semantic_count == 0 + and schema_version == _DIAGNOSTIC_SCHEMA_VERSION + ) + integrity_ready = quick_check == "ok" and foreign_key_violation_count == 0 + summary: dict[str, Any] = { + "quick_check": quick_check, + "integrity_ready": integrity_ready, + "foreign_key_violation_count": foreign_key_violation_count, + "schema_version": schema_version, + "schema_ready": schema_ready, + "schema_version_supported": schema_version == _DIAGNOSTIC_SCHEMA_VERSION, + "missing_table_count": len(missing), + "missing_column_count": missing_column_count, + "incompatible_column_count": incompatible_column_count, + "missing_index_count": missing_index_count, + "missing_foreign_key_count": missing_foreign_key_count, + "missing_table_semantic_count": missing_table_semantic_count, + "episode_counts": ( + _status_counts(connection, "processed_episodes") + if "status" in actual_columns.get("processed_episodes", {}) + else {} + ), + "approval_jobs": ( + _status_counts(connection, "approval_jobs") + if "status" in actual_columns.get("approval_jobs", {}) + else {} + ), + "notification_jobs": ( + _status_counts(connection, "notification_jobs") + if "status" in actual_columns.get("notification_jobs", {}) + else {} + ), + "pending_messages": ( + int( + connection.execute( + "SELECT COUNT(*) FROM message_buffer WHERE consumed=0" + ).fetchone()[0] + ) + if "consumed" in actual_columns.get("message_buffer", {}) + else 0 + ), + "dead_letters": ( + int(connection.execute("SELECT COUNT(*) FROM dead_letter").fetchone()[0]) + if "dead_letter" in tables + else 0 + ), + } + return summary + finally: + connection.close() + + +def _diagnostic_error( + error: str, *, unchanged: bool | None, code: int +) -> tuple[dict[str, Any], int]: + return ( + { + "version": __version__, + "mode": "diagnostic", + "read_only": True, + "ok": False, + "error": error, + "operating_files_unchanged": unchanged, + "external_calls": [], + }, + code, + ) + + +def _diagnostic_payload( + config: Any, + *, + directory_fd: int | None = None, + config_fingerprint: tuple[Any, ...] | None = None, +) -> tuple[dict[str, Any], int]: + temp_parent = Path("/tmp") if os.name == "posix" and Path("/tmp").is_dir() else None + opened_here = directory_fd is None + try: + if directory_fd is None: + directory_fd = _open_state_directory(config.state_db) + _verify_state_directory_binding(config.state_db, directory_fd) + _verify_config_binding(config.path, directory_fd, config_fingerprint) + operating_entry = _capture_state_artifacts(config.state_db, directory_fd) + config_errors = validate_config(config, require_runtime=True) + with tempfile.TemporaryDirectory(prefix="hegi-diagnostic-", dir=temp_parent) as raw_temp: + temp_root = Path(raw_temp) + if os.name == "posix": + os.chmod(temp_root, 0o700) + if stat.S_IMODE(temp_root.stat().st_mode) & 0o077: + raise RuntimeError("private diagnostic storage is unavailable") + snapshot = temp_root / "state.db" + operating_before = _copy_stable_state_snapshot(config.state_db, directory_fd, snapshot) + state = _diagnostic_state_summary(snapshot) + operating_after = _capture_state_artifacts(config.state_db, directory_fd) + _verify_config_binding(config.path, directory_fd, config_fingerprint) + _verify_state_directory_binding(config.state_db, directory_fd) + except _DiagnosticBusy: + return _diagnostic_error("hegi-state-changing", unchanged=False, code=4) + except _DiagnosticArtifactError: + return _diagnostic_error("hegi-state-unsafe", unchanged=None, code=1) + except sqlite3.DatabaseError: + return _diagnostic_error("hegi-state-invalid", unchanged=None, code=1) + finally: + if opened_here and directory_fd is not None: + os.close(directory_fd) + unchanged = operating_entry == operating_before == operating_after + if not unchanged: + return _diagnostic_error("hegi-state-changing", unchanged=False, code=4) + ok = ( + not config_errors + and state["integrity_ready"] is True + and state["schema_ready"] is True + ) + payload = { + "version": __version__, + "mode": "diagnostic", + "read_only": True, + "ok": ok, + "config": {"valid": not config_errors, "error_count": len(config_errors)}, + "state": state, + "operating_files_unchanged": unchanged, + "external_calls": [], + } + return payload, 0 if ok else 1 + + def cmd_run_once(args: argparse.Namespace) -> int: + if bool(getattr(args, "diagnostic", False)): + try: + config, directory_fd, config_fingerprint = _load_diagnostic_config(args.config) + except _DiagnosticBusy: + payload, return_code = _diagnostic_error( + "hegi-state-changing", unchanged=False, code=4 + ) + _json(payload) + return return_code + except _DiagnosticArtifactError: + payload, return_code = _diagnostic_error( + "hegi-state-unsafe", unchanged=None, code=1 + ) + _json(payload) + return return_code + try: + _require_state_boundary(config) + payload, return_code = _diagnostic_payload( + config, + directory_fd=directory_fd, + config_fingerprint=config_fingerprint, + ) + _json(payload) + return return_code + finally: + if directory_fd is not None: + os.close(directory_fd) config = load_config(args.config) _require_state_boundary(config) dry_run = not bool(args.send) @@ -465,6 +1102,11 @@ def _configure_parser(parser: argparse.ArgumentParser) -> None: run_once = subparsers.add_parser("run-once", help="run one embedded pipeline cycle") run_mode = run_once.add_mutually_exclusive_group() + run_mode.add_argument( + "--diagnostic", + action="store_true", + help="inspect a private state snapshot without writes or external calls", + ) run_mode.add_argument("--dry-run", action="store_true") run_mode.add_argument("--send", action="store_true") run_once.set_defaults(hegi_handler=cmd_run_once) @@ -499,6 +1141,14 @@ def handle_hermes_cli(args: argparse.Namespace) -> int: return int(handler(args)) +def handle_hermes_cli_entry(args: argparse.Namespace) -> None: + """Preserve HEGI exit codes for Hermes hosts that ignore handler returns.""" + + return_code = handle_hermes_cli(args) + if return_code: + raise SystemExit(return_code) + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="HEGI AI Research Secretary") _configure_parser(parser) diff --git a/hegi/config.py b/hegi/config.py index 6c94a60..fa33ea9 100644 --- a/hegi/config.py +++ b/hegi/config.py @@ -168,16 +168,14 @@ def default_config_path(explicit_home: str | Path | None = None) -> Path: return resolve_runtime_home(hermes_root=resolve_hermes_home()) / "hegi" / "config.yaml" -def load_config(path: str | Path | None = None) -> HegiConfig: - config_path = Path(path).expanduser() if path else default_config_path() - if not config_path.exists(): - raise FileNotFoundError( - f"HEGI 설정 파일이 없습니다: {config_path}. " - "hegi/config/default.yaml을 복사하여 값을 설정하세요." - ) - loaded = load_yaml_mapping(config_path) +def _config_from_mapping( + config_path: Path, + loaded: dict[str, Any], + *, + home: Path | None = None, +) -> HegiConfig: raw = _deep_merge(DEFAULT_CONFIG, loaded) - home = config_path.resolve().parent.parent + home = home if home is not None else config_path.resolve().parent.parent chat_id = str(raw.get("telegram", {}).get("chat_id", "")).strip() agents: list[AgentSourceConfig] = [] for entry in raw.get("agents", []): @@ -220,6 +218,16 @@ def load_config(path: str | Path | None = None) -> HegiConfig: ) +def load_config(path: str | Path | None = None) -> HegiConfig: + config_path = Path(path).expanduser() if path else default_config_path() + if not config_path.exists(): + raise FileNotFoundError( + f"HEGI 설정 파일이 없습니다: {config_path}. " + "hegi/config/default.yaml을 복사하여 값을 설정하세요." + ) + return _config_from_mapping(config_path, load_yaml_mapping(config_path)) + + def state_path_is_bound(config: HegiConfig) -> bool: # Compare normalized lexical paths without dereferencing the state leaf. # ``Path.resolve()`` would make ``hegi/state.db -> foreign.db`` appear diff --git a/hegi/gateway_plugin.py b/hegi/gateway_plugin.py index d705bb6..88e5561 100644 --- a/hegi/gateway_plugin.py +++ b/hegi/gateway_plugin.py @@ -4,8 +4,10 @@ import json import logging +import os import re import sqlite3 +import sys import threading from pathlib import Path from typing import Any @@ -28,6 +30,21 @@ _CONFIG_SIGNATURE: tuple[str, int, int] | None = None _STATE_CACHE: StateStore | None = None + +def _is_gateway_process() -> bool: + arguments = [str(item).strip().lower() for item in sys.argv[1:]] + if any( + arguments[index] == "gateway" and arguments[index + 1] == "run" + for index in range(len(arguments) - 1) + ): + return True + return ( + os.environ.get("_HERMES_GATEWAY", "").strip() == "1" + and "gateway.run" in sys.modules + and "hegi" not in arguments + ) + + def _platform_name(event: Any) -> str: platform = getattr(getattr(event, "source", None), "platform", "") return str(getattr(platform, "value", platform)).lower() @@ -315,6 +332,14 @@ def _with_process_lock(config: HegiConfig) -> bool: return acquired +def _initialize_worker_state(config: HegiConfig) -> bool: + acquired, _ = run_with_process_lock( + config.state_db.parent / "daemon.lock", + lambda: _state_for(config), + ) + return acquired + + def _worker_iteration() -> bool: """Run one reload/validation cycle without allowing the daemon thread to die.""" @@ -333,6 +358,7 @@ def _worker_iteration() -> bool: def _worker_loop() -> None: + state_initialized = False while True: try: config = _cached_config() @@ -345,6 +371,20 @@ def _worker_loop() -> None: _LOGGER.warning("HEGI worker config reload paused: %s", type(exc).__name__) config = None poll_seconds = 60 + if not state_initialized and config is not None and config.enabled: + try: + if not validate_config(config, require_runtime=True): + state_initialized = _initialize_worker_state(config) + except Exception as exc: + _LOGGER.warning("HEGI worker state initialization paused: %s", type(exc).__name__) + if not state_initialized: + _WORKER_WAKE.wait(min(poll_seconds, 1)) + _WORKER_WAKE.clear() + continue + elif not state_initialized: + _WORKER_WAKE.wait(poll_seconds) + _WORKER_WAKE.clear() + continue _WORKER_WAKE.wait(poll_seconds) _WORKER_WAKE.clear() _worker_iteration() @@ -369,13 +409,13 @@ def register(context: Any) -> None: identity = id(context) if identity in _REGISTERED_CONTEXTS: return - from .cli import handle_hermes_cli, setup_hermes_cli + from .cli import handle_hermes_cli_entry, setup_hermes_cli context.register_cli_command( name="hegi", help="HEGI AI Research Secretary", setup_fn=setup_hermes_cli, - handler_fn=handle_hermes_cli, + handler_fn=handle_hermes_cli_entry, description="Configure, diagnose, and operate the HEGI plugin", ) context.register_hook("pre_gateway_dispatch", intercept_telegram_approval) @@ -386,6 +426,8 @@ def register(context: Any) -> None: except Exception as exc: runtime_ready = False _LOGGER.warning("HEGI host adapters unavailable: %s", type(exc).__name__) + if not _is_gateway_process(): + return config = _cached_config() if config is None: _LOGGER.warning("HEGI is installed but not configured; run `hermes hegi setup`") @@ -395,7 +437,6 @@ def register(context: Any) -> None: _LOGGER.warning("HEGI config is not ready: %s", "; ".join(errors)) return if runtime_ready: - _state_for(config) _ensure_worker() diff --git a/plugin.yaml b/plugin.yaml index 1374e49..ff32333 100644 --- a/plugin.yaml +++ b/plugin.yaml @@ -1,5 +1,5 @@ name: hegi-telegram -version: "2.1.0" +version: "2.1.1" description: AI Research Secretary with professor-gated Telegram approval workflows author: Lee Yong Wook kind: standalone diff --git a/pyproject.toml b/pyproject.toml index afac15a..251b3b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "hermes-hegi" -version = "2.1.0" +version = "2.1.1" description = "AI Research Secretary plugin for Hermes Agent" readme = "README.md" requires-python = ">=3.11" diff --git a/scripts/migrate-existing-install.sh b/scripts/migrate-existing-install.sh index 929eca7..a972aa8 100755 --- a/scripts/migrate-existing-install.sh +++ b/scripts/migrate-existing-install.sh @@ -28,8 +28,18 @@ mode="dry-run" mode_was_set="false" runtime_home="${HERMES_HOME:-${HOME:-}/.hermes}" keep_old_service="false" +gateway_quiesced="false" +migration_mutation_started="false" +gateway_lock_timeout="${HEGI_MIGRATION_GATEWAY_LOCK_TIMEOUT:-15}" +lock_control_dir="" +lock_error="" +hegi_lock_pid="" +gateway_lock_pid="" repo_root="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd -P)" +[[ "${gateway_lock_timeout}" =~ ^[0-9]+([.][0-9]+)?$ ]] \ + || die "HEGI_MIGRATION_GATEWAY_LOCK_TIMEOUT must be a non-negative number" + while (($#)); do case "$1" in --dry-run | --apply) @@ -115,6 +125,10 @@ elif command -v python >/dev/null 2>&1; then else die "Python was not found; it is required for migration and SQLite backup" fi +"${python_bin}" -c \ + 'import sys; raise SystemExit(0 if float(sys.argv[1]) <= 60 else 1)' \ + "${gateway_lock_timeout}" \ + || die "HEGI_MIGRATION_GATEWAY_LOCK_TIMEOUT must not exceed 60 seconds" "${python_bin}" - "${state_path}" <<'PY' import stat @@ -157,35 +171,532 @@ legacy_unit_file_exists() { return 1 } -acquire_hegi_lock() { - local lock_path="${hegi_home}/daemon.lock" - command -v flock >/dev/null 2>&1 || die "flock is required for a safe HEGI migration" - "${python_bin}" - "${lock_path}" <<'PY' +initialize_lock_control() { + local temp_parent="${TMPDIR:-/tmp}" + [[ -d "${temp_parent}" ]] || die "temporary directory is unavailable: ${temp_parent}" + lock_control_dir="$(mktemp -d -- "${temp_parent%/}/hegi-migration-locks.XXXXXXXX")" + chmod 700 -- "${lock_control_dir}" + printf 'hegi-migration-lock-control-v1\n' >"${lock_control_dir}/.marker" +} + +start_lock_holder() { + local name="$1" + local lock_path="$2" + local timeout="$3" + local ready_path="${lock_control_dir}/${name}.ready" + local error_path="${lock_control_dir}/${name}.error" + local pid_variable="${name}_lock_pid" + local holder_pid + + rm -f -- "${ready_path}" "${error_path}" + "${python_bin}" - \ + "${lock_path}" "${ready_path}" "${error_path}" "${timeout}" "$$" <<'PY' & +import ctypes +import fcntl import os +import signal import stat import sys +import time from pathlib import Path -path = Path(sys.argv[1]) -if path.parent.is_symlink() or path.parent.absolute() != path.parent.resolve(): - raise SystemExit("HEGI lock directory must not contain symbolic links") -flags = os.O_RDWR | os.O_CREAT -if hasattr(os, "O_CLOEXEC"): - flags |= os.O_CLOEXEC -if hasattr(os, "O_NOFOLLOW"): - flags |= os.O_NOFOLLOW -descriptor = os.open(path, flags, 0o600) +lock_arg, ready_arg, error_arg, timeout_arg, parent_arg = sys.argv[1:] +path = Path(lock_arg) +ready = Path(ready_arg) +error = Path(error_arg) +timeout = float(timeout_arg) +expected_parent = int(parent_arg) +descriptor = None + + +def fail(code: str) -> None: + fd = os.open(error, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o600) + try: + os.write(fd, (code + "\n").encode()) + finally: + os.close(fd) + raise SystemExit(1) + + try: + if sys.platform.startswith("linux"): + libc = ctypes.CDLL(None, use_errno=True) + if libc.prctl(1, signal.SIGTERM, 0, 0, 0) != 0: + fail("parent-watch-unavailable") + if os.getppid() != expected_parent: + raise SystemExit(1) + if path.parent.is_symlink() or path.parent.absolute() != path.parent.resolve(): + fail("unsafe-path") + flags = os.O_RDWR | os.O_CREAT | os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(path, flags, 0o600) + except OSError: + fail("unsafe-path") metadata = os.fstat(descriptor) if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: - raise SystemExit("HEGI lock must be a regular single-link file") - os.fchmod(descriptor, 0o600) + fail("unsafe-artifact") + deadline = time.monotonic() + timeout + while True: + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + if time.monotonic() >= deadline: + fail("busy") + time.sleep(0.05) + try: + current = path.lstat() + except OSError: + fail("path-changed") + if ( + not stat.S_ISREG(current.st_mode) + or current.st_nlink != 1 + or (current.st_dev, current.st_ino) != (metadata.st_dev, metadata.st_ino) + ): + fail("path-changed") + ready_fd = os.open( + ready, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o600 + ) + try: + os.write(ready_fd, f"{os.getpid()}\n".encode()) + finally: + os.close(ready_fd) + + stopping = False + + def stop(_signum, _frame): + global stopping + stopping = True + + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + while not stopping: + if os.getppid() != expected_parent: + break + try: + current = path.lstat() + except OSError: + fail("path-changed") + if ( + not stat.S_ISREG(current.st_mode) + or current.st_nlink != 1 + or (current.st_dev, current.st_ino) != (metadata.st_dev, metadata.st_ino) + ): + fail("path-changed") + time.sleep(0.05) +finally: + if descriptor is not None: + try: + fcntl.flock(descriptor, fcntl.LOCK_UN) + except OSError: + pass + os.close(descriptor) +PY + holder_pid="$!" + printf -v "${pid_variable}" '%s' "${holder_pid}" + while [[ ! -f "${ready_path}" ]]; do + if ! kill -0 "${holder_pid}" 2>/dev/null; then + wait "${holder_pid}" 2>/dev/null || true + lock_error="unknown" + [[ -f "${error_path}" ]] && IFS= read -r lock_error <"${error_path}" + printf -v "${pid_variable}" '%s' "" + return 1 + fi + sleep 0.02 + done +} + +assert_lock_holder_alive() { + local name="$1" + local pid_variable="${name}_lock_pid" + local holder_pid="${!pid_variable:-}" + local process_state="" + [[ -n "${holder_pid}" ]] \ + && process_state="$(ps -o stat= -p "${holder_pid}" 2>/dev/null)" \ + || return 1 + [[ -n "${holder_pid}" \ + && -f "${lock_control_dir}/${name}.ready" \ + && ! -f "${lock_control_dir}/${name}.error" \ + && -n "${process_state}" \ + && "${process_state:0:1}" != "Z" ]] \ + && kill -0 "${holder_pid}" 2>/dev/null +} + +release_lock_holder() { + local name="$1" + local pid_variable="${name}_lock_pid" + local holder_pid="${!pid_variable:-}" + if [[ -n "${holder_pid}" ]]; then + kill -TERM "${holder_pid}" 2>/dev/null || true + wait "${holder_pid}" 2>/dev/null || true + printf -v "${pid_variable}" '%s' "" + fi +} + +run_guarded() { + local hegi_pid="${hegi_lock_pid:--}" + local gateway_pid="${gateway_lock_pid:--}" + local child_pid_path="${lock_control_dir}/guarded-child.pid" + local guard_result=0 + local child_group="" + rm -f -- "${child_pid_path}" + if "${python_bin}" - \ + "$$" "${lock_control_dir}" "${hegi_pid}" "${gateway_pid}" \ + "${child_pid_path}" "$@" <<'PY' +import ctypes +import os +import signal +import subprocess +import sys +import time +from pathlib import Path + +expected_parent = int(sys.argv[1]) +control = Path(sys.argv[2]) +holder_specs = { + "hegi": None if sys.argv[3] == "-" else int(sys.argv[3]), + "gateway": None if sys.argv[4] == "-" else int(sys.argv[4]), +} +child_pid_path = Path(sys.argv[5]) +command = sys.argv[6:] +if not command: + raise SystemExit(125) + +stopping = False + + +def stop(_signum, _frame): + global stopping + stopping = True + + +signal.signal(signal.SIGTERM, stop) +signal.signal(signal.SIGINT, stop) +if sys.platform.startswith("linux"): + libc = ctypes.CDLL(None, use_errno=True) + if libc.prctl(1, signal.SIGTERM, 0, 0, 0) != 0: + raise SystemExit(125) +if os.getppid() != expected_parent: + raise SystemExit(125) + + +def holder_is_healthy(name: str, pid: int | None) -> bool: + if pid is None: + return True + if not (control / f"{name}.ready").is_file() or (control / f"{name}.error").exists(): + return False + try: + raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8") + state = raw.rpartition(") ")[2].split()[0] + os.kill(pid, 0) + except (OSError, IndexError): + return False + return state != "Z" + + +def locks_are_healthy() -> bool: + return all(holder_is_healthy(name, pid) for name, pid in holder_specs.items()) + + +if not locks_are_healthy(): + raise SystemExit(125) +guard_pid = os.getpid() + + +def prepare_child() -> None: + signal.signal(signal.SIGTERM, signal.SIG_DFL) + signal.signal(signal.SIGINT, signal.SIG_DFL) + os.setsid() + if sys.platform.startswith("linux"): + child_libc = ctypes.CDLL(None, use_errno=True) + if child_libc.prctl(1, signal.SIGTERM, 0, 0, 0) != 0: + os._exit(125) + if os.getppid() != guard_pid: + os._exit(125) + + +child = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + preexec_fn=prepare_child, +) +pid_descriptor = os.open( + child_pid_path, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, + 0o600, +) +try: + os.write(pid_descriptor, f"{child.pid}\n".encode()) + os.fsync(pid_descriptor) finally: - os.close(descriptor) + os.close(pid_descriptor) +try: + while child.poll() is None: + if stopping or os.getppid() != expected_parent or not locks_are_healthy(): + try: + os.killpg(child.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + child.wait(timeout=5) + except subprocess.TimeoutExpired: + try: + os.killpg(child.pid, signal.SIGKILL) + except ProcessLookupError: + pass + child.wait() + raise SystemExit(125) + time.sleep(0.02) + return_code = child.returncode + if not locks_are_healthy(): + raise SystemExit(125) + raise SystemExit(return_code) +finally: + if child.poll() is None: + try: + os.killpg(child.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + child.wait(timeout=5) + except subprocess.TimeoutExpired: + try: + os.killpg(child.pid, signal.SIGKILL) + except ProcessLookupError: + pass + child.wait() +PY + then + guard_result=0 + else + guard_result="$?" + fi + if [[ -f "${child_pid_path}" ]]; then + IFS= read -r child_group <"${child_pid_path}" || true + fi + if [[ "${child_group}" =~ ^[1-9][0-9]*$ ]] \ + && kill -0 -- "-${child_group}" 2>/dev/null; then + kill -TERM -- "-${child_group}" 2>/dev/null || true + for _attempt in {1..50}; do + kill -0 -- "-${child_group}" 2>/dev/null || break + sleep 0.1 + done + if kill -0 -- "-${child_group}" 2>/dev/null; then + kill -KILL -- "-${child_group}" 2>/dev/null || true + fi + fi + rm -f -- "${child_pid_path}" + return "${guard_result}" +} + +acquire_hegi_lock() { + lock_error="" + start_lock_holder hegi "${hegi_home}/daemon.lock" 0 \ + || die "another HEGI worker is active or its lock is unsafe (${lock_error})" + printf 'exclusive_lock=%s\n' "${hegi_home}/daemon.lock" +} + +acquire_gateway_lock() { + lock_error="" + start_lock_holder gateway "${runtime_home}/gateway.lock" "${gateway_lock_timeout}" \ + || die "Hermes gateway did not quiesce; refusing a live snapshot (${lock_error})" + gateway_quiesced="true" + printf 'exclusive_gateway_lock=%s\n' "${runtime_home}/gateway.lock" +} + +release_gateway_lock() { + release_lock_holder gateway +} + +wait_for_gateway_ready() { + local timeout="${1:-${gateway_lock_timeout}}" + local phase="${2:-initial}" + "${python_bin}" - \ + "${runtime_home}" "${timeout}" "${lock_control_dir}/gateway.identity" "${phase}" <<'PY' +import fcntl +import json +import os +import stat +import sys +import time +from pathlib import Path + +home = Path(sys.argv[1]).resolve() +timeout = float(sys.argv[2]) +proof_path = Path(sys.argv[3]) +phase = sys.argv[4] +if phase not in {"initial", "final"}: + raise SystemExit(1) + + +def read_record(path: Path, *, lock_required: bool = False): + flags = os.O_RDONLY | os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(path, flags) + except OSError: + return None + try: + before = os.fstat(descriptor) + if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or before.st_size > 65536: + return None + if lock_required: + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + pass + else: + fcntl.flock(descriptor, fcntl.LOCK_UN) + return None + chunks = [] + remaining = 65537 + while remaining: + block = os.read(descriptor, min(remaining, 8192)) + if not block: + break + chunks.append(block) + remaining -= len(block) + if remaining == 0: + return None + after = os.fstat(descriptor) + current = path.lstat() + if ( + (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) + != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns) + or (after.st_dev, after.st_ino) != (current.st_dev, current.st_ino) + or not stat.S_ISREG(current.st_mode) + or current.st_nlink != 1 + ): + return None + payload = json.loads(b"".join(chunks)) + return payload if isinstance(payload, dict) else None + except (OSError, UnicodeError, json.JSONDecodeError): + return None + finally: + os.close(descriptor) + + +def process_start_time(pid: int): + try: + raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8") + return int(raw.rpartition(") ")[2].split()[19]) + except (OSError, ValueError, IndexError): + return None + + +def process_matches(pid: int, start_time: int) -> bool: + if process_start_time(pid) != start_time: + return False + try: + command = Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0") + arguments = [item.decode("utf-8", "replace").lower() for item in command if item] + except OSError: + return False + return any( + arguments[index] == "gateway" + and arguments[index + 1] in {"run", "restart"} + for index in range(len(arguments) - 1) + ) + + +def probe(): + lock = read_record(home / "gateway.lock", lock_required=True) + pid_record = read_record(home / "gateway.pid") + runtime = read_record(home / "gateway_state.json") + if not all(isinstance(item, dict) for item in (lock, pid_record, runtime)): + return None + try: + pid = int(lock["pid"]) + start_time = int(lock["start_time"]) + except (KeyError, TypeError, ValueError): + return None + expected_home = str(home) + for record in (lock, pid_record): + if ( + record.get("kind") != "hermes-gateway" + or record.get("pid") != pid + or record.get("start_time") != start_time + or str(Path(str(record.get("hermes_home", ""))).resolve()) != expected_home + ): + return None + if ( + runtime.get("kind") != "hermes-gateway" + or runtime.get("pid") != pid + or runtime.get("start_time") != start_time + or runtime.get("gateway_state") != "running" + or not process_matches(pid, start_time) + ): + return None + final_lock = read_record(home / "gateway.lock", lock_required=True) + if not isinstance(final_lock, dict): + return None + if (final_lock.get("pid"), final_lock.get("start_time")) != (pid, start_time): + return None + return pid, start_time + + +deadline = time.monotonic() + timeout +required_identity = None +first_observed = None +if phase == "final" and proof_path.is_file(): + try: + proof = json.loads(proof_path.read_text(encoding="utf-8")) + required_identity = (int(proof["pid"]), int(proof["start_time"])) + first_observed = float(proof["first_observed"]) + except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError): + raise SystemExit(1) from None +stable_identity = required_identity +stable_samples = 0 +while time.monotonic() <= deadline: + identity = probe() + now = time.monotonic() + if required_identity is not None and identity != required_identity: + raise SystemExit(1) + if identity is not None and identity == stable_identity: + stable_samples += 1 + elif identity is not None: + stable_identity = identity + stable_samples = 1 + if required_identity is None: + first_observed = now + else: + stable_identity = None + stable_samples = 0 + if required_identity is None: + first_observed = None + if stable_samples >= 3 and stable_identity is not None: + if phase == "initial": + payload = { + "pid": stable_identity[0], + "start_time": stable_identity[1], + "first_observed": first_observed, + } + descriptor = os.open( + proof_path, + os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_CLOEXEC, + 0o600, + ) + try: + os.write(descriptor, json.dumps(payload).encode()) + os.fsync(descriptor) + finally: + os.close(descriptor) + raise SystemExit(0) + if first_observed is not None and now - first_observed >= 2.0: + raise SystemExit(0) + time.sleep(0.15) +raise SystemExit(1) PY - exec {hegi_lock_fd}<>"${lock_path}" - flock -n "${hegi_lock_fd}" || die "another HEGI worker is active; stop it before migration" - printf 'exclusive_lock=%s\n' "${lock_path}" +} + +start_gateway_detached() { + ( + HERMES_HOME="${runtime_home}" \ + nohup hermes gateway restart /dev/null 2>&1 & + ) } disable_legacy_service_safely() { @@ -224,7 +735,7 @@ disable_legacy_service_safely() { loaded) ;; *) die "cannot determine legacy hegi.service load state" ;; esac - systemctl --user disable --now hegi.service \ + run_guarded systemctl --user disable --now hegi.service \ || die "failed to disable legacy hegi.service" service_info="$( systemctl --user show hegi.service \ @@ -282,6 +793,43 @@ require_inactive_legacy_service() { esac } +restart_gateway_after_early_failure() { + local exit_status="$?" + trap - EXIT + if [[ "${exit_status}" -ne 0 \ + && "${gateway_quiesced}" == "true" \ + && "${migration_mutation_started}" == "false" ]]; then + printf 'migration failed before install or schema changes; restarting the original gateway\n' >&2 + release_gateway_lock + release_lock_holder hegi + start_gateway_detached + if ! wait_for_gateway_ready "${gateway_lock_timeout}" final; then + printf 'warning: original gateway restart did not become stably ready\n' >&2 + fi + fi + release_gateway_lock + release_lock_holder hegi + cleanup_lock_control + exit "${exit_status}" +} + +cleanup_lock_control() { + if [[ -n "${lock_control_dir}" \ + && -f "${lock_control_dir}/.marker" \ + && "$(<"${lock_control_dir}/.marker")" == "hegi-migration-lock-control-v1" ]]; then + rm -f -- \ + "${lock_control_dir}/hegi.ready" \ + "${lock_control_dir}/hegi.error" \ + "${lock_control_dir}/gateway.ready" \ + "${lock_control_dir}/gateway.error" \ + "${lock_control_dir}/gateway.identity" \ + "${lock_control_dir}/guarded-child.pid" \ + "${lock_control_dir}/.marker" + rmdir -- "${lock_control_dir}" 2>/dev/null || true + fi + lock_control_dir="" +} + if [[ "${mode}" == "dry-run" ]]; then run_hegi "${migrate_args[@]}" printf 'dry-run complete; no operating files were modified\n' @@ -289,6 +837,10 @@ if [[ "${mode}" == "dry-run" ]]; then fi command -v hermes >/dev/null 2>&1 || die "hermes command was not found" +command -v nohup >/dev/null 2>&1 || die "nohup is required for a safe gateway restart" +command -v ps >/dev/null 2>&1 || die "ps is required for lock-holder verification" +initialize_lock_control +trap restart_gateway_after_early_failure EXIT acquire_hegi_lock # Keeping the legacy unit means preserving its installation/enablement state, @@ -298,6 +850,14 @@ if [[ "${keep_old_service}" == "true" ]]; then require_inactive_legacy_service fi +# The legacy feature/hegi-v2 gateway hook can write approval state without the +# standalone worker lock. Quiesce that profile before taking the rollback +# snapshot so the backup is an exact point-in-time image of all HEGI writers. +HERMES_HOME="${runtime_home}" hermes gateway stop +acquire_gateway_lock +assert_lock_holder_alive hegi && assert_lock_holder_alive gateway \ + || die "a migration lock path changed before the rollback snapshot" + install -d -m 700 -- "${backup_dir}" cp -p -- "${config_path}" "${backup_dir}/config.yaml" chmod 600 -- "${backup_dir}/config.yaml" @@ -367,13 +927,24 @@ for path in sorted(root.rglob("*")): PY mv -- "${manifest_tmp}" "${backup_dir}/.hegi-migration-manifest.sha256" chmod -R go-rwx -- "${backup_dir}" +assert_lock_holder_alive hegi && assert_lock_holder_alive gateway \ + || die "a migration lock path changed while the rollback snapshot was created" # Install and migrate before the gateway loads the standalone plugin. The # explicit enable is retained for Hermes versions where install --enable only # records intent but does not normalize the plugin key. -HERMES_HOME="${runtime_home}" hermes plugins install icerain-cmd/hermes-hegi --force --enable -HERMES_HOME="${runtime_home}" hermes plugins enable hegi-telegram --no-allow-tool-override -run_hegi "${migrate_args[@]}" +migration_mutation_started="true" +run_guarded env HERMES_HOME="${runtime_home}" \ + hermes plugins install icerain-cmd/hermes-hegi --force --enable \ + || die "standalone HEGI plugin installation failed" +run_guarded env HERMES_HOME="${runtime_home}" \ + hermes plugins enable hegi-telegram --no-allow-tool-override \ + || die "standalone HEGI plugin enablement failed" +run_guarded env \ + HERMES_HOME="${runtime_home}" \ + PYTHONPATH="${repo_root}${PYTHONPATH:+:${PYTHONPATH}}" \ + "${hegi_command[@]}" "${migrate_args[@]}" \ + || die "standalone HEGI state migration failed" # Unless the operator explicitly opts out, stop the old worker before the # gateway starts the embedded worker. The previous service state is backed up. @@ -381,14 +952,27 @@ if [[ "${keep_old_service}" != "true" ]]; then disable_legacy_service_safely fi -( - exec {hegi_lock_fd}>&- - HERMES_HOME="${runtime_home}" hermes gateway restart -) || die "failed to restart the gateway with standalone HEGI" -flock -u "${hegi_lock_fd}" -exec {hegi_lock_fd}>&- -HERMES_HOME="${runtime_home}" hermes hegi doctor -HERMES_HOME="${runtime_home}" hermes hegi run-once --dry-run +assert_lock_holder_alive hegi && assert_lock_holder_alive gateway \ + || die "a migration lock failed during installation" +release_gateway_lock +start_gateway_detached +wait_for_gateway_ready "${gateway_lock_timeout}" initial \ + || die "restarted gateway did not become stably ready" +assert_lock_holder_alive hegi \ + || die "the HEGI worker lock failed during gateway restart" +gateway_quiesced="false" +run_guarded env HERMES_HOME="${runtime_home}" hermes hegi doctor \ + || die "HEGI doctor failed after migration" +run_guarded env HERMES_HOME="${runtime_home}" \ + hermes hegi run-once --diagnostic \ + || die "read-only HEGI diagnostic failed after migration" +wait_for_gateway_ready 5 final \ + || die "gateway readiness did not remain stable through post-checks" +assert_lock_holder_alive hegi \ + || die "the HEGI worker lock path changed during post-checks" +release_lock_holder hegi +cleanup_lock_control +trap - EXIT printf 'migration complete\n' printf 'archive_preserved=%s\n' "${archive_path}" diff --git a/tests/compatibility/test_migration.py b/tests/compatibility/test_migration.py index 900740a..8fd35e0 100644 --- a/tests/compatibility/test_migration.py +++ b/tests/compatibility/test_migration.py @@ -5,9 +5,11 @@ import json import os import shlex +import signal import sqlite3 import subprocess import sys +import time from pathlib import Path import pytest @@ -66,6 +68,34 @@ def _approval_columns(path: Path) -> set[str]: connection.close() +def _wait_for_lock_release(paths: list[Path], timeout: float = 5) -> None: + deadline = time.monotonic() + timeout + while True: + handles = [] + try: + for path in paths: + handle = path.open("a+") + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + handle.close() + raise + handles.append(handle) + except BlockingIOError: + for handle in handles: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + handle.close() + if time.monotonic() >= deadline: + raise AssertionError("orphaned process retained a migration lock") from None + time.sleep(0.05) + continue + else: + for handle in handles: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + handle.close() + return + + def _seal_backup(path: Path) -> None: (path / "old-service-status").write_text("absent\n", encoding="utf-8") (path / ".hegi-migration-backup").write_text( @@ -138,10 +168,137 @@ def _fake_commands( f"exec {shlex.quote(sys.executable)} -m hegi \"$@\"\n", encoding="utf-8", ) + gateway_runtime = bin_dir / "fake_gateway_runtime.py" + gateway_runtime.write_text( + """import fcntl +import json +import os +import sys +import time +from pathlib import Path + +lock_path = Path(sys.argv[1]) +runtime_home = Path(sys.argv[2]).resolve() +duration = float(sys.argv[3]) +pid_output = sys.argv[4] +descriptor = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) +fcntl.flock(descriptor, fcntl.LOCK_EX) +pid = os.getpid() +raw_stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8") +start_time = int(raw_stat.rpartition(") ")[2].split()[19]) +record = { + "pid": pid, + "kind": "hermes-gateway", + "argv": ["hermes", "gateway", "run"], + "start_time": start_time, + "hermes_home": str(runtime_home), +} +os.ftruncate(descriptor, 0) +os.write(descriptor, json.dumps(record).encode()) +os.fsync(descriptor) +(runtime_home / "gateway.pid").write_text(json.dumps(record), encoding="utf-8") +runtime = dict(record) +runtime.pop("hermes_home") +runtime["gateway_state"] = "running" +(runtime_home / "gateway_state.json").write_text(json.dumps(runtime), encoding="utf-8") +if pid_output != "-": + Path(pid_output).write_text(f"{pid}\\n", encoding="utf-8") +time.sleep(duration) +""", + encoding="utf-8", + ) + rebind_gateway_lock = bin_dir / "rebind_gateway_lock.py" + rebind_gateway_lock.write_text( + """import fcntl +import os +import sys +import time +from pathlib import Path + +lock = Path(sys.argv[1]) +replacement_arg = sys.argv[2] +replacement = Path(replacement_arg) +marker = Path(sys.argv[3]) +deadline = time.monotonic() + 5 +while time.monotonic() < deadline: + descriptor = os.open(lock, os.O_RDWR | os.O_CREAT, 0o600) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + os.close(descriptor) + break + else: + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) + time.sleep(0.005) +else: + raise SystemExit(2) +if replacement_arg == "-": + lock.unlink() +else: + replacement.write_text("replacement", encoding="utf-8") + os.replace(replacement, lock) +marker.write_text("rebound", encoding="utf-8") +""", + encoding="utf-8", + ) hermes = bin_dir / "hermes" hermes.write_text( "#!/usr/bin/env bash\n" "printf 'hermes %s\\n' \"$*\" >>\"${HEGI_TEST_COMMAND_LOG}\"\n" + "for inherited_fd in /proc/$$/fd/*; do\n" + " inherited_target=\"$(readlink \"${inherited_fd}\" 2>/dev/null || true)\"\n" + " case \"${inherited_target}\" in\n" + " \"${HERMES_HOME}/gateway.lock\"|\"${HERMES_HOME}/hegi/daemon.lock\") exit 94 ;;\n" + " esac\n" + "done\n" + "if [[ \"$*\" == 'gateway stop' " + "&& -n \"${HEGI_TEST_GATEWAY_STOP_STATE:-}\" ]]; then\n" + " \"${HEGI_TEST_PYTHON}\" -c 'import sqlite3, sys; " + "connection = sqlite3.connect(sys.argv[1]); " + "connection.execute(\"UPDATE legacy_notes SET value = ?\", " + "(\"quiesced-at-stop\",)); connection.commit(); connection.close()' " + "\"${HEGI_TEST_GATEWAY_STOP_STATE}\"\n" + "fi\n" + "if [[ \"$*\" == 'gateway stop' " + "&& -n \"${HEGI_TEST_REBIND_GATEWAY_LOCK:-}\" ]]; then\n" + f" \"${{HEGI_TEST_PYTHON}}\" {shlex.quote(str(rebind_gateway_lock))} " + "\"${HERMES_HOME}/gateway.lock\" " + "\"${HEGI_TEST_REBIND_GATEWAY_REPLACEMENT}\" " + "\"${HEGI_TEST_REBIND_GATEWAY_LOCK}\" >/dev/null 2>&1 &\n" + "fi\n" + "if [[ \"$*\" == 'plugins install icerain-cmd/hermes-hegi --force --enable' " + "&& -n \"${HEGI_TEST_BLOCK_INSTALL_PID:-}\" ]]; then\n" + " printf '%s\\n' \"$$\" >\"${HEGI_TEST_BLOCK_INSTALL_PID}\"\n" + " if [[ -n \"${HEGI_TEST_MUTATION_REBIND_MARKER:-}\" ]]; then\n" + f" \"${{HEGI_TEST_PYTHON}}\" {shlex.quote(str(rebind_gateway_lock))} " + "\"${HERMES_HOME}/gateway.lock\" " + "\"${HEGI_TEST_MUTATION_REBIND_REPLACEMENT}\" " + "\"${HEGI_TEST_MUTATION_REBIND_MARKER}\" >/dev/null 2>&1 &\n" + " fi\n" + " sleep \"${HEGI_TEST_BLOCK_INSTALL_SECONDS:-30}\"\n" + " if [[ -n \"${HEGI_TEST_BLOCK_INSTALL_CONTINUED:-}\" ]]; then\n" + " printf 'continued\\n' >\"${HEGI_TEST_BLOCK_INSTALL_CONTINUED}\"\n" + " fi\n" + "fi\n" + "if [[ \"$*\" == 'gateway restart' ]]; then\n" + " if [[ -n \"${HEGI_TEST_RESTART_HEGI_LOCK_PROBE:-}\" ]]; then\n" + " exec {hegi_probe_fd}<>\"${HEGI_TEST_RESTART_HEGI_LOCK_PROBE}\"\n" + " flock -n \"${hegi_probe_fd}\" || exit 93\n" + " flock -u \"${hegi_probe_fd}\"\n" + " exec {hegi_probe_fd}>&-\n" + " printf 'acquired\\n' >\"${HEGI_TEST_RESTART_LOCK_RESULT}\"\n" + " fi\n" + " if [[ \"${HEGI_TEST_FOREGROUND_RESTART:-false}\" == 'true' ]]; then\n" + f" exec \"${{HEGI_TEST_PYTHON}}\" {shlex.quote(str(gateway_runtime))} " + "\"${HERMES_HOME}/gateway.lock\" \"${HERMES_HOME}\" 30 " + "\"${HEGI_TEST_FOREGROUND_RESTART_PID}\" gateway run\n" + " fi\n" + f" \"${{HEGI_TEST_PYTHON}}\" {shlex.quote(str(gateway_runtime))} " + "\"${HERMES_HOME}/gateway.lock\" \"${HERMES_HOME}\" " + "\"${HEGI_TEST_GATEWAY_HOLD_SECONDS:-10}\" - gateway run " + ">/dev/null 2>&1 &\n" + "fi\n" "if [[ \"$*\" == 'plugins list --enabled --json' ]]; then\n" " printf '[]\\n'\n" "fi\n" @@ -176,6 +333,7 @@ def _fake_commands( environment["HOME"] = str(tmp_path) environment["HERMES_HOME"] = str(runtime_home) environment["HEGI_TEST_COMMAND_LOG"] = str(log_path) + environment["HEGI_TEST_PYTHON"] = sys.executable environment["PYTHON_BIN"] = sys.executable environment["PATH"] = f"{bin_dir}{os.pathsep}{environment['PATH']}" current_pythonpath = environment.get("PYTHONPATH", "") @@ -414,6 +572,337 @@ def test_migration_apply_backs_up_and_additively_migrates_state(tmp_path): assert not (runtime_home / "hegi" / "daemon.ready").exists() +def test_migration_quiesces_gateway_before_state_snapshot(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, _config_path, state_path = _runtime(tmp_path) + environment, log_path = _fake_commands(tmp_path, repo, runtime_home) + environment["HEGI_TEST_GATEWAY_STOP_STATE"] = str(state_path) + foreground_pid_path = tmp_path / "foreground-gateway.pid" + environment["HEGI_TEST_FOREGROUND_RESTART"] = "true" + environment["HEGI_TEST_FOREGROUND_RESTART_PID"] = str(foreground_pid_path) + + try: + result = _run_script( + repo / "scripts" / "migrate-existing-install.sh", + "--apply", + "--runtime-home", + str(runtime_home), + environment=environment, + repo=repo, + ) + finally: + if foreground_pid_path.exists(): + os.kill(int(foreground_pid_path.read_text(encoding="utf-8")), 15) + + assert result.returncode == 0, result.stderr + backup_dirs = list((runtime_home / "hegi" / "migration-backups").iterdir()) + assert len(backup_dirs) == 1 + assert _marker(backup_dirs[0] / "state.db") == "quiesced-at-stop" + assert _marker(state_path) == "quiesced-at-stop" + commands = log_path.read_text(encoding="utf-8").splitlines() + assert commands[0] == "hermes gateway stop" + assert commands.index("hermes gateway stop") < commands.index( + "hermes plugins install icerain-cmd/hermes-hegi --force --enable" + ) + + +def test_migration_refuses_snapshot_while_gateway_lock_remains_active(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, config_path, state_path = _runtime(tmp_path) + environment, log_path = _fake_commands(tmp_path, repo, runtime_home) + environment["HEGI_MIGRATION_GATEWAY_LOCK_TIMEOUT"] = "0.1" + gateway_lock = runtime_home / "gateway.lock" + ready = tmp_path / "gateway-lock-ready" + holder = subprocess.Popen( + [ + sys.executable, + "-c", + ( + "import fcntl, pathlib, sys, time; " + "handle = open(sys.argv[1], 'a+'); " + "fcntl.flock(handle, fcntl.LOCK_EX); " + "pathlib.Path(sys.argv[2]).write_text('ready'); " + "time.sleep(30)" + ), + str(gateway_lock), + str(ready), + ] + ) + try: + for _ in range(100): + if ready.exists(): + break + time.sleep(0.01) + assert ready.exists() + result = _run_script( + repo / "scripts" / "migrate-existing-install.sh", + "--apply", + "--runtime-home", + str(runtime_home), + environment=environment, + repo=repo, + ) + finally: + holder.terminate() + holder.wait(timeout=5) + + assert result.returncode != 0 + assert "gateway did not quiesce" in result.stderr + assert config_path.is_file() + assert _marker(state_path) == "live-state" + assert not (runtime_home / "hegi" / "migration-backups").exists() + command_log = log_path.read_text(encoding="utf-8") + assert "hermes gateway stop" in command_log + assert "hermes gateway restart" not in command_log + assert "hermes plugins install" not in command_log + + +def test_migration_rejects_gateway_lock_path_rebinding(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, _config_path, state_path = _runtime(tmp_path) + environment, log_path = _fake_commands(tmp_path, repo, runtime_home) + rebound_marker = tmp_path / "gateway-lock-rebound" + environment["HEGI_TEST_REBIND_GATEWAY_LOCK"] = str(rebound_marker) + environment["HEGI_TEST_REBIND_GATEWAY_REPLACEMENT"] = str( + tmp_path / "replacement-gateway.lock" + ) + + result = _run_script( + repo / "scripts" / "migrate-existing-install.sh", + "--apply", + "--runtime-home", + str(runtime_home), + environment=environment, + repo=repo, + ) + + assert result.returncode != 0 + assert rebound_marker.read_text(encoding="utf-8") == "rebound" + assert "lock path changed" in result.stderr + assert _marker(state_path) == "live-state" + command_log = log_path.read_text(encoding="utf-8") + assert "hermes plugins install" not in command_log + + +def test_migration_rejects_gateway_lock_unlink(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, _config_path, _state_path = _runtime(tmp_path) + environment, log_path = _fake_commands(tmp_path, repo, runtime_home) + rebound_marker = tmp_path / "gateway-lock-unlinked" + environment["HEGI_TEST_REBIND_GATEWAY_LOCK"] = str(rebound_marker) + environment["HEGI_TEST_REBIND_GATEWAY_REPLACEMENT"] = "-" + + result = _run_script( + repo / "scripts" / "migrate-existing-install.sh", + "--apply", + "--runtime-home", + str(runtime_home), + environment=environment, + repo=repo, + ) + + assert result.returncode != 0 + assert rebound_marker.read_text(encoding="utf-8") == "rebound" + assert "migration complete" not in result.stdout + assert "hermes plugins install" not in log_path.read_text(encoding="utf-8") + + +def test_guarded_install_stops_on_gateway_lock_rebinding(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, _config_path, _state_path = _runtime(tmp_path) + environment, log_path = _fake_commands(tmp_path, repo, runtime_home) + install_pid_path = tmp_path / "blocked-install.pid" + rebound_marker = tmp_path / "mutation-lock-rebound" + environment["HEGI_TEST_BLOCK_INSTALL_PID"] = str(install_pid_path) + environment["HEGI_TEST_MUTATION_REBIND_MARKER"] = str(rebound_marker) + environment["HEGI_TEST_MUTATION_REBIND_REPLACEMENT"] = str( + tmp_path / "mutation-replacement.lock" + ) + + result = _run_script( + repo / "scripts" / "migrate-existing-install.sh", + "--apply", + "--runtime-home", + str(runtime_home), + environment=environment, + repo=repo, + ) + + assert result.returncode != 0 + assert rebound_marker.read_text(encoding="utf-8") == "rebound" + assert "standalone HEGI plugin installation failed" in result.stderr + install_pid = int(install_pid_path.read_text(encoding="utf-8")) + assert not Path(f"/proc/{install_pid}").exists() + command_log = log_path.read_text(encoding="utf-8") + assert "hermes plugins install" in command_log + assert "hermes plugins enable" not in command_log + assert "migration complete" not in result.stdout + + +def test_guard_kill_cannot_orphan_mutation_child(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, _config_path, _state_path = _runtime(tmp_path) + environment, _log_path = _fake_commands(tmp_path, repo, runtime_home) + install_pid_path = tmp_path / "blocked-install.pid" + continued_path = tmp_path / "install-continued" + environment["HEGI_TEST_BLOCK_INSTALL_PID"] = str(install_pid_path) + environment["HEGI_TEST_BLOCK_INSTALL_SECONDS"] = "1" + environment["HEGI_TEST_BLOCK_INSTALL_CONTINUED"] = str(continued_path) + process = subprocess.Popen( + [ + "bash", + str(repo / "scripts" / "migrate-existing-install.sh"), + "--apply", + "--runtime-home", + str(runtime_home), + ], + cwd=repo, + env=environment, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and not install_pid_path.exists(): + if process.poll() is not None: + break + time.sleep(0.01) + assert install_pid_path.is_file(), "guarded install never started" + install_pid = int(install_pid_path.read_text(encoding="utf-8")) + status = Path(f"/proc/{install_pid}/status").read_text(encoding="utf-8") + guard_pid = int( + next(line for line in status.splitlines() if line.startswith("PPid:")) + .split(":", 1)[1] + .strip() + ) + os.kill(guard_pid, signal.SIGKILL) + assert process.wait(timeout=10) != 0 + time.sleep(1.2) + assert not continued_path.exists() + assert not Path(f"/proc/{install_pid}").exists() + _wait_for_lock_release( + [runtime_home / "hegi" / "daemon.lock", runtime_home / "gateway.lock"] + ) + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + +def test_guard_child_resets_death_signals_before_parent_watch() -> None: + repo = Path(__file__).resolve().parents[2] + migration = (repo / "scripts" / "migrate-existing-install.sh").read_text( + encoding="utf-8" + ) + prepare_child = migration.split("def prepare_child() -> None:", 1)[1].split( + "child = subprocess.Popen", 1 + )[0] + + term_reset = prepare_child.index( + "signal.signal(signal.SIGTERM, signal.SIG_DFL)" + ) + int_reset = prepare_child.index( + "signal.signal(signal.SIGINT, signal.SIG_DFL)" + ) + new_session = prepare_child.index("os.setsid()") + death_signal = prepare_child.index("child_libc.prctl(1, signal.SIGTERM") + parent_check = prepare_child.index("os.getppid() != guard_pid") + + assert max(term_reset, int_reset) < new_session < death_signal < parent_check + + +def test_migration_rejects_transient_gateway_readiness(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, _config_path, _state_path = _runtime(tmp_path) + environment, _log_path = _fake_commands(tmp_path, repo, runtime_home) + environment["HEGI_MIGRATION_GATEWAY_LOCK_TIMEOUT"] = "1.5" + environment["HEGI_TEST_GATEWAY_HOLD_SECONDS"] = "0.8" + + result = _run_script( + repo / "scripts" / "migrate-existing-install.sh", + "--apply", + "--runtime-home", + str(runtime_home), + environment=environment, + repo=repo, + ) + + assert result.returncode != 0 + assert "gateway readiness" in result.stderr + assert "migration complete" not in result.stdout + + +def test_migration_lock_holders_exit_when_parent_is_killed(tmp_path): + repo = Path(__file__).resolve().parents[2] + runtime_home, _config_path, _state_path = _runtime(tmp_path) + environment, _log_path = _fake_commands(tmp_path, repo, runtime_home) + control_root = tmp_path / "lock-control" + control_root.mkdir() + environment["TMPDIR"] = str(control_root) + process = subprocess.Popen( + [ + "bash", + str(repo / "scripts" / "migrate-existing-install.sh"), + "--apply", + "--runtime-home", + str(runtime_home), + ], + cwd=repo, + env=environment, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + ready_paths: list[Path] = [] + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + ready_paths = list( + control_root.glob("hegi-migration-locks.*/gateway.ready") + ) + if ready_paths: + break + if process.poll() is not None: + break + time.sleep(0.01) + assert ready_paths, "migration never acquired the gateway lock" + control_dir = ready_paths[0].parent + holder_pids = [ + int((control_dir / f"{name}.ready").read_text(encoding="utf-8")) + for name in ("hegi", "gateway") + ] + process.send_signal(signal.SIGKILL) + process.wait(timeout=5) + + lock_paths = [ + runtime_home / "hegi" / "daemon.lock", + runtime_home / "gateway.lock", + ] + _wait_for_lock_release(lock_paths) + process_deadline = time.monotonic() + 2 + while time.monotonic() < process_deadline: + live_states = [] + for pid in holder_pids: + status_path = Path(f"/proc/{pid}/stat") + if status_path.exists(): + live_states.append( + status_path.read_text(encoding="utf-8").rpartition(") ")[2][0] + ) + if all(state == "Z" for state in live_states): + break + time.sleep(0.02) + for pid in holder_pids: + status_path = Path(f"/proc/{pid}/stat") + if status_path.exists(): + state = status_path.read_text(encoding="utf-8").rpartition(") ")[2][0] + assert state == "Z" + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + def test_rollback_dry_run_and_default_apply_preserve_live_data(tmp_path): repo = Path(__file__).resolve().parents[2] runtime_home, config_path, state_path = _runtime(tmp_path) @@ -511,7 +1000,8 @@ def test_migration_install_failure_keeps_legacy_plugin_and_complete_backup(tmp_p repo=repo, ) - assert result.returncode == 42 + assert result.returncode != 0 + assert "plugin installation failed" in result.stderr assert config_path.read_bytes() == config_before assert state_path.read_bytes() == state_before assert plugin.read_bytes() == plugin_before @@ -528,6 +1018,11 @@ def test_migration_rejects_hardlinked_backup_artifacts_before_install(tmp_path): repo = Path(__file__).resolve().parents[2] runtime_home, config_path, state_path = _runtime(tmp_path) environment, log_path = _fake_commands(tmp_path, repo, runtime_home) + restart_probe_result = tmp_path / "restart-lock-probe" + environment["HEGI_TEST_RESTART_HEGI_LOCK_PROBE"] = str( + runtime_home / "hegi" / "daemon.lock" + ) + environment["HEGI_TEST_RESTART_LOCK_RESULT"] = str(restart_probe_result) plugin = runtime_home / "plugins" / "hegi-telegram" os.link(plugin / "legacy.txt", plugin / "legacy-alias.txt") config_before = config_path.read_bytes() @@ -548,7 +1043,11 @@ def test_migration_rejects_hardlinked_backup_artifacts_before_install(tmp_path): assert state_path.read_bytes() == state_before assert (plugin / "legacy.txt").read_text(encoding="utf-8") == "legacy-plugin\n" assert (plugin / "legacy-alias.txt").read_text(encoding="utf-8") == "legacy-plugin\n" - assert "hermes " not in log_path.read_text(encoding="utf-8") + command_log = log_path.read_text(encoding="utf-8") + assert "hermes gateway stop" in command_log + assert "hermes gateway restart" in command_log + assert "hermes plugins install" not in command_log + assert restart_probe_result.read_text(encoding="utf-8") == "acquired\n" def test_migration_reads_the_exact_state_path_with_uri_metacharacters(tmp_path): diff --git a/tests/compatibility/test_read_only_diagnostic.py b/tests/compatibility/test_read_only_diagnostic.py new file mode 100644 index 0000000..a1a35a0 --- /dev/null +++ b/tests/compatibility/test_read_only_diagnostic.py @@ -0,0 +1,618 @@ +from __future__ import annotations + +import argparse +import json +import os +import shutil +import sqlite3 +import stat +import subprocess +import sys +from pathlib import Path + +import pytest +import yaml + +import hegi.cli as cli +from hegi.state import StateStore + + +def _runtime_snapshot(root: Path) -> dict[str, tuple[int, int, int, bytes]]: + snapshot: dict[str, tuple[int, int, int, bytes]] = {} + for path in sorted(candidate for candidate in root.rglob("*") if candidate.is_file()): + metadata = path.stat() + snapshot[path.relative_to(root).as_posix()] = ( + stat.S_IMODE(metadata.st_mode), + metadata.st_size, + metadata.st_mtime_ns, + path.read_bytes(), + ) + return snapshot + + +def _create_agent_database(path: Path, chat_id: str) -> None: + connection = sqlite3.connect(path) + try: + connection.executescript( + """ + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + chat_id TEXT NOT NULL, + chat_type TEXT, + user_id TEXT + ); + CREATE TABLE messages ( + id INTEGER PRIMARY KEY, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + timestamp REAL NOT NULL, + active INTEGER NOT NULL DEFAULT 1, + compacted INTEGER NOT NULL DEFAULT 0, + platform_message_id TEXT + ); + """ + ) + connection.execute( + "INSERT INTO sessions(id, chat_id, chat_type, user_id) VALUES (?, ?, ?, ?)", + (path.stem, chat_id, "group", "77"), + ) + connection.execute( + """ + INSERT INTO messages( + id, session_id, role, content, timestamp, platform_message_id + ) VALUES (?, ?, ?, ?, ?, ?) + """, + (1, path.stem, "user", "diagnostic sentinel", 1.0, f"{path.stem}-1"), + ) + connection.commit() + finally: + connection.close() + + +def _create_runtime(tmp_path: Path) -> tuple[Path, Path, Path]: + runtime = tmp_path / "runtime" + hegi_home = runtime / "hegi" + archive = hegi_home / "archive" + archive.mkdir(parents=True) + env_path = runtime / "curator.env" + env_path.write_text("TELEGRAM_BOT_TOKEN=123456789:" + "A" * 35 + "\n", encoding="utf-8") + env_path.chmod(0o600) + + chat_id = "-10077" + agent_paths = [runtime / "agent-a.db", runtime / "agent-b.db"] + for path in agent_paths: + _create_agent_database(path, chat_id) + + queue_root = runtime / "approval-queue" + forest_root = runtime / "memory-forest" + queue_root.mkdir() + forest_root.mkdir() + forbidden_call = tmp_path / "external-call-was-made" + executable = runtime / "must-not-run" + executable.write_text( + f"#!/usr/bin/env bash\ntouch {forbidden_call}\nexit 99\n", + encoding="utf-8", + ) + executable.chmod(0o700) + + config_path = hegi_home / "config.yaml" + config_path.write_text( + yaml.safe_dump( + { + "version": 1, + "enabled": True, + "telegram": { + "chat_id": chat_id, + "curator_env": str(env_path), + "enabled": True, + }, + "agents": [ + {"name": "agent-a", "db_path": str(agent_paths[0])}, + {"name": "agent-b", "db_path": str(agent_paths[1])}, + ], + "archive": {"local_spool": str(archive)}, + "memory": { + "enabled": True, + "auto_commit": False, + "auto_draft": False, + "require_professor_approval": True, + "professor_user_ids": ["77"], + "default_project": "diagnostic-test", + "approval": { + "cli": str(executable), + "queue_root": str(queue_root), + "forest_cli": str(executable), + "forest_root": str(forest_root), + "backup_script": str(executable), + }, + }, + }, + sort_keys=False, + ), + encoding="utf-8", + ) + config_path.chmod(0o600) + state = StateStore(hegi_home / "state.db") + state.add_dead_letter("preexisting", {}, "preexisting") + return runtime, config_path, forbidden_call + + +def _run_diagnostic( + repo: Path, runtime: Path, config_path: Path +) -> subprocess.CompletedProcess[str]: + environment = os.environ.copy() + environment["HERMES_HOME"] = str(runtime) + environment["PYTHONPATH"] = str(repo) + return subprocess.run( + [ + sys.executable, + "-m", + "hegi", + "--config", + str(config_path), + "run-once", + "--diagnostic", + ], + cwd=repo, + env=environment, + text=True, + capture_output=True, + timeout=20, + check=False, + ) + + +def test_run_once_diagnostic_is_strictly_read_only(tmp_path: Path) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, forbidden_call = _create_runtime(tmp_path) + before = _runtime_snapshot(runtime) + state_artifacts = sorted((runtime / "hegi").glob("state.db*")) + old_atime = 1_700_000_000_000_000_000 + for path in state_artifacts: + metadata = path.stat() + os.utime(path, ns=(old_atime, metadata.st_mtime_ns), follow_symlinks=False) + artifact_atimes = {path.name: path.stat().st_atime_ns for path in state_artifacts} + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 0, completed.stderr + payload = json.loads(completed.stdout) + assert payload["mode"] == "diagnostic" + assert payload["read_only"] is True + assert payload["ok"] is True + assert payload["config"] == {"valid": True, "error_count": 0} + assert payload["state"]["quick_check"] == "ok" + assert payload["state"]["schema_version"] == 8 + assert payload["state"]["dead_letters"] == 1 + assert payload["operating_files_unchanged"] is True + assert payload["external_calls"] == [] + assert {path.name: path.stat().st_atime_ns for path in state_artifacts} == artifact_atimes + assert _runtime_snapshot(runtime) == before + assert not forbidden_call.exists() + assert not (runtime / "hegi" / "runs.jsonl").exists() + + +def test_run_once_diagnostic_reads_wal_without_touching_live_sidecars( + tmp_path: Path, +) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + state_path = runtime / "hegi" / "state.db" + writer = sqlite3.connect(state_path) + try: + writer.execute("PRAGMA wal_autocheckpoint=0") + writer.execute( + """ + INSERT INTO dead_letter(kind, payload_json, error, created_at) + VALUES ('wal-only', '{}', 'wal-only', 2.0) + """ + ) + writer.commit() + assert Path(f"{state_path}-wal").is_file() + assert Path(f"{state_path}-shm").is_file() + before = _runtime_snapshot(runtime) + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 0, completed.stderr + payload = json.loads(completed.stdout) + assert payload["state"]["dead_letters"] == 2 + assert payload["operating_files_unchanged"] is True + assert _runtime_snapshot(runtime) == before + finally: + writer.close() + + +@pytest.mark.parametrize("schema_version", [0, 7, 9]) +def test_run_once_diagnostic_rejects_unsupported_schema_versions( + tmp_path: Path, schema_version: int +) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + connection = sqlite3.connect(runtime / "hegi" / "state.db") + try: + connection.execute(f"PRAGMA user_version={schema_version}") + connection.commit() + finally: + connection.close() + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["ok"] is False + assert payload["state"]["schema_version"] == schema_version + assert payload["state"]["schema_version_supported"] is False + assert payload["state"]["schema_ready"] is False + + +def test_run_once_diagnostic_rejects_missing_required_column(tmp_path: Path) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + connection = sqlite3.connect(runtime / "hegi" / "state.db") + try: + connection.execute("ALTER TABLE action_items DROP COLUMN item_json") + connection.commit() + finally: + connection.close() + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["ok"] is False + assert payload["state"]["schema_version_supported"] is True + assert payload["state"]["missing_column_count"] == 1 + assert payload["state"]["schema_ready"] is False + + +def test_run_once_diagnostic_rejects_missing_required_index(tmp_path: Path) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + connection = sqlite3.connect(runtime / "hegi" / "state.db") + try: + connection.execute("DROP INDEX idx_approval_jobs_idempotency") + connection.commit() + finally: + connection.close() + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["ok"] is False + assert payload["state"]["schema_version_supported"] is True + assert payload["state"]["missing_index_count"] == 1 + assert payload["state"]["schema_ready"] is False + + +def test_run_once_diagnostic_rejects_missing_autoincrement_semantics( + tmp_path: Path, +) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + connection = sqlite3.connect(runtime / "hegi" / "state.db") + try: + schema_version = int(connection.execute("PRAGMA schema_version").fetchone()[0]) + connection.execute("PRAGMA writable_schema=ON") + connection.execute( + """ + UPDATE sqlite_master + SET sql=REPLACE(sql, ' AUTOINCREMENT', '') + WHERE type='table' AND name='dead_letter' + """ + ) + connection.execute("PRAGMA writable_schema=OFF") + connection.execute(f"PRAGMA schema_version={schema_version + 1}") + connection.commit() + finally: + connection.close() + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["state"]["missing_table_semantic_count"] == 1 + assert payload["state"]["schema_ready"] is False + + +def test_run_once_diagnostic_preserves_string_default_semantics(tmp_path: Path) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + connection = sqlite3.connect(runtime / "hegi" / "state.db") + try: + schema_version = int(connection.execute("PRAGMA schema_version").fetchone()[0]) + connection.execute("PRAGMA writable_schema=ON") + changed = connection.execute( + """ + UPDATE sqlite_master + SET sql=REPLACE(sql, "DEFAULT 'received'", "DEFAULT 'RECEIVED'") + WHERE type='table' AND name='approval_jobs' + """ + ) + assert changed.rowcount == 1 + connection.execute("PRAGMA writable_schema=OFF") + connection.execute(f"PRAGMA schema_version={schema_version + 1}") + connection.commit() + finally: + connection.close() + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["state"]["incompatible_column_count"] == 1 + assert payload["state"]["schema_ready"] is False + + +def test_run_once_diagnostic_rejects_foreign_key_orphans(tmp_path: Path) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + connection = sqlite3.connect(runtime / "hegi" / "state.db") + try: + connection.execute( + """ + INSERT INTO approval_transitions(job_id, state, details_json, created_at) + VALUES (999999, 'orphaned', '{}', 1.0) + """ + ) + connection.commit() + finally: + connection.close() + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["state"]["schema_ready"] is True + assert payload["state"]["foreign_key_violation_count"] == 1 + assert payload["state"]["integrity_ready"] is False + assert payload["ok"] is False + + +def test_run_once_diagnostic_separates_unsafe_artifacts_from_busy_state( + tmp_path: Path, +) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + foreign = tmp_path / "foreign-wal" + foreign.write_bytes(b"not a wal") + Path(f"{runtime / 'hegi' / 'state.db'}-wal").symlink_to(foreign) + + completed = _run_diagnostic(repo, runtime, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["error"] == "hegi-state-unsafe" + assert payload["operating_files_unchanged"] is None + + +def test_run_once_diagnostic_rejects_ancestor_directory_symlink(tmp_path: Path) -> None: + repo = Path(__file__).resolve().parents[2] + runtime, _config_path, _forbidden_call = _create_runtime(tmp_path / "real") + alias = tmp_path / "runtime-alias" + alias.symlink_to(runtime, target_is_directory=True) + config_path = alias / "hegi" / "config.yaml" + raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) + raw["state_db"] = str(alias / "hegi" / "state.db") + (runtime / "hegi" / "config.yaml").write_text( + yaml.safe_dump(raw, sort_keys=False), encoding="utf-8" + ) + (runtime / "hegi" / "config.yaml").chmod(0o600) + + completed = _run_diagnostic(repo, alias, config_path) + + assert completed.returncode == 1, completed.stderr + payload = json.loads(completed.stdout) + assert payload["error"] == "hegi-state-unsafe" + + +def test_run_once_diagnostic_rejects_directory_rebinding( + tmp_path: Path, monkeypatch +) -> None: + runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + replacement = runtime / "hegi-replacement" + replacement.mkdir() + shutil.copy2(config_path, replacement / "config.yaml") + replacement_state = StateStore(replacement / "state.db") + replacement_state.add_dead_letter("replacement", {}, "replacement") + config, directory_fd, config_fingerprint = cli._load_diagnostic_config(config_path) + assert directory_fd is not None + original_summary = cli._diagnostic_state_summary + + def replace_directory(snapshot: Path) -> dict: + current = runtime / "hegi" + current.rename(runtime / "hegi-original") + replacement.rename(current) + return original_summary(snapshot) + + monkeypatch.setattr(cli, "_diagnostic_state_summary", replace_directory) + try: + payload, return_code = cli._diagnostic_payload( + config, + directory_fd=directory_fd, + config_fingerprint=config_fingerprint, + ) + finally: + os.close(directory_fd) + + assert return_code == 4 + assert payload["error"] == "hegi-state-changing" + assert payload["operating_files_unchanged"] is False + + +def test_run_once_diagnostic_rejects_config_rebinding(tmp_path: Path) -> None: + _runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + config, directory_fd, config_fingerprint = cli._load_diagnostic_config(config_path) + assert directory_fd is not None + replacement = config_path.with_name("replacement.yaml") + raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) + raw["enabled"] = False + replacement.write_text(yaml.safe_dump(raw, sort_keys=False), encoding="utf-8") + replacement.chmod(0o600) + os.replace(replacement, config_path) + + try: + payload, return_code = cli._diagnostic_payload( + config, + directory_fd=directory_fd, + config_fingerprint=config_fingerprint, + ) + finally: + os.close(directory_fd) + + assert return_code == 4 + assert payload["error"] == "hegi-state-changing" + assert payload["operating_files_unchanged"] is False + + +@pytest.mark.skipif(shutil.which("hermes") is None, reason="Hermes CLI is not installed") +def test_public_hermes_diagnostic_path_is_strictly_read_only(tmp_path: Path) -> None: + repo = Path(__file__).resolve().parents[2] + runtime = tmp_path / "runtime" + plugin_root = runtime / "plugins" / "hermes-hegi" + plugin_root.mkdir(parents=True) + shutil.copy2(repo / "__init__.py", plugin_root / "__init__.py") + shutil.copy2(repo / "plugin.yaml", plugin_root / "plugin.yaml") + shutil.copytree( + repo / "hegi", + plugin_root / "hegi", + ignore=shutil.ignore_patterns("__pycache__", "*.pyc"), + ) + environment = os.environ.copy() + environment["HERMES_HOME"] = str(runtime) + environment["PYTHONDONTWRITEBYTECODE"] = "1" + environment["_HERMES_GATEWAY"] = "1" + hermes = str(shutil.which("hermes")) + enabled = subprocess.run( + [ + hermes, + "plugins", + "enable", + "hegi-telegram", + "--no-allow-tool-override", + ], + env=environment, + text=True, + capture_output=True, + timeout=30, + check=False, + ) + assert enabled.returncode == 0, enabled.stderr + + runtime, config_path, forbidden_call = _create_runtime(tmp_path) + before = _runtime_snapshot(runtime / "hegi") + queue_before = _runtime_snapshot(runtime / "approval-queue") + completed = subprocess.run( + [ + hermes, + "hegi", + "--config", + str(config_path), + "run-once", + "--diagnostic", + ], + env=environment, + text=True, + capture_output=True, + timeout=30, + check=False, + ) + + assert completed.returncode == 0, completed.stderr + payload = json.loads(completed.stdout) + assert payload["read_only"] is True + assert payload["operating_files_unchanged"] is True + assert payload["external_calls"] == [] + assert _runtime_snapshot(runtime / "hegi") == before + assert _runtime_snapshot(runtime / "approval-queue") == queue_before + assert not forbidden_call.exists() + + connection = sqlite3.connect(runtime / "hegi" / "state.db") + try: + connection.execute("PRAGMA user_version=7") + connection.commit() + finally: + connection.close() + invalid_before = _runtime_snapshot(runtime / "hegi") + failed = subprocess.run( + [ + hermes, + "hegi", + "--config", + str(config_path), + "run-once", + "--diagnostic", + ], + env=environment, + text=True, + capture_output=True, + timeout=30, + check=False, + ) + + assert failed.returncode == 1, failed.stderr + failed_payload = json.loads(failed.stdout) + assert failed_payload["ok"] is False + assert failed_payload["state"]["schema_version_supported"] is False + assert _runtime_snapshot(runtime / "hegi") == invalid_before + + +def test_run_once_diagnostic_never_constructs_the_pipeline( + tmp_path: Path, monkeypatch, capsys +) -> None: + _runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + + class ForbiddenPipeline: + def __init__(self, _config) -> None: + raise AssertionError("diagnostic constructed the stateful pipeline") + + monkeypatch.setattr(cli, "HegiPipeline", ForbiddenPipeline) + + result = cli.cmd_run_once( + argparse.Namespace(config=config_path, diagnostic=True, dry_run=False, send=False) + ) + + assert result == 0 + assert json.loads(capsys.readouterr().out)["external_calls"] == [] + + +def test_existing_dry_run_remains_a_stateful_pipeline_mode( + tmp_path: Path, monkeypatch, capsys +) -> None: + _runtime, config_path, _forbidden_call = _create_runtime(tmp_path) + observed: list[bool] = [] + + class RecordingPipeline: + def __init__(self, _config) -> None: + pass + + def run_once(self, *, dry_run: bool) -> list[dict]: + observed.append(dry_run) + return [] + + monkeypatch.setattr(cli, "HegiPipeline", RecordingPipeline) + monkeypatch.setattr( + cli, + "run_with_process_lock", + lambda _path, operation: (True, operation()), + ) + + result = cli.cmd_run_once( + argparse.Namespace(config=config_path, diagnostic=False, dry_run=True, send=False) + ) + + assert result == 0 + assert observed == [True] + assert json.loads(capsys.readouterr().out) == {"dry_run": True, "results": []} + + +def test_migration_postcheck_uses_the_read_only_diagnostic() -> None: + repo = Path(__file__).resolve().parents[2] + migration = (repo / "scripts" / "migrate-existing-install.sh").read_text(encoding="utf-8") + + assert "hermes hegi run-once --diagnostic" in migration + assert "hermes hegi run-once --dry-run" not in migration + assert migration.index("hermes hegi run-once --diagnostic") < migration.rindex( + "release_lock_holder hegi" + ) diff --git a/tests/integration/test_approval_integration.py b/tests/integration/test_approval_integration.py index 262cd8f..fe31cdc 100644 --- a/tests/integration/test_approval_integration.py +++ b/tests/integration/test_approval_integration.py @@ -643,6 +643,7 @@ def register_hook(self, name: str, callback: Any) -> None: def test_register_starts_worker_after_cli_hook_and_runtime(tmp_path, monkeypatch): _config(tmp_path, monkeypatch) + monkeypatch.setattr(gateway_plugin, "_is_gateway_process", lambda: True) calls: list[str] = [] class Context: diff --git a/tests/integration/test_daemon_lifecycle.py b/tests/integration/test_daemon_lifecycle.py index dd47df2..420ca41 100644 --- a/tests/integration/test_daemon_lifecycle.py +++ b/tests/integration/test_daemon_lifecycle.py @@ -13,6 +13,7 @@ def test_plugin_registration_owns_embedded_worker_lifecycle(tmp_path, monkeypatch): runtime_home = tmp_path / "runtime" monkeypatch.setenv("HERMES_HOME", str(runtime_home)) + monkeypatch.setattr(plugin, "_is_gateway_process", lambda: True) plugin.reset_plugin_state_for_tests() config = SimpleNamespace(enabled=True, state_db=tmp_path / "runtime" / "hegi" / "state.db") started: list[object] = [] @@ -33,6 +34,7 @@ def test_plugin_registration_owns_embedded_worker_lifecycle(tmp_path, monkeypatc assert cli_commands == ["hegi"] assert hooks == ["pre_gateway_dispatch"] assert started == [config] + assert not (runtime_home / "hegi" / "state.db").exists() assert not (runtime_home / "hegi" / "daemon.pid").exists() assert not (runtime_home / "hegi" / "daemon.ready").exists() repo = Path(__file__).resolve().parents[2] @@ -92,6 +94,71 @@ def test_embedded_worker_process_lock_prevents_duplicate_cycle(tmp_path, monkeyp assert calls == [config] +def test_embedded_worker_state_initialization_uses_process_lock(tmp_path, monkeypatch): + if locking.fcntl is None: + pytest.skip("POSIX process locking is unavailable") + config = SimpleNamespace(state_db=tmp_path / "hegi" / "state.db") + calls: list[object] = [] + monkeypatch.setattr(plugin, "_state_for", lambda value: calls.append(value)) + lock_path = config.state_db.parent / "daemon.lock" + lock_path.parent.mkdir(parents=True) + + with lock_path.open("a+", encoding="ascii") as external_lock: + locking.fcntl.flock( + external_lock.fileno(), + locking.fcntl.LOCK_EX | locking.fcntl.LOCK_NB, + ) + assert plugin._initialize_worker_state(config) is False + assert calls == [] + locking.fcntl.flock(external_lock.fileno(), locking.fcntl.LOCK_UN) + + assert plugin._initialize_worker_state(config) is True + assert calls == [config] + + +def test_embedded_worker_retries_initialization_before_normal_poll(monkeypatch): + class StopLoop(Exception): + pass + + class WakeEvent: + def __init__(self): + self.waits: list[int] = [] + + def wait(self, timeout): + self.waits.append(timeout) + + def clear(self): + return None + + config = SimpleNamespace( + enabled=True, + section=lambda _name: {"poll_seconds": 60}, + ) + wake = WakeEvent() + initialization_results = iter((False, True)) + calls: list[object] = [] + monkeypatch.setattr(plugin, "_WORKER_WAKE", wake) + monkeypatch.setattr(plugin, "_cached_config", lambda: config) + monkeypatch.setattr(plugin, "validate_config", lambda *_args, **_kwargs: []) + + def initialize(value): + calls.append(value) + return next(initialization_results) + + monkeypatch.setattr(plugin, "_initialize_worker_state", initialize) + monkeypatch.setattr( + plugin, + "_worker_iteration", + lambda: (_ for _ in ()).throw(StopLoop), + ) + + with pytest.raises(StopLoop): + plugin._worker_loop() + + assert calls == [config, config] + assert wake.waits == [1, 60] + + @pytest.mark.parametrize("alias_kind", ["symlink", "hardlink"]) def test_process_lock_rejects_inode_aliases(tmp_path, alias_kind): foreign = tmp_path / "foreign.lock"