diff --git a/hermes-agent/patches/apply_response_transform_hook.py b/hermes-agent/patches/apply_response_transform_hook.py index c2263b2..fe68e31 100644 --- a/hermes-agent/patches/apply_response_transform_hook.py +++ b/hermes-agent/patches/apply_response_transform_hook.py @@ -1,17 +1,19 @@ #!/usr/bin/env python3 -"""Install the Hermes post-LLM response transform before persistence.""" +"""Install the managed Hermes response and Telegram name-provenance patches.""" from __future__ import annotations import argparse import ast import hashlib +import importlib.util import json import os import tempfile from pathlib import Path PATCH_VERSION = "teleo-response-transform-hook-v1" +RUNTIME_PATCH_VERSION = "teleo-hermes-runtime-patches-v2" PATCH_MARKER = " # TELEO_RESPONSE_TRANSFORM_HOOK_V1\n" START_ANCHOR = " # Save trajectory if enabled\n" END_ANCHOR = " # Extract reasoning from the last assistant message (if any)\n" @@ -137,12 +139,49 @@ def install(path: Path, *, check: bool) -> tuple[dict[str, str | bool], int]: }, 0 +def _load_telegram_name_policy_patcher(): + path = Path(__file__).with_name("apply_telegram_sender_name_policy.py") + spec = importlib.util.spec_from_file_location("teleo_telegram_sender_name_policy", path) + if not spec or not spec.loader: + raise RuntimeError(f"could not load Telegram name-policy patcher: {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def install_runtime(run_agent_path: Path, *, check: bool) -> tuple[dict[str, object], int]: + response_result, response_code = install(run_agent_path, check=check) + name_policy = _load_telegram_name_policy_patcher() + telegram_path = run_agent_path.parent / "gateway" / "platforms" / "telegram.py" + name_result, name_code = name_policy.install(telegram_path, check=check) + + if check: + changed = bool(response_code or name_code) + status = "patch_required" if changed else "installed" + code = 1 if changed else 0 + else: + changed = any( + result.get("status") == "installed_now" + for result in (response_result, name_result) + ) + status = "installed_now" if changed else "already_installed" + code = 0 + + return { + "patch_version": RUNTIME_PATCH_VERSION, + "status": status, + "agent_root": str(run_agent_path.parent), + "response_transform": response_result, + "telegram_sender_name_policy": name_result, + }, code + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("target", type=Path) parser.add_argument("--check", action="store_true") args = parser.parse_args() - result, code = install(args.target, check=args.check) + result, code = install_runtime(args.target, check=args.check) print(json.dumps(result, sort_keys=True)) return code diff --git a/hermes-agent/patches/apply_telegram_sender_name_policy.py b/hermes-agent/patches/apply_telegram_sender_name_policy.py new file mode 100644 index 0000000..f68bc96 --- /dev/null +++ b/hermes-agent/patches/apply_telegram_sender_name_policy.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Bind Hermes Telegram participant names to the current update's visible handle.""" + +from __future__ import annotations + +import argparse +import ast +import hashlib +import json +import os +import tempfile +from pathlib import Path + +PATCH_VERSION = "teleo-telegram-visible-handle-name-policy-v1" +PATCH_MARKER = " # TELEO_TELEGRAM_VISIBLE_HANDLE_NAME_POLICY_V1\n" +SOURCE_ANCHOR = " user_name=user.full_name if user else None,\n" + +REPLACEMENT = ''' # TELEO_TELEGRAM_VISIBLE_HANDLE_NAME_POLICY_V1 + # The current Telegram update is authoritative for the visible + # handle. The stable Telegram user_id continues to own session + # isolation; profile full_name is only a fallback when no public + # username exists on this update. + user_name=( + (user.username or "").strip().lstrip("@") + if user and (user.username or "").strip() + else user.full_name if user else None + ), +''' + + +def sha256_text(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def patch_source(source: str) -> tuple[str, bool]: + if PATCH_MARKER in source: + required = ( + '(user.username or "").strip().lstrip("@")', + "else user.full_name if user else None", + "user_id=str(user.id) if user else None", + ) + missing = [item for item in required if item not in source] + if missing: + raise ValueError(f"installed Telegram name-policy marker is incomplete: {missing}") + if SOURCE_ANCHOR in source: + raise ValueError("installed Telegram name-policy marker still contains the stale full-name anchor") + return source, False + + if source.count(SOURCE_ANCHOR) != 1: + raise ValueError("Hermes Telegram sender construction changed; refusing an unreviewed name-policy patch") + + patched = source.replace(SOURCE_ANCHOR, REPLACEMENT, 1) + ast.parse(patched) + return patched, True + + +def install(path: Path, *, check: bool) -> tuple[dict[str, str | bool], int]: + source = path.read_text(encoding="utf-8") + patched, changed = patch_source(source) + before_sha = sha256_text(source) + after_sha = sha256_text(patched) + if check: + return { + "patch_version": PATCH_VERSION, + "status": "patch_required" if changed else "installed", + "target": str(path), + "sha256": before_sha, + }, 1 if changed else 0 + + if changed: + mode = path.stat().st_mode & 0o777 + with tempfile.NamedTemporaryFile( + "w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False + ) as handle: + handle.write(patched) + temp_path = Path(handle.name) + try: + temp_path.chmod(mode) + os.replace(temp_path, path) + finally: + temp_path.unlink(missing_ok=True) + + ast.parse(path.read_text(encoding="utf-8")) + return { + "patch_version": PATCH_VERSION, + "status": "installed_now" if changed else "already_installed", + "target": str(path), + "before_sha256": before_sha, + "after_sha256": after_sha, + }, 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("target", type=Path) + parser.add_argument("--check", action="store_true") + args = parser.parse_args() + result, code = install(args.target, check=args.check) + print(json.dumps(result, sort_keys=True)) + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_hermes_response_transform_patch.py b/tests/test_hermes_response_transform_patch.py index 2bc7356..4e030a0 100644 --- a/tests/test_hermes_response_transform_patch.py +++ b/tests/test_hermes_response_transform_patch.py @@ -86,3 +86,34 @@ def test_install_check_reports_patch_required_then_installed(tmp_path: Path) -> after, after_code = module.install(target, check=True) assert after_code == 0 assert after["status"] == "installed" + + +def test_runtime_installer_manages_response_and_telegram_name_patches(tmp_path: Path) -> None: + module = load_patcher() + run_agent = tmp_path / "run_agent.py" + telegram = tmp_path / "gateway" / "platforms" / "telegram.py" + telegram.parent.mkdir(parents=True) + run_agent.write_text(upstream_fixture(), encoding="utf-8") + telegram.write_text( + '''def build_source(user): + return dict( + user_id=str(user.id) if user else None, + user_name=user.full_name if user else None, + ) +''', + encoding="utf-8", + ) + + before, before_code = module.install_runtime(run_agent, check=True) + assert before_code == 1 + assert before["status"] == "patch_required" + + applied, applied_code = module.install_runtime(run_agent, check=False) + assert applied_code == 0 + assert applied["status"] == "installed_now" + assert applied["response_transform"]["status"] == "installed_now" + assert applied["telegram_sender_name_policy"]["status"] == "installed_now" + + after, after_code = module.install_runtime(run_agent, check=True) + assert after_code == 0 + assert after["status"] == "installed" diff --git a/tests/test_hermes_telegram_sender_name_policy.py b/tests/test_hermes_telegram_sender_name_policy.py new file mode 100644 index 0000000..15388d6 --- /dev/null +++ b/tests/test_hermes_telegram_sender_name_policy.py @@ -0,0 +1,104 @@ +"""Tests for deterministic Telegram participant-name provenance in Hermes.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +PATCHER = ROOT / "hermes-agent" / "patches" / "apply_telegram_sender_name_policy.py" + + +def load_patcher(): + spec = importlib.util.spec_from_file_location("hermes_telegram_sender_name_policy", PATCHER) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def upstream_fixture() -> str: + return '''class Adapter: + def build_source(self, **values): + return values + + def build_event_source(self, user): + return self.build_source( + user_id=str(user.id) if user else None, + user_name=user.full_name if user else None, + ) +''' + + +def patched_adapter_class(): + module = load_patcher() + patched, changed = module.patch_source(upstream_fixture()) + assert changed is True + namespace: dict[str, object] = {} + exec(patched, namespace) + return namespace["Adapter"], module, patched + + +def test_current_visible_handle_overrides_stale_profile_name() -> None: + adapter_class, _module, _patched = patched_adapter_class() + user = SimpleNamespace(id=9070919, username="m3taversal", full_name="STALE-PERSONAL-NAME") + + source = adapter_class().build_event_source(user) + + assert "m3taversal" not in PATCHER.read_text(encoding="utf-8") + assert source == {"user_id": "9070919", "user_name": "m3taversal"} + assert "STALE-PERSONAL-NAME" not in source.values() + + +def test_each_participant_keeps_own_handle_and_stable_user_id() -> None: + adapter_class, _module, _patched = patched_adapter_class() + target = SimpleNamespace(id=9070919, username="m3taversal", full_name="STALE-PERSONAL-NAME") + other = SimpleNamespace(id=8181818, username="other_operator", full_name="Other Operator") + + first_target = adapter_class().build_event_source(target) + other_source = adapter_class().build_event_source(other) + restarted_target = adapter_class().build_event_source(target) + + assert first_target == restarted_target == {"user_id": "9070919", "user_name": "m3taversal"} + assert other_source == {"user_id": "8181818", "user_name": "other_operator"} + assert first_target["user_id"] != other_source["user_id"] + assert first_target["user_name"] != other_source["user_name"] + + +def test_full_name_is_only_the_no_username_fallback() -> None: + adapter_class, _module, _patched = patched_adapter_class() + user = SimpleNamespace(id=7171717, username=None, full_name="Visible Display Name") + + assert adapter_class().build_event_source(user) == { + "user_id": "7171717", + "user_name": "Visible Display Name", + } + + +def test_patch_is_idempotent_and_rejects_unknown_upstream_shape() -> None: + _adapter_class, module, patched = patched_adapter_class() + + assert module.patch_source(patched) == (patched, False) + with pytest.raises(ValueError, match="sender construction changed"): + module.patch_source("class Adapter:\n pass\n") + + +def test_install_check_reports_patch_required_then_installed(tmp_path: Path) -> None: + module = load_patcher() + target = tmp_path / "telegram.py" + target.write_text(upstream_fixture(), encoding="utf-8") + + before, before_code = module.install(target, check=True) + assert before_code == 1 + assert before["status"] == "patch_required" + + applied, applied_code = module.install(target, check=False) + assert applied_code == 0 + assert applied["status"] == "installed_now" + + after, after_code = module.install(target, check=True) + assert after_code == 0 + assert after["status"] == "installed"