bin/v2/idempotency.py

#!/usr/bin/env python3
"""Surrogate-1 — idempotency keys (research §autonomous-24x7 pattern 2).

Every autonomous action computes idempotency_key = sha256(plan). If the
same key has been seen within the TTL, the action is treated as already-
applied and SKIPPED (preventing replay storms when the same anomaly fires
twice in a row). Records live in a JSONL ledger.

Ledger entry:
    {"key":"<sha256>", "ts":"...", "daemon":"sre|release", "outcome":"applied|queued"}

Usage:
    # Check if seen recently — exit 0 if seen (skip), 1 if new
    idempotency.py check --plan /path/to/plan.json --ttl-hours 4

    # Record after applying
    idempotency.py record --plan /path/to/plan.json \
        --daemon sre --outcome applied
"""
from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import sys
from pathlib import Path

LEDGER = Path(os.environ.get(
    "SURROGATE_IDEMPOTENCY_LEDGER",
    str(Path.home() / ".surrogate/state/idempotency.jsonl")))


def compute_key(plan_path: Path) -> str:
    txt = plan_path.read_text() if plan_path.is_file() else str(plan_path)
    h = hashlib.sha256()
    h.update(txt.encode())
    return h.hexdigest()


def load_ledger() -> list[dict]:
    if not LEDGER.exists():
        return []
    out = []
    for L in LEDGER.read_text().splitlines():
        try:
            out.append(json.loads(L))
        except Exception:
            continue
    return out


def append_ledger(rec: dict) -> None:
    LEDGER.parent.mkdir(parents=True, exist_ok=True)
    with LEDGER.open("a") as f:
        f.write(json.dumps(rec) + "\n")


def is_recent(key: str, ttl_hours: float) -> bool:
    cutoff = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=ttl_hours)
    for r in load_ledger():
        if r.get("key") != key:
            continue
        try:
            ts = dt.datetime.strptime(r["ts"], "%Y-%m-%dT%H:%M:%SZ")
        except Exception:
            continue
        if ts > cutoff:
            return True
    return False


def main() -> int:
    p = argparse.ArgumentParser()
    sp = p.add_subparsers(dest="cmd", required=True)

    pc = sp.add_parser("check")
    pc.add_argument("--plan", required=True)
    pc.add_argument("--ttl-hours", type=float, default=4.0)

    pr = sp.add_parser("record")
    pr.add_argument("--plan", required=True)
    pr.add_argument("--daemon", required=True)
    pr.add_argument("--outcome", required=True)

    pk = sp.add_parser("key")
    pk.add_argument("--plan", required=True)

    args = p.parse_args()

    if args.cmd == "key":
        print(compute_key(Path(args.plan)))
        return 0

    key = compute_key(Path(args.plan))

    if args.cmd == "check":
        seen = is_recent(key, args.ttl_hours)
        print(json.dumps({"key": key, "seen_recently": seen,
                          "ttl_hours": args.ttl_hours}))
        return 0 if seen else 1   # 0 = seen (skip); 1 = new (proceed)

    if args.cmd == "record":
        append_ledger({
            "key": key,
            "ts": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "daemon": args.daemon,
            "outcome": args.outcome,
        })
        print(f"recorded {key[:12]}…")
        return 0

    return 2


if __name__ == "__main__":
    sys.exit(main())