From 6fa86d5e01c647e13575d8e1fba016ee0668f4d8 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Fri, 10 Jul 2026 02:34:18 +0200 Subject: [PATCH] Add Leo direct-claim handler harness --- .../gcp/teleo-kb-bridge/SKILL.md | 14 + .../vps/teleo-kb-bridge/SKILL.md | 14 + scripts/run_leo_direct_claim_handler_suite.py | 430 ++++++++++++++++++ tests/test_hermes_leoclean_skill_surfaces.py | 4 + 4 files changed, 462 insertions(+) create mode 100755 scripts/run_leo_direct_claim_handler_suite.py diff --git a/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md index a7844a1..72bfe8f 100644 --- a/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md +++ b/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md @@ -71,6 +71,20 @@ If `search-proposals` finds an `approved` proposal with `applied_at` empty, say it is approved/staged or packet-ready but not canonical. Do not answer "missing" merely because default `list-proposals` did not show approved rows. +For these no-context direct claims, use this compact answer shape so Cory gets +the expected follow-up without needing to ask twice: + +1. Direct answer: yes/no/partly, with the truth ceiling. +2. Readback used: the exact bridge commands or row facts checked. +3. Canonical vs staged split: name `public.*`, `kb_stage.kb_proposals`, status, + proposal id, and `applied_at` when relevant. +4. Next Cory-style follow-up: the one proof-changing or admin action that would + change the answer. + +Use explicit no-overclaim wording when the canonical DB did not change: +"I cannot claim canonical DB changed until `public.*` readback plus +`applied_at`/postflight proof says it changed." + ## Telegram Rendering Make KB answers easy to scan in Telegram: diff --git a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md index f908cf4..ff08be8 100644 --- a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md +++ b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md @@ -63,6 +63,20 @@ If `search-proposals` finds an `approved` proposal with `applied_at` empty, say it is approved/staged or packet-ready but not canonical. Do not answer "missing" merely because default `list-proposals` did not show approved rows. +For these no-context direct claims, use this compact answer shape so Cory gets +the expected follow-up without needing to ask twice: + +1. Direct answer: yes/no/partly, with the truth ceiling. +2. Readback used: the exact bridge commands or row facts checked. +3. Canonical vs staged split: name `public.*`, `kb_stage.kb_proposals`, status, + proposal id, and `applied_at` when relevant. +4. Next Cory-style follow-up: the one proof-changing or admin action that would + change the answer. + +Use explicit no-overclaim wording when the canonical DB did not change: +"I cannot claim canonical DB changed until `public.*` readback plus +`applied_at`/postflight proof says it changed." + ## Telegram Rendering Make KB answers easy to scan in Telegram: diff --git a/scripts/run_leo_direct_claim_handler_suite.py b/scripts/run_leo_direct_claim_handler_suite.py new file mode 100755 index 0000000..011b174 --- /dev/null +++ b/scripts/run_leo_direct_claim_handler_suite.py @@ -0,0 +1,430 @@ +#!/usr/bin/env python3 +"""Run the Cory-style direct-claim suite through live VPS GatewayRunner. + +The harness does not post to Telegram and does not write to the production KB. +It copies the live leoclean profile to a temporary profile on the VPS, invokes +GatewayRunner._handle_message with Telegram-shaped MessageEvents, and removes +the temporary profile after the run. +""" + +from __future__ import annotations + +import json +import re +import subprocess +import uuid +from pathlib import Path +from typing import Any + +import working_leo_open_ended_benchmark as benchmark + +ROOT = Path(__file__).resolve().parents[1] +REPORT_DIR = ROOT / "docs" / "reports" / "leo-working-state-20260709" +OUTPUT_JSON = REPORT_DIR / "telegram-handler-direct-claim-suite-current.json" +SSH_KEY = Path.home() / ".ssh/livingip_hetzner_20260604_ed25519" +VPS = "root@77.42.65.182" + +CHAT_ID = "-5146042086" +USER_ID = "9070919" +USER_NAME = "codex handler direct claim" + +SECRET_PATTERNS = [ + re.compile(r"\b\d{6,}:[A-Za-z0-9_-]{20,}\b"), + re.compile(r"(Authorization: Bearer )([A-Za-z0-9._-]+)", re.IGNORECASE), + re.compile(r"((?:api[_-]?key|token|secret|password)[=:]\s*)([A-Za-z0-9._-]+)", re.IGNORECASE), +] + + +REMOTE_SCRIPT = r''' +import asyncio +import json +import os +import shutil +import subprocess +import sys +import tempfile +import traceback +from datetime import datetime, timezone +from pathlib import Path + +LIVE_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean") +AGENT_ROOT = Path("/home/teleo/.hermes/hermes-agent") +CHAT_ID = "__CHAT_ID__" +USER_ID = "__USER_ID__" +USER_NAME = "__USER_NAME__" +RUN_ID = "__RUN_ID__" +PROMPTS = __PROMPTS_JSON__ +REPORT_PATH = Path(f"/tmp/leo-direct-claim-handler-report-{RUN_ID}.json") + + +def service_state(): + raw = subprocess.check_output( + [ + "systemctl", + "show", + "leoclean-gateway.service", + "-p", + "ActiveState", + "-p", + "SubState", + "-p", + "MainPID", + "-p", + "NRestarts", + "-p", + "ExecMainStartTimestamp", + ], + text=True, + ) + result = {} + for line in raw.splitlines(): + if "=" in line: + key, value = line.split("=", 1) + result[key] = value + return result + + +def db_counts(): + sql = """ +select jsonb_build_object( + 'public.claims', (select count(*) from public.claims), + 'public.sources', (select count(*) from public.sources), + 'public.claim_evidence', (select count(*) from public.claim_evidence), + 'public.claim_edges', (select count(*) from public.claim_edges), + 'kb_stage.kb_proposals', (select count(*) from kb_stage.kb_proposals) +)::text; +""" + try: + raw = subprocess.check_output( + ["docker", "exec", "-i", "teleo-pg", "psql", "-U", "postgres", "-d", "teleo", "-At", "-q"], + input=sql, + text=True, + stderr=subprocess.STDOUT, + ) + return json.loads(raw.strip()) + except Exception as exc: + return {"error": str(exc)} + + +def copy_profile() -> Path: + tmp_root = Path(tempfile.mkdtemp(prefix="leo-direct-claim-handler-")) + target = tmp_root / "profile" + + def ignore(_dir, names): + ignored = { + "logs", + "cache", + "image_cache", + "memory_backups", + ".pytest_cache", + "__pycache__", + "gateway.pid", + "auth.lock", + } + return { + name + for name in names + if name in ignored + or name.endswith(".lock") + or ".bak-" in name + or name.endswith(".bak") + } + + try: + shutil.copytree(LIVE_PROFILE, target, symlinks=True, ignore=ignore) + except Exception: + shutil.rmtree(tmp_root, ignore_errors=True) + raise + return target + + +def remove_tree(path): + if not path or not path.exists(): + return True + try: + shutil.rmtree(path) + except Exception: + for root, dirs, files in os.walk(path, topdown=False): + for name in files: + p = Path(root) / name + try: + p.chmod(0o600) + p.unlink() + except Exception: + pass + for name in dirs: + p = Path(root) / name + try: + p.chmod(0o700) + p.rmdir() + except Exception: + pass + try: + path.chmod(0o700) + path.rmdir() + except Exception: + pass + return not path.exists() + + +def write_report_snapshot(report): + try: + REPORT_PATH.write_text(json.dumps(report, sort_keys=True), encoding="utf-8") + except Exception: + pass + + +async def run_suite(): + before_service = service_state() + before_counts = db_counts() + temp_profile = None + temp_removed = False + report = { + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "mode": "live_vps_gatewayrunner_temp_profile_direct_claim_suite", + "posted_to_telegram": False, + "changed_live_profile": False, + "mutates_kb_by_harness": False, + "source": { + "platform": "telegram", + "chat_id": CHAT_ID, + "chat_name": "Leo", + "chat_type": "group", + "user_id": USER_ID, + "user_name": USER_NAME, + }, + "harness_notes": [ + "Prompts are the exact DC-01..DC-06 direct-claim benchmark messages.", + "The harness uses a temporary copy of the leoclean profile and does not post to Telegram.", + "This proves handler-level live VPS GatewayRunner behavior, not human-visible Telegram delivery.", + ], + "db_counts_before": before_counts, + "before_service": before_service, + "remote_report_path": str(REPORT_PATH), + } + write_report_snapshot(report) + try: + temp_profile = copy_profile() + os.environ["HERMES_HOME"] = str(temp_profile) + os.environ["HOME"] = "/home/teleo" + sys.path.insert(0, str(AGENT_ROOT)) + + from gateway.config import Platform + from gateway.platforms.base import MessageEvent, MessageType + from gateway.run import GatewayRunner + from gateway.session import SessionSource + + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id=CHAT_ID, + chat_name="Leo", + chat_type="group", + user_id=USER_ID, + user_name=USER_NAME, + ) + runner = GatewayRunner() + session_key = runner._session_key_for_source(source) + authorized = runner._is_user_authorized(source) + report["handler"] = { + "temp_profile": str(temp_profile), + "session_key": session_key, + "authorized": authorized, + } + write_report_snapshot(report) + if not authorized: + report["results"] = [] + report["pass_runtime"] = False + report["failure"] = "telegram group source was not authorized by live config" + write_report_snapshot(report) + return report + + results = [] + for index, prompt in enumerate(PROMPTS, start=1): + event = MessageEvent( + text=prompt["message"], + message_type=MessageType.TEXT, + source=source, + message_id=f"direct-claim-{RUN_ID}-{prompt['id']}", + ) + started = datetime.now(timezone.utc) + reply = await asyncio.wait_for(runner._handle_message(event), timeout=300) + ended = datetime.now(timezone.utc) + results.append( + { + "turn": index, + "prompt_id": prompt["id"], + "dimension": prompt["dimension"], + "prompt": prompt["message"], + "reply": reply or "", + "started_at_utc": started.isoformat(), + "ended_at_utc": ended.isoformat(), + "ok": bool((reply or "").strip()), + "mutates_kb": False, + "evidence_tier": "live_vps_gatewayrunner_temp_profile", + "claim_ceiling": "Live VPS GatewayRunner reply from temp leoclean profile; no Telegram post; no production DB apply authorized.", + } + ) + report["results"] = results + write_report_snapshot(report) + report["results"] = results + report["pass_runtime"] = all(row["ok"] for row in results) + return report + except Exception as exc: + report["pass_runtime"] = False + report["error"] = str(exc) + report["traceback_tail"] = traceback.format_exc().splitlines()[-12:] + write_report_snapshot(report) + return report + finally: + after_counts = db_counts() + after_service = service_state() + report["db_counts_after"] = after_counts + report["db_counts_changed"] = after_counts != before_counts + report["service_before_after"] = { + "before": before_service, + "after": after_service, + "unchanged_from_preexisting_live_readback": before_service == after_service, + } + if temp_profile is not None: + tmp_root = temp_profile.parent + temp_removed = remove_tree(tmp_root) + report["temp_profile_removed"] = temp_removed + write_report_snapshot(report) + + +print(json.dumps(asyncio.run(run_suite()), sort_keys=True)) +''' + + +def direct_claim_prompts() -> list[dict[str, Any]]: + return [ + { + "id": prompt["id"], + "dimension": prompt["dimension"], + "message": prompt["message"], + } + for prompt in benchmark.CORY_DIRECT_CLAIM_FOLLOWUP_SCENARIOS + ] + + +def build_remote_script(run_id: str) -> str: + return ( + REMOTE_SCRIPT.replace("__CHAT_ID__", CHAT_ID) + .replace("__USER_ID__", USER_ID) + .replace("__USER_NAME__", USER_NAME) + .replace("__RUN_ID__", run_id) + .replace("__PROMPTS_JSON__", json.dumps(direct_claim_prompts())) + ) + + +def redact(text: str) -> str: + result = text + for pattern in SECRET_PATTERNS: + if pattern.groups >= 2: + result = pattern.sub(r"\1[REDACTED]", result) + else: + result = pattern.sub("[REDACTED_TOKEN]", result) + return result + + +def fetch_remote_report(run_id: str, *, unlink: bool = True) -> dict[str, Any] | None: + report_path = f"/tmp/leo-direct-claim-handler-report-{run_id}.json" + unlink_line = " p.unlink()\n" if unlink else "" + proc = subprocess.run( + [ + "ssh", + "-i", + str(SSH_KEY), + "-o", + "BatchMode=yes", + "-o", + "StrictHostKeyChecking=accept-new", + VPS, + "sudo -u teleo -H /home/teleo/.hermes/hermes-agent/venv/bin/python -", + ], + input=( + "import json\n" + "from pathlib import Path\n" + f"p = Path({report_path!r})\n" + "if p.exists():\n" + " print(p.read_text(encoding='utf-8'))\n" + f"{unlink_line}" + ), + text=True, + capture_output=True, + timeout=60, + ) + if proc.returncode != 0 or not proc.stdout.strip(): + return None + return json.loads(redact(proc.stdout.strip())) + + +def run_remote() -> dict[str, Any]: + run_id = uuid.uuid4().hex[:12] + proc = subprocess.run( + [ + "ssh", + "-i", + str(SSH_KEY), + "-o", + "BatchMode=yes", + "-o", + "StrictHostKeyChecking=accept-new", + VPS, + "cd /home/teleo && sudo -u teleo -H /home/teleo/.hermes/hermes-agent/venv/bin/python -", + ], + input=build_remote_script(run_id), + text=True, + capture_output=True, + timeout=2100, + ) + result: dict[str, Any] = { + "returncode": proc.returncode, + "stdout": redact(proc.stdout), + "stderr": redact(proc.stderr), + "run_id": run_id, + } + if proc.returncode == 0 and proc.stdout.strip(): + for line in reversed(proc.stdout.splitlines()): + candidate = line.strip() + if not candidate.startswith("{"): + continue + try: + result["parsed"] = json.loads(candidate) + break + except json.JSONDecodeError: + continue + if "parsed" not in result: + result["parse_error"] = "remote stdout contained no parseable JSON object line" + if "parsed" not in result: + fetched = fetch_remote_report(run_id) + if fetched is not None: + result["parsed"] = fetched + result["parsed_from_report_file"] = True + else: + fetch_remote_report(run_id) + return result + + +def write_output(remote: dict[str, Any]) -> dict[str, Any]: + REPORT_DIR.mkdir(parents=True, exist_ok=True) + parsed = remote.get("parsed") or {} + report = parsed if isinstance(parsed, dict) else {} + report["source_report_path"] = str(OUTPUT_JSON) + report["remote_returncode"] = remote.get("returncode") + report["remote_run_id"] = remote.get("run_id") + if remote.get("stderr"): + report["remote_stderr"] = remote["stderr"] + OUTPUT_JSON.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return report + + +def main() -> int: + remote = run_remote() + report = write_output(remote) + print(json.dumps({"json": str(OUTPUT_JSON), "pass_runtime": report.get("pass_runtime")}, indent=2)) + return 0 if remote.get("returncode") == 0 and report.get("pass_runtime") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_hermes_leoclean_skill_surfaces.py b/tests/test_hermes_leoclean_skill_surfaces.py index 6c57ce2..bdeaab0 100644 --- a/tests/test_hermes_leoclean_skill_surfaces.py +++ b/tests/test_hermes_leoclean_skill_surfaces.py @@ -14,6 +14,8 @@ def test_gcp_kb_skill_uses_cloudsql_bridge_not_vps_docker() -> None: assert "search-proposals" in text assert "decision-matrix-status" in text assert "Do not answer" in text or "do not infer matrix approval" in text + assert "Next Cory-style follow-up" in text + assert "I cannot claim canonical DB changed" in text assert "docker exec" not in text assert "teleo-pg" not in text assert "on the VPS" not in text @@ -33,6 +35,8 @@ def test_vps_kb_skill_keeps_vps_scope_explicit() -> None: assert "decision-matrix-status" in text assert "do not infer matrix approval" in text assert "approved/staged or packet-ready but not canonical" in text + assert "Next Cory-style follow-up" in text + assert "I cannot claim canonical DB changed" in text assert "Do not call an approved proposal \"implemented\"" in text assert "needs reviewer/operator apply tooling rather than inviting ad hoc SQL from chat" in squashed assert "Never end a normal Telegram answer by offering to run direct `INSERT`, `UPDATE`" in text