hermes-mcp/scripts/replay_decisions.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
if str(SRC) not in sys.path:
    sys.path.insert(0, str(SRC))

from hermes_mcp.replay import compare_to_baseline  # noqa: E402
from hermes_mcp.store import init_db  # noqa: E402
import sqlite3  # noqa: E402


def _load_rows(limit: int, concern_id: str | None) -> list[dict]:
    init_db()
    db_path = ROOT / "data" / "hermes_mcp.sqlite3"
    with sqlite3.connect(db_path) as conn:
        conn.row_factory = sqlite3.Row
        if concern_id:
            rows = conn.execute(
                "select * from decisions where concern_id = ? order by created_at desc limit ?",
                (concern_id, limit),
            ).fetchall()
        else:
            rows = conn.execute(
                "select * from decisions order by created_at desc limit ?",
                (limit,),
            ).fetchall()
    return [dict(r) for r in rows]


def main() -> int:
    parser = argparse.ArgumentParser(description="Replay stored Hermes decisions against the current decision engine.")
    parser.add_argument("--limit", type=int, default=20, help="How many stored decisions to replay")
    parser.add_argument("--concern-id", help="Optional concern id filter")
    parser.add_argument("--only-changed", action="store_true", help="Print only changed decisions")
    parser.add_argument("--json", action="store_true", help="Emit JSON lines instead of plain text")
    args = parser.parse_args()

    rows = _load_rows(limit=max(1, args.limit), concern_id=args.concern_id)
    checked = 0
    changed = 0
    skipped = 0

    for row in rows:
        payload = json.loads(row.get("target_policy_json") or "{}")
        replay_input = payload.get("replay_input") if isinstance(payload.get("replay_input"), dict) else None
        if not replay_input:
            skipped += 1
            continue

        result = compare_to_baseline(
            replay_input=replay_input,
            baseline={
                "mode": row.get("mode"),
                "action": row.get("action"),
                "target_strategy": row.get("target_strategy"),
            },
        )
        checked += 1
        changed += 1 if result["changed"] else 0
        if args.only_changed and not result["changed"]:
            continue

        output = {
            "decision_id": row.get("id"),
            "concern_id": row.get("concern_id"),
            "created_at": row.get("created_at"),
            **result,
        }
        if args.json:
            print(json.dumps(output, ensure_ascii=False))
        else:
            marker = "CHANGED" if result["changed"] else "same"
            print(f"[{marker}] {row.get('created_at')} concern={row.get('concern_id')} baseline={result['baseline']} replayed={result['replayed']}")

    summary = {
        "rows_loaded": len(rows),
        "checked": checked,
        "changed": changed,
        "skipped_without_replay_input": skipped,
    }
    if args.json:
        print(json.dumps({"summary": summary}, ensure_ascii=False))
    else:
        print(f"summary: {summary}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())