Enforce Telegram visible-handle name provenance

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
 

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())

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"

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"