Add Leo direct-claim handler harness

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:

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:

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

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