diff --git a/scripts/record_leo_direct_claim_handler_suite.py b/scripts/record_leo_direct_claim_handler_suite.py
new file mode 100644
index 0000000..6d5c605
--- /dev/null
+++ b/scripts/record_leo_direct_claim_handler_suite.py
@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Record a retained direct-claim handler suite JSON emitted by the VPS runner."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+REPORT_DIR = ROOT / "docs" / "reports" / "leo-working-state-20260709"
+DEFAULT_OUT = REPORT_DIR / "telegram-handler-direct-claim-suite-current.json"
+
+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),
+]
+
+
+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 parse_report(raw: str) -> dict[str, Any]:
+    cleaned = redact(raw).strip()
+    for line in reversed(cleaned.splitlines()):
+        candidate = line.strip()
+        if not candidate.startswith("{"):
+            continue
+        return json.loads(candidate)
+    raise ValueError("stdin contained no JSON object line")
+
+
+def record_report(report: dict[str, Any], *, out: Path = DEFAULT_OUT) -> dict[str, Any]:
+    report = dict(report)
+    report["source_report_path"] = str(out)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return report
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    report = record_report(parse_report(sys.stdin.read()), out=args.out)
+    print(
+        json.dumps(
+            {
+                "out": str(args.out),
+                "pass_runtime": report.get("pass_runtime"),
+                "results": len(report.get("results") or []),
+                "posted_to_telegram": report.get("posted_to_telegram"),
+                "db_counts_changed": report.get("db_counts_changed"),
+                "temp_profile_removed": report.get("temp_profile_removed"),
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+    return 0 if report.get("pass_runtime") else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/run_leo_direct_claim_handler_remote.py b/scripts/run_leo_direct_claim_handler_remote.py
new file mode 100644
index 0000000..416b5a2
--- /dev/null
+++ b/scripts/run_leo_direct_claim_handler_remote.py
@@ -0,0 +1,306 @@
+#!/usr/bin/env python3
+"""Run the Working Leo direct-claim handler suite on the VPS.
+
+This script is intended to be executed on the VPS from the deploy-infra checkout
+as the ``teleo`` user. It does not post to Telegram and does not write to the
+production KB. It copies the live leoclean profile to a temporary profile,
+invokes GatewayRunner._handle_message with Telegram-shaped MessageEvents, and
+removes the temporary profile after the run.
+"""
+
+from __future__ import annotations
+
+import argparse
+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
+from typing import Any
+
+import working_leo_open_ended_benchmark as benchmark
+
+LIVE_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean")
+AGENT_ROOT = Path("/home/teleo/.hermes/hermes-agent")
+DEFAULT_CHAT_ID = "-5146042086"
+DEFAULT_USER_ID = "9070919"
+DEFAULT_USER_NAME = "codex handler direct claim"
+
+
+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 run_command(command: list[str], *, input_text: str | None = None) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(command, input=input_text, text=True, capture_output=True, check=False)
+
+
+def service_state() -> dict[str, str]:
+    proc = run_command(
+        [
+            "systemctl",
+            "show",
+            "leoclean-gateway.service",
+            "-p",
+            "ActiveState",
+            "-p",
+            "SubState",
+            "-p",
+            "MainPID",
+            "-p",
+            "NRestarts",
+            "-p",
+            "ExecMainStartTimestamp",
+            "--no-pager",
+        ]
+    )
+    state = {}
+    for line in proc.stdout.splitlines():
+        if "=" in line:
+            key, value = line.split("=", 1)
+            state[key] = value
+    return state
+
+
+def db_counts() -> dict[str, Any]:
+    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;
+"""
+    proc = run_command(
+        ["docker", "exec", "-i", "teleo-pg", "psql", "-U", "postgres", "-d", "teleo", "-At", "-q"],
+        input_text=sql,
+    )
+    if proc.returncode == 0 and proc.stdout.strip():
+        return json.loads(proc.stdout.strip())
+    return {"error": proc.stderr.strip() or proc.stdout.strip() or f"psql exited {proc.returncode}"}
+
+
+def copy_profile() -> Path:
+    tmp_root = Path(tempfile.mkdtemp(prefix="leo-direct-claim-handler-"))
+    target = tmp_root / "profile"
+
+    def ignore(_dir: str, names: list[str]) -> set[str]:
+        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: Path | None) -> bool:
+    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_path: Path, report: dict[str, Any]) -> None:
+    try:
+        report_path.write_text(json.dumps(report, sort_keys=True), encoding="utf-8")
+    except Exception:
+        pass
+
+
+async def run_suite(args: argparse.Namespace) -> dict[str, Any]:
+    before_service = service_state()
+    before_counts = db_counts()
+    temp_profile: Path | None = None
+    temp_removed = False
+    prompts = direct_claim_prompts()
+    report_path = Path(args.remote_report_path)
+    report: dict[str, Any] = {
+        "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": args.chat_id,
+            "chat_name": "Leo",
+            "chat_type": "group",
+            "user_id": args.user_id,
+            "user_name": args.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_path, 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=args.chat_id,
+            chat_name="Leo",
+            chat_type="group",
+            user_id=args.user_id,
+            user_name=args.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_path, 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_path, 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-{args.run_id}-{prompt['id']}",
+            )
+            started = datetime.now(timezone.utc)
+            reply = await asyncio.wait_for(runner._handle_message(event), timeout=args.turn_timeout)
+            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_path, 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_path, 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:
+            temp_removed = remove_tree(temp_profile.parent)
+        report["temp_profile_removed"] = temp_removed
+        write_report_snapshot(report_path, report)
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--chat-id", default=DEFAULT_CHAT_ID)
+    parser.add_argument("--user-id", default=DEFAULT_USER_ID)
+    parser.add_argument("--user-name", default=DEFAULT_USER_NAME)
+    parser.add_argument("--run-id", default=datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S"))
+    parser.add_argument("--turn-timeout", type=int, default=300)
+    parser.add_argument(
+        "--remote-report-path",
+        default="/tmp/leo-direct-claim-handler-report-current.json",
+        help="Temporary remote JSON snapshot path; overwritten on each run.",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    report = asyncio.run(run_suite(args))
+    print(json.dumps(report, sort_keys=True))
+    return 0 if report.get("pass_runtime") else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_record_leo_direct_claim_handler_suite.py b/tests/test_record_leo_direct_claim_handler_suite.py
new file mode 100644
index 0000000..4f65bf3
--- /dev/null
+++ b/tests/test_record_leo_direct_claim_handler_suite.py
@@ -0,0 +1,53 @@
+"""Tests for scripts/record_leo_direct_claim_handler_suite.py."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import record_leo_direct_claim_handler_suite as recorder  # noqa: E402
+import run_leo_direct_claim_handler_remote as remote_runner  # noqa: E402
+
+
+def sample_report() -> dict:
+    return {
+        "mode": "live_vps_gatewayrunner_temp_profile_direct_claim_suite",
+        "posted_to_telegram": False,
+        "changed_live_profile": False,
+        "mutates_kb_by_harness": False,
+        "pass_runtime": True,
+        "db_counts_changed": False,
+        "temp_profile_removed": True,
+        "results": [{"prompt_id": "DC-01", "reply": "ok", "ok": True}],
+    }
+
+
+def test_remote_runner_preserves_direct_claim_prompt_catalog():
+    prompts = remote_runner.direct_claim_prompts()
+
+    assert [prompt["id"] for prompt in prompts] == ["DC-01", "DC-02", "DC-03", "DC-04", "DC-05", "DC-06"]
+    assert prompts[0]["message"] == "Did we actually update the knowledge base, or is it still just proposals?"
+
+
+def test_recorder_parses_last_json_line_and_redacts_token():
+    raw = "noise\n" + json.dumps(sample_report() | {"secret": "token=123456:ABCdefghijklmnopqrstuvwxyz"}) + "\n"
+
+    parsed = recorder.parse_report(raw)
+
+    assert parsed["pass_runtime"] is True
+    assert parsed["secret"] == "token=[REDACTED_TOKEN]"
+
+
+def test_recorder_writes_report_with_source_path(tmp_path: Path):
+    out = tmp_path / "suite.json"
+
+    report = recorder.record_report(sample_report(), out=out)
+
+    assert report["source_report_path"] == str(out)
+    written = json.loads(out.read_text(encoding="utf-8"))
+    assert written["posted_to_telegram"] is False
+    assert written["temp_profile_removed"] is True