#!/usr/bin/env python3 """20 global stamps of the same hash; write a full HTTP + function timing log.""" from __future__ import annotations import hashlib import json import time from datetime import datetime, timezone from pathlib import Path from typing import Any from verae_mcp_timestamping_server.global_client import ( GlobalClient, UrllibHttp, load_auth_token, ) from verae_mcp_timestamping_server.ledger import validate_sha256 DIGEST = hashlib.sha256(b"/stamp").hexdigest() OUT = Path(__file__).resolve().parents[1] / "transaction-log-test.md" N = 20 def _now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") def _redact(headers: dict[str, str]) -> dict[str, str]: out = dict(headers) auth = out.get("Authorization") or out.get("authorization") if auth: key = "Authorization" if "Authorization" in out else "authorization" token = auth.split(" ", 1)[-1] if " " in auth else auth out[key] = f"Bearer [redacted len={len(token)}]" return out class LoggingHttp: def __init__(self, inner: UrllibHttp, exchanges: list[dict[str, Any]], spans: list[dict[str, Any]]): self.inner = inner self.exchanges = exchanges self.spans = spans def request( self, method: str, url: str, payload: dict[str, Any], headers: dict[str, str] ) -> Any: req_at = _now() t0 = time.perf_counter() res = self.inner.request(method, url, payload=payload, headers=headers) ms = (time.perf_counter() - t0) * 1000 resp_at = _now() self.exchanges.append( { "request_at": req_at, "response_at": resp_at, "method": method, "url": url, "request_headers": _redact(headers), "request_body": payload if method != "GET" else None, "status": res.status_code, "response_headers": dict(getattr(res, "headers", {}) or {}), "response_body": res.json() if hasattr(res, "json") else res.text, "response_text": res.text, "round_trip_ms": round(ms, 3), } ) self.spans.append( { "fn": f"http.{method}", "url": url, "ms": round(ms, 3), "at": req_at, } ) return res def _span(spans: list[dict[str, Any]], name: str, fn, **extra): t0 = time.perf_counter() at = _now() try: return fn() finally: spans.append({"fn": name, "ms": round((time.perf_counter() - t0) * 1000, 3), "at": at, **extra}) def run() -> dict[str, Any]: spans: list[dict[str, Any]] = [] exchanges: list[dict[str, Any]] = [] token = _span(spans, "load_auth_token", load_auth_token) client = _span( spans, "GlobalClient.from_env", lambda: GlobalClient( token=token, http=LoggingHttp(UrllibHttp(), exchanges, spans), ), ) digest = _span(spans, "validate_sha256", lambda: validate_sha256(DIGEST)) rounds: list[dict[str, Any]] = [] for i in range(1, N + 1): t0 = time.perf_counter() at = _now() n_ex = len(exchanges) try: rec = _span( spans, "GlobalClient.submit", lambda: client.submit(digest, cube_id="verae-mcp-research"), iteration=i, ) err = None receipt = rec.model_dump() except Exception as exc: # noqa: BLE001 — log live failures err = str(exc) receipt = None rounds.append( { "n": i, "at": at, "ms": round((time.perf_counter() - t0) * 1000, 3), "receipt": receipt, "error": err, "exchange_indexes": list(range(n_ex, len(exchanges))), } ) ids = [r["receipt"]["receipt_id"] for r in rounds if r["receipt"]] unique = sorted(set(ids)) return { "digest": digest, "n": N, "unique_receipt_ids": unique, "same_receipt_every_time": len(unique) == 1 and len(ids) == N, "rounds": rounds, "exchanges": exchanges, "spans": spans, } def render(data: dict[str, Any]) -> str: lines: list[str] = [] lines.append("# transaction-log-test") lines.append("") lines.append("20 global `/stamp` calls of the same SHA-256 with **no local cache**.") lines.append("Authorization bearer values are redacted (`len=` only).") lines.append("") lines.append(f"- Started (UTC of write): `{_now()}`") lines.append(f"- Digest: `{data['digest']}` (SHA-256 of bytes `/stamp`)") lines.append(f"- Unique `receipt_id` count: **{len(data['unique_receipt_ids'])}**") lines.append(f"- Same receipt every time: **{data['same_receipt_every_time']}**") lines.append("") lines.append("## Unique receipt ids") lines.append("") for rid in data["unique_receipt_ids"]: lines.append(f"- `{rid}`") if not data["unique_receipt_ids"]: lines.append("- *(none)*") lines.append("") lines.append("## Rounds") lines.append("") lines.append("| n | at (UTC) | submit ms | receipt_id | error |") lines.append("|---|---|---|---|---|") for r in data["rounds"]: rid = (r["receipt"] or {}).get("receipt_id", "") lines.append( f"| {r['n']} | {r['at']} | {r['ms']} | `{rid}` | {r['error'] or ''} |" ) lines.append("") lines.append("## Function profile") lines.append("") by_fn: dict[str, list[float]] = {} for s in data["spans"]: by_fn.setdefault(s["fn"], []).append(float(s["ms"])) lines.append("| function | calls | total ms | min | max | mean |") lines.append("|---|---|---|---|---|---|") for name, vals in sorted(by_fn.items()): total = sum(vals) lines.append( f"| `{name}` | {len(vals)} | {total:.3f} | {min(vals):.3f} | {max(vals):.3f} | {total/len(vals):.3f} |" ) lines.append("") lines.append("### Every internal span") lines.append("") lines.append("| at (UTC) | function | ms | extra |") lines.append("|---|---|---|---|") for s in data["spans"]: extra = {k: v for k, v in s.items() if k not in ("fn", "ms", "at")} extra_s = json.dumps(extra) if extra else "" lines.append(f"| {s['at']} | `{s['fn']}` | {s['ms']} | {extra_s} |") lines.append("") lines.append("## Network exchanges") lines.append("") for i, ex in enumerate(data["exchanges"], 1): lines.append(f"### Exchange {i}") lines.append("") lines.append("#### Request") lines.append("") lines.append(f"- Time/Date: `{ex['request_at']}`") lines.append(f"- METHOD: `{ex['method']}`") lines.append(f"- URL: `{ex['url']}`") lines.append("- Headers:") lines.append("") lines.append("```json") lines.append(json.dumps(ex["request_headers"], indent=2)) lines.append("```") lines.append("") lines.append("- Body:") lines.append("") lines.append("```json") lines.append(json.dumps(ex["request_body"], indent=2)) lines.append("```") lines.append("") lines.append("#### Response") lines.append("") lines.append(f"- Time/Date: `{ex['response_at']}`") lines.append(f"- Status: `{ex['status']}`") lines.append(f"- Round-trip: `{ex['round_trip_ms']} ms`") lines.append("- Headers:") lines.append("") lines.append("```json") lines.append(json.dumps(ex["response_headers"], indent=2)) lines.append("```") lines.append("") lines.append("- Body:") lines.append("") lines.append("```json") lines.append(json.dumps(ex["response_body"], indent=2) if not isinstance(ex["response_body"], str) else ex["response_text"]) lines.append("```") lines.append("") return "\n".join(lines) + "\n" def main() -> int: data = run() OUT.write_text(render(data)) print(f"wrote {OUT}") print("unique", len(data["unique_receipt_ids"]), "same", data["same_receipt_every_time"]) return 0 if __name__ == "__main__": raise SystemExit(main())