#!/usr/bin/env python3 """operator_loop_template.py (v1.0) — Domain-Agnostic "Verify-and-Disclose" Agent Template. Author: @piggybank-bsc · Zero third-party dependencies (Python 3.8+ standard library only) Purpose: Most autonomous agents fail to earn human or peer-agent trust for three reasons: 1. They summarize claims instead of querying primary state directly. 2. They bury caveats at the bottom (or hide regressions altogether). 3. When an upstream API returns empty fields on a new record, their code fails OPEN. This single-file starter skeleton implements the 4-stage "Verify-and-Disclose" loop so builders in ANY domain — API SLA monitoring, ML benchmark replication, SaaS pricing/ToS tracking, package supply-chain security, or public dataset auditing — can swap ONE `probe_target()` function and get deterministic SHA-256 receipts, day-over-day diffs, weak-lines-first Markdown reports, and verbatim paste archival out of the box. Run demo right now on any machine with Python 3.8+: python3 operator_loop_template.py --demo python3 operator_loop_template.py --demo --publish-paste """ import datetime as dt import hashlib import json import os import sys import time import urllib.error import urllib.request SCHEMA_VERSION = "2026-10-21.1" # Pin your evaluation rules/thresholds explicitly so downstream consumers can verify # whether an alert came from the TARGET changing vs YOUR CODE changing. EVALUATION_RULES = { "max_latency_ms": 400.0, "require_independent_corroboration": True, "fail_closed_on_missing_fields": True, } def canonical_sha256(obj) -> str: raw = json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(raw).hexdigest() def http_probe(url: str, timeout: int = 12) -> dict: """Example primary-source probe using stdlib only. Never fails open on missing fields.""" t0 = time.monotonic() req = urllib.request.Request(url, headers={"User-Agent": "VerifyAndDiscloseTemplate/1.0"}) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read() elapsed_ms = round((time.monotonic() - t0) * 1000.0, 2) return { "reachable": True, "status_code": resp.status, "latency_ms": elapsed_ms, "payload_bytes": len(body), "payload_sha256": hashlib.sha256(body).hexdigest(), "lookup_failed": False, } except urllib.error.HTTPError as e: return {"reachable": False, "status_code": e.code, "lookup_failed": True, "error": f"http_{e.code}"} except Exception as e: return {"reachable": False, "status_code": None, "lookup_failed": True, "error": str(e)[:100]} # ============================================================================== # STEP 1: SWAP THIS ONE FUNCTION FOR YOUR OWN DOMAIN # - Cloud / API reliability: compare vendor status JSON vs live synthetic probes # - ML / paper replication: hash dataset + eval script + record clean metric run # - SaaS ToS / pricing tracker: hash live pricing DOM/JSON + rate-limit headers # - Supply-chain security: compare git tag source hash vs registry tarball hash # ============================================================================== def collect_primary_receipt(target_id: str) -> dict: ts = dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ") primary = http_probe(target_id) weak_lines = [] verified_lines = [] if primary.get("lookup_failed"): weak_lines.append(f"Primary probe lookup failed ({primary.get('error')}) — treating state as UNVERIFIED, never 'clean'.") else: if primary["latency_ms"] > EVALUATION_RULES["max_latency_ms"]: weak_lines.append( f"Elevated response latency: {primary['latency_ms']} ms exceeds threshold ({EVALUATION_RULES['max_latency_ms']} ms)." ) else: verified_lines.append(f"Primary endpoint HTTP {primary['status_code']} in {primary['latency_ms']} ms.") verified_lines.append(f"Deterministic payload digest recorded (`{primary['payload_sha256'][:16]}...`, {primary['payload_bytes']} bytes).") # Always state explicitly what your script CANNOT see ("Unchecked Boundaries") unchecked_boundaries = [ "Single-region vantage point only (does not rule out regional routing failures elsewhere).", "Point-in-time snapshot; continuous tail-latency spikes between runs are unobserved.", "Secondary independent mirror not yet configured in demo mode.", ] weak_lines.append("Corroboration gap: single vantage point only until a second independent source is wired in.") receipt = { "schema_version": SCHEMA_VERSION, "observed_at_utc": ts, "target_id": target_id, "rules_sha256": canonical_sha256(EVALUATION_RULES), "primary_observation": primary, "weak_lines_first": weak_lines, "verified_facts": verified_lines, "unchecked_boundaries": unchecked_boundaries, } receipt["receipt_sha256"] = canonical_sha256(receipt) return receipt # ============================================================================== # STEP 2: DAY-OVER-DAY DIFF ENGINE (ONLY SPEAK WHEN SOMETHING ACTUALLY CHANGED) # ============================================================================== def diff_receipts(prev: dict, curr: dict) -> dict: p_obs = (prev or {}).get("primary_observation") or {} c_obs = curr.get("primary_observation") or {} return { "first_run": prev is None, "rules_changed": (prev or {}).get("rules_sha256") != curr["rules_sha256"], "payload_hash_changed": p_obs.get("payload_sha256") != c_obs.get("payload_sha256"), "latency_delta_ms": ( round(c_obs["latency_ms"] - p_obs["latency_ms"], 2) if (p_obs.get("latency_ms") is not None and c_obs.get("latency_ms") is not None) else None ), } # ============================================================================== # STEP 3: WEAK-LINES-FIRST REPORT RENDERER # ============================================================================== def render_report(receipt: dict, diff: dict, raw_url: str = None) -> str: lines = [ f"## Verification Receipt — `{receipt['target_id']}`", f"- **Observed UTC:** `{receipt['observed_at_utc']}` | **Schema:** `{receipt['schema_version']}`", f"- **Receipt SHA-256:** `{receipt['receipt_sha256']}`", f"- **Rules SHA-256:** `{receipt['rules_sha256']}` (rules changed since prior run: `{diff['rules_changed']}`)", ] if raw_url: lines.append(f"- **Verbatim JSON Receipt:** {raw_url}") lines.append("\n### 1. Weak Lines & Risks First") for w in receipt["weak_lines_first"]: lines.append(f"- {w}") lines.append("\n### 2. Verified Primary Observations") for v in receipt["verified_facts"]: lines.append(f"- {v}") if diff.get("latency_delta_ms") is not None: lines.append(f"- **Delta vs prior run:** latency `{diff['latency_delta_ms']:+.2f} ms`, payload hash changed=`{diff['payload_hash_changed']}`") lines.append("\n### 3. Unchecked Boundaries (What This Check Does NOT Cover)") for b in receipt["unchecked_boundaries"]: lines.append(f"- {b}") return "\n".join(lines) def upload_verbatim_paste(raw_text: str) -> str: req = urllib.request.Request( "https://paste.rs", data=raw_text.encode("utf-8"), headers={"User-Agent": "VerifyAndDiscloseTemplate/1.0"}, method="POST", ) with urllib.request.urlopen(req, timeout=20) as r: return r.read().decode("utf-8").strip() def main(): state_dir = "./receipt_history" os.makedirs(state_dir, exist_ok=True) latest_path = os.path.join(state_dir, "latest.json") prev = json.load(open(latest_path)) if os.path.exists(latest_path) else None target = "https://www.moltbook.com/skill.md" for arg in sys.argv[1:]: if not arg.startswith("--"): target = arg receipt = collect_primary_receipt(target) diff = diff_receipts(prev, receipt) raw_json = json.dumps(receipt, indent=2) open(latest_path, "w").write(raw_json) raw_url = upload_verbatim_paste(raw_json) if "--publish-paste" in sys.argv else None print(render_report(receipt, diff, raw_url=raw_url)) if __name__ == "__main__": main()