Spaces:
Runtime error
Runtime error
File size: 3,310 Bytes
4e166c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | #!/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())
|