mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-29 06:31:32 +00:00
feat(xai): apply_migration — rewrite config.yaml in-place via ruamel round-trip
Extends hermes_cli.xai_retirement with apply_migration(config_path,
issues, backup=True), used by the upcoming `hermes migrate xai`
sub-command.
Uses ruamel.yaml round-trip mode so that comments, key order,
indentation, quoting style, and scalar types are preserved on
rewrite — config.yaml is treated as a user-edited file, not a
data dump.
Behavior:
- Each issue rewrites parent[leaf] to issue.replacement
- When issue.reasoning_effort is set (non-reasoning variants
that map to grok-4.3), a sibling reasoning_effort key is
added/updated alongside the model
- Empty issues list or missing slots are no-ops (no backup,
no rewrite)
- When changes occur, a timestamped backup
(.bak-pre-migrate-xai-YYYYMMDD-HHMMSS) is written first
unless backup=False
17 unit tests cover dry-run/no-op, surgical replacement (each
slot), comment + key-order preservation, backup creation, and
idempotence (apply twice → no-op the second time).
This commit is contained in:
parent
a8a05c8ea7
commit
9ff98daf71
2 changed files with 344 additions and 0 deletions
|
|
@ -132,3 +132,124 @@ def format_issue(issue: RetirementIssue) -> str:
|
|||
if issue.note:
|
||||
parts.append(f"[note: {issue.note}]")
|
||||
return " ".join(parts)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Apply migration to config.yaml (round-trip preserves comments/order/types)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
import datetime as _dt
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ApplyResult:
|
||||
"""Outcome of an apply_migration call."""
|
||||
|
||||
file_path: Path
|
||||
backup_path: Optional[Path]
|
||||
issues_resolved: List[RetirementIssue]
|
||||
config_changed: bool
|
||||
|
||||
|
||||
def _walk_to_parent(yaml_doc: Any, dotted_path: str) -> "tuple[Any, str]":
|
||||
"""Resolve a dotted slot path to (parent_mapping, leaf_key).
|
||||
|
||||
Example: "auxiliary.vision.model" -> (yaml_doc["auxiliary"]["vision"], "model").
|
||||
Raises KeyError if any intermediate node is missing or not a mapping.
|
||||
"""
|
||||
parts = dotted_path.split(".")
|
||||
if len(parts) < 2:
|
||||
raise ValueError(f"Path must have at least one parent: {dotted_path!r}")
|
||||
node = yaml_doc
|
||||
for segment in parts[:-1]:
|
||||
if not isinstance(node, dict) or segment not in node:
|
||||
raise KeyError(f"Path segment {segment!r} missing in {dotted_path!r}")
|
||||
node = node[segment]
|
||||
return node, parts[-1]
|
||||
|
||||
|
||||
def apply_migration(
|
||||
config_path: Path,
|
||||
issues: List[RetirementIssue],
|
||||
backup: bool = True,
|
||||
) -> ApplyResult:
|
||||
"""Rewrite ``config_path`` in-place so each issue is resolved.
|
||||
|
||||
For every issue, the model name is replaced by ``issue.replacement``. If the
|
||||
issue has ``reasoning_effort`` set (i.e. the migration is from a
|
||||
``*-non-reasoning`` variant), a sibling ``reasoning_effort`` key is added
|
||||
or updated alongside the model.
|
||||
|
||||
Uses ``ruamel.yaml`` round-trip mode so comments, key order, indentation,
|
||||
and type literals (booleans, ints) are preserved.
|
||||
|
||||
A backup copy is written to
|
||||
``<config_path>.bak-pre-migrate-xai-YYYYMMDD-HHMMSS`` before rewriting,
|
||||
unless ``backup=False``.
|
||||
"""
|
||||
from ruamel.yaml import YAML # local import — avoid hard dep at module load
|
||||
|
||||
config_path = Path(config_path)
|
||||
if not config_path.exists():
|
||||
raise FileNotFoundError(config_path)
|
||||
|
||||
if not issues:
|
||||
return ApplyResult(
|
||||
file_path=config_path,
|
||||
backup_path=None,
|
||||
issues_resolved=[],
|
||||
config_changed=False,
|
||||
)
|
||||
|
||||
yaml = YAML(typ="rt")
|
||||
yaml.preserve_quotes = True
|
||||
with config_path.open("r", encoding="utf-8") as fh:
|
||||
doc = yaml.load(fh)
|
||||
|
||||
if doc is None:
|
||||
return ApplyResult(
|
||||
file_path=config_path,
|
||||
backup_path=None,
|
||||
issues_resolved=[],
|
||||
config_changed=False,
|
||||
)
|
||||
|
||||
resolved: List[RetirementIssue] = []
|
||||
for issue in issues:
|
||||
try:
|
||||
parent, leaf = _walk_to_parent(doc, issue.config_path)
|
||||
except KeyError:
|
||||
# Slot vanished between scan and apply — skip silently
|
||||
continue
|
||||
parent[leaf] = issue.replacement
|
||||
if issue.reasoning_effort:
|
||||
parent["reasoning_effort"] = issue.reasoning_effort
|
||||
resolved.append(issue)
|
||||
|
||||
if not resolved:
|
||||
return ApplyResult(
|
||||
file_path=config_path,
|
||||
backup_path=None,
|
||||
issues_resolved=[],
|
||||
config_changed=False,
|
||||
)
|
||||
|
||||
backup_path: Optional[Path] = None
|
||||
if backup:
|
||||
ts = _dt.datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
backup_path = config_path.with_name(
|
||||
f"{config_path.name}.bak-pre-migrate-xai-{ts}"
|
||||
)
|
||||
shutil.copy2(config_path, backup_path)
|
||||
|
||||
with config_path.open("w", encoding="utf-8") as fh:
|
||||
yaml.dump(doc, fh)
|
||||
|
||||
return ApplyResult(
|
||||
file_path=config_path,
|
||||
backup_path=backup_path,
|
||||
issues_resolved=resolved,
|
||||
config_changed=True,
|
||||
)
|
||||
|
|
|
|||
223
tests/hermes_cli/test_migrate_xai.py
Normal file
223
tests/hermes_cli/test_migrate_xai.py
Normal file
|
|
@ -0,0 +1,223 @@
|
|||
"""Tests for ``hermes migrate xai`` — apply path with ruamel round-trip."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.xai_retirement import (
|
||||
RetirementIssue,
|
||||
apply_migration,
|
||||
find_retired_xai_refs,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
def trap_config(tmp_path: Path) -> Path:
|
||||
"""A config.yaml with retired models AND comments to verify round-trip."""
|
||||
p = tmp_path / "config.yaml"
|
||||
p.write_text(
|
||||
"# Hermes config (sample)\n"
|
||||
"principal:\n"
|
||||
" provider: xai # the main model\n"
|
||||
" model: grok-4-1-fast-non-reasoning # retiring May 15\n"
|
||||
" temperature: 0.5\n"
|
||||
"auxiliary:\n"
|
||||
" vision:\n"
|
||||
" provider: xai\n"
|
||||
" model: grok-4-fast # retiring\n"
|
||||
" compression:\n"
|
||||
" provider: openai # not affected\n"
|
||||
" model: gpt-4o-mini\n"
|
||||
"delegation:\n"
|
||||
" model: grok-code-fast-1 # retiring\n"
|
||||
"plugins:\n"
|
||||
" image_gen:\n"
|
||||
" xai:\n"
|
||||
" model: grok-imagine-image-pro # retiring\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return p
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clean_config(tmp_path: Path) -> Path:
|
||||
p = tmp_path / "config.yaml"
|
||||
p.write_text(
|
||||
"principal:\n"
|
||||
" provider: xai\n"
|
||||
" model: grok-4.3\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return p
|
||||
|
||||
|
||||
def _parse(path: Path) -> dict:
|
||||
"""Load with ruamel for assertion convenience."""
|
||||
from ruamel.yaml import YAML
|
||||
yaml = YAML(typ="rt")
|
||||
with path.open("r", encoding="utf-8") as fh:
|
||||
return yaml.load(fh)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dry-run / no-op
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestNoOpPaths:
|
||||
def test_clean_config_returns_unchanged_result(self, clean_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(clean_config))
|
||||
assert issues == []
|
||||
result = apply_migration(clean_config, issues)
|
||||
assert result.config_changed is False
|
||||
assert result.backup_path is None
|
||||
# File untouched
|
||||
assert "grok-4.3" in clean_config.read_text(encoding="utf-8")
|
||||
|
||||
def test_empty_issues_list_is_noop(self, trap_config: Path):
|
||||
original = trap_config.read_text(encoding="utf-8")
|
||||
result = apply_migration(trap_config, issues=[])
|
||||
assert result.config_changed is False
|
||||
assert trap_config.read_text(encoding="utf-8") == original
|
||||
|
||||
def test_missing_file_raises(self, tmp_path: Path):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
apply_migration(tmp_path / "absent.yaml", issues=[
|
||||
RetirementIssue(
|
||||
config_path="principal.model",
|
||||
current_model="grok-4",
|
||||
replacement="grok-4.3",
|
||||
)
|
||||
])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Apply: surgical replacement
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyReplacement:
|
||||
def test_replaces_principal_model(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
result = apply_migration(trap_config, issues)
|
||||
assert result.config_changed is True
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["principal"]["model"] == "grok-4.3"
|
||||
|
||||
def test_adds_reasoning_effort_for_non_reasoning_variant(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
# Principal was grok-4-1-fast-non-reasoning → reasoning_effort: "none"
|
||||
assert cfg["principal"]["reasoning_effort"] == "none"
|
||||
|
||||
def test_replaces_auxiliary_vision(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["auxiliary"]["vision"]["model"] == "grok-4.3"
|
||||
|
||||
def test_replaces_delegation(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["delegation"]["model"] == "grok-4.3"
|
||||
|
||||
def test_replaces_image_gen_plugin(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["plugins"]["image_gen"]["xai"]["model"] == "grok-imagine-image-quality"
|
||||
|
||||
def test_does_not_touch_unrelated_slots(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
# auxiliary.compression was never xAI, must remain untouched
|
||||
assert cfg["auxiliary"]["compression"]["model"] == "gpt-4o-mini"
|
||||
assert cfg["auxiliary"]["compression"]["provider"] == "openai"
|
||||
# principal.temperature must survive
|
||||
assert cfg["principal"]["temperature"] == 0.5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Round-trip preservation (the hard part)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRoundTripPreservation:
|
||||
def test_preserves_top_of_file_comment(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
text = trap_config.read_text(encoding="utf-8")
|
||||
assert "# Hermes config (sample)" in text
|
||||
|
||||
def test_preserves_inline_comments_on_unmodified_lines(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
text = trap_config.read_text(encoding="utf-8")
|
||||
assert "# the main model" in text
|
||||
assert "# not affected" in text
|
||||
|
||||
def test_preserves_top_level_key_order(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
text = trap_config.read_text(encoding="utf-8")
|
||||
order = [
|
||||
text.index("principal:"),
|
||||
text.index("auxiliary:"),
|
||||
text.index("delegation:"),
|
||||
text.index("plugins:"),
|
||||
]
|
||||
assert order == sorted(order)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Backup behaviour
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestBackup:
|
||||
def test_backup_is_written_by_default(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
original = trap_config.read_text(encoding="utf-8")
|
||||
result = apply_migration(trap_config, issues)
|
||||
assert result.backup_path is not None
|
||||
assert result.backup_path.exists()
|
||||
assert result.backup_path.read_text(encoding="utf-8") == original
|
||||
|
||||
def test_backup_filename_prefixed(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
result = apply_migration(trap_config, issues)
|
||||
assert result.backup_path is not None
|
||||
assert result.backup_path.name.startswith("config.yaml.bak-pre-migrate-xai-")
|
||||
|
||||
def test_no_backup_when_disabled(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
result = apply_migration(trap_config, issues, backup=False)
|
||||
assert result.backup_path is None
|
||||
# No bak file in the directory
|
||||
assert not list(trap_config.parent.glob("*.bak-pre-migrate-xai-*"))
|
||||
|
||||
def test_no_backup_when_no_changes(self, clean_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(clean_config))
|
||||
result = apply_migration(clean_config, issues, backup=True)
|
||||
assert result.backup_path is None # nothing to back up
|
||||
assert not list(clean_config.parent.glob("*.bak-pre-migrate-xai-*"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Idempotence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestIdempotence:
|
||||
def test_apply_twice_is_safe(self, trap_config: Path):
|
||||
# First pass: replace
|
||||
issues_1 = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues_1)
|
||||
# Second pass: nothing to do
|
||||
issues_2 = find_retired_xai_refs(_parse(trap_config))
|
||||
assert issues_2 == []
|
||||
result_2 = apply_migration(trap_config, issues_2)
|
||||
assert result_2.config_changed is False
|
||||
Loading…
Add table
Add a link
Reference in a new issue