#!/usr/bin/env python3 """Local evidence recorder and explicit TurnScope accounting/study client. Download client.py beside this file (legacy turnscope.py is also accepted). Only accounting and compare-study access the explicitly supplied server. All other commands are local. Run with --help for commands. """ import contextlib import copy import hashlib import importlib.util import json import os from pathlib import Path import re import stat import sys import time def _load_client(): for name in ("client.py", "turnscope.py"): path = Path(__file__).resolve().with_name(name) if path.is_file(): spec = importlib.util.spec_from_file_location("_turnscope_evidence_transport", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module raise ImportError("Download client.py beside evidence.py.") try: client = _load_client() except (ImportError, OSError): sys.stderr.write('{"error":{"code":"missing_client","message":"Download client.py beside evidence.py."}}\n') raise SystemExit(2) ClientError = client.ClientError fail = client.fail MAX_EVENTS = 5000 MAX_TIMESTAMP = 9_000_000_000_000_000 COVERAGE = {"events_complete", "model_calls_complete", "charges_complete", "dependencies_complete", "timing_complete"} STATUSES = {"completed", "failed", "cancelled", "timeout", "unknown"} RECORD_STATUSES = {"completed", "failed", "cancelled", "timeout", "evaluator_error"} PROVENANCE = {"manifest_id", "config_id", "config_revision", "task_id", "trial_id", "task_fingerprint", "evaluator_id", "scope_id", "environment_id"} LOCAL_FORMAT = "turnscope.study.local.v1" def fingerprint(value): """Hash UTF-8 text/bytes, or canonical JSON for structured tool arguments.""" if isinstance(value, str): data = value.encode("utf-8") elif isinstance(value, bytes): data = value else: data = client._json_bytes(value) if len(data) > client.MAX_INPUT: fail("input_too_large", "Local content exceeds the 16 MiB limit.") return "sha256:" + hashlib.sha256(data).hexdigest() def _fingerprint(value, code): if not isinstance(value, str) or not re.fullmatch(r"sha256:[a-f0-9]{64}", value): fail(code, "Fingerprints must be lowercase SHA-256 identifiers.") return value def _local_hash(item, content_key, fingerprint_key, code): if content_key in item: if content_key == "content" and not isinstance(item[content_key], str): fail(code, "Region content must be UTF-8 text.") digest = fingerprint(item.pop(content_key)) if item.get(fingerprint_key) is not None and item[fingerprint_key] != digest: fail(code, "A supplied fingerprint disagrees with locally hashed content.") item[fingerprint_key] = digest if item.get(fingerprint_key) is not None: _fingerprint(item[fingerprint_key], code) def _labels(item, keys, code, limit=128): for key in keys: if item.get(key) is not None: client._label(item[key], limit, code) def _integers(item, keys, code, maximum=client.MAX_TOKENS): for key in keys: if item.get(key) is not None: client._integer(item[key], maximum, code) def _array(value, maximum, code, minimum=0): result = client._array(value, code) if not minimum <= len(result) <= maximum: fail(code, "An array exceeds its supported bounds.") return result def _enum(value, allowed, code): if not isinstance(value, str) or value not in allowed: fail(code, "A field contains an unsupported choice.") def _bounded(document, maximum=client.MAX_REQUEST): if len(client._json_bytes(document)) > maximum: fail("request_too_large", "The sanitized document exceeds the 2 MiB request limit.") return document def _classes(items, value_key, code, *, rate=False): seen = set() for item in _array(items, 32, code): client._keys(item, {"class", value_key}, {"class", value_key}, code) name = client._label(item["class"], 128, code) if name in seen: fail(code, "Cache-write classes must be unique.") seen.add(name) if rate: client._number(item[value_key], 1_000_000, code) else: client._integer(item[value_key], client.MAX_TOKENS, code) def sanitize_accounting(document): """Allowlist the accounting wire schema and hash local-only content/arguments. This validates the local upload surface; graph, timing and pricing consistency are also checked by the accounting endpoint. No count, rate or coverage is inferred. """ code = "invalid_accounting" client._keys(document, {"version", "scope_id", "coverage", "root", "rate_catalog", "charges", "events"}, {"version", "scope_id", "events"}, code) if len(client._json_bytes(document)) > client.MAX_INPUT: fail("input_too_large", "Local evidence exceeds the 16 MiB input limit.") result = copy.deepcopy(document) if type(result["version"]) is not int or result["version"] != 1: fail(code, "Only accounting version 1 is supported.") client._label(result["scope_id"], 128, code) coverage = result.setdefault("coverage", {key: False for key in sorted(COVERAGE)}) client._keys(coverage, COVERAGE, code=code) if any(type(value) is not bool for value in coverage.values()): fail(code, "Coverage declarations must be explicit booleans.") if result.get("root") is not None: root = result["root"] client._keys(root, {"start_ms", "end_ms"}, {"start_ms", "end_ms"}, code) for key in root: client._integer(root[key], MAX_TIMESTAMP, code) if not 0 <= root["end_ms"] - root["start_ms"] <= client.MAX_METRIC: fail(code, "Root elapsed time is negative or exceeds its bound.") catalog = result.get("rate_catalog") if catalog is not None: client._keys(catalog, {"id", "currency", "source", "as_of", "cards"}, {"id", "currency", "source", "as_of", "cards"}, code) for key in ("id", "as_of"): client._label(catalog[key], 128, code) client._label(catalog["source"], 256, code) if not isinstance(catalog["currency"], str) or not re.fullmatch(r"[A-Z]{3}", catalog["currency"]): fail(code, "Currency must be three uppercase letters.") rates = {"input_per_million", "cache_read_per_million", "cache_write_per_million", "output_per_million"} pairs = set() for card in _array(catalog["cards"], 1000, code): client._keys(card, {"provider", "model", "cache_write_classes"} | rates, {"provider", "model"}, code) for key in ("provider", "model"): client._label(card[key], 128, code) pair = (card["provider"], card["model"]) if pair in pairs: fail(code, "Rate cards must have unique provider/model identities.") pairs.add(pair) for key in rates: if card.get(key) is not None: client._number(card[key], 1_000_000, code) if "cache_write_classes" in card: _classes(card["cache_write_classes"], "per_million", code, rate=True) seen_charges = set() for charge in _array(result.get("charges", []), 1000, code): client._keys(charge, {"id", "amount", "source"}, {"id", "source"}, code) identity = client._label(charge["id"], 128, code) if identity in seen_charges: fail(code, "Charge IDs must be unique.") seen_charges.add(identity) client._label(charge["source"], 256, code) if charge.get("amount") is not None: client._number(charge["amount"], client.MAX_METRIC, code) ids = set() edges = regions_count = models = 0 event_keys = {"id", "kind", "status", "usage_ownership", "provider", "model", "parent_id", "depends_on", "start_ms", "end_ms", "agent_id", "retry_of", "purpose", "usage", "regions", "tool"} for event in _array(result["events"], MAX_EVENTS, code, minimum=1): client._keys(event, event_keys, {"id", "kind", "status", "usage_ownership"}, code) identity = client._label(event["id"], 128, code) if identity in ids: fail(code, "Event IDs must be unique.") ids.add(identity) _enum(event["kind"], {"model", "tool", "wait", "container"}, code) _enum(event["status"], STATUSES, code) _enum(event["usage_ownership"], {"self", "inclusive", "none"}, code) models += event["kind"] == "model" _labels(event, ("provider", "model", "parent_id", "agent_id", "retry_of"), code) if event.get("purpose") is not None: _enum(event["purpose"], {"task", "compaction", "helper", "evaluation"}, code) _integers(event, ("start_ms", "end_ms"), code, MAX_TIMESTAMP) if event.get("start_ms") is not None and event.get("end_ms") is not None: if not 0 <= event["end_ms"] - event["start_ms"] <= client.MAX_METRIC: fail(code, "An event elapsed time is negative or exceeds its bound.") dependencies = _array(event.get("depends_on", []), 64, code) for dependency in dependencies: client._label(dependency, 128, code) if len(set(dependencies)) != len(dependencies): fail(code, "Event dependencies must be unique.") edges += len(dependencies) + int(event.get("parent_id") is not None) + int(event.get("retry_of") is not None) usage = event.get("usage") if usage is not None: counts = {"input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens"} client._keys(usage, counts | {"cache_write_classes"}, code=code) _integers(usage, counts, code) if "cache_write_classes" in usage: _classes(usage["cache_write_classes"], "tokens", code) if usage.get("input_tokens") is not None: if (usage.get("cache_read_tokens") or 0) + (usage.get("cache_write_tokens") or 0) > usage["input_tokens"]: fail(code, "Cache subsets exceed inclusive input usage.") if usage.get("reasoning_tokens") is not None and usage.get("output_tokens") is not None: if usage["reasoning_tokens"] > usage["output_tokens"]: fail(code, "Reasoning tokens exceed inclusive output usage.") regions = event.get("regions") if regions is not None: client._keys(regions, {"actual_order", "complete", "items"}, {"actual_order", "complete", "items"}, code) if any(type(regions[key]) is not bool for key in ("actual_order", "complete")): fail(code, "Region ordering and coverage must be explicit booleans.") items = _array(regions["items"], 128, code) regions_count += len(items) for item in items: client._keys(item, {"fingerprint", "tokens", "content"}, code=code) _local_hash(item, "content", "fingerprint", code) _fingerprint(item.get("fingerprint"), code) _integers(item, ("tokens",), code) tool = event.get("tool") if tool is not None: client._keys(tool, {"name", "operation", "arguments_fingerprint", "source_revision", "permission_scope", "arguments"}, {"name", "operation"}, code) client._label(tool["name"], 128, code) _enum(tool["operation"], {"read", "search", "other"}, code) _labels(tool, ("source_revision", "permission_scope"), code) _local_hash(tool, "arguments", "arguments_fingerprint", code) if models > 1000 or edges > 20_000 or regions_count > 20_000: fail(code, "Model, graph, or region totals exceed the supported limits.") return _bounded(result) class Recorder: """Explicit local instrumentation; callers supply usage, timings and coverage.""" def __init__(self, scope_id, *, coverage=None, root=None, rate_catalog=None, charges=None): self._document = {"version": 1, "scope_id": scope_id, "events": [], "coverage": {key: False for key in sorted(COVERAGE)}} for key, value in (("coverage", coverage), ("root", root), ("rate_catalog", rate_catalog), ("charges", charges)): if value is not None: self._document[key] = copy.deepcopy(value) def add_event(self, event): candidate = copy.deepcopy(self._document) candidate["events"].append(event) self._document = sanitize_accounting(candidate) def document(self): return sanitize_accounting(self._document) def write(self, path): client._write(os.fspath(path), self.document()) def normalize_bytes(data, format, *, scope_id, trace_id=None, service=None, scope=None): """Import selected OTLP calls; root duration and completeness remain unknown. Exact nanosecond times stay available in client.normalize_bytes' opt-in call metadata. Accounting uses integer millisecond offsets, rounded down, from the earliest selected call. This origin is not a root start or a scheduling edge. """ imported = client.normalize_bytes(data, format, trace_id, service, scope, include_call_metadata=True) calls = imported["import"]["calls"] origin = min(int(call["start_time_unix_nano"]) for call in calls) selected = {call["id"] for call in calls} events = [] for call in calls: event = {"id": call["id"], "kind": "model", "status": call["status"], "usage_ownership": "self", "start_ms": (int(call["start_time_unix_nano"]) - origin) // 1_000_000, "end_ms": (int(call["end_time_unix_nano"]) - origin) // 1_000_000, "usage": {key: call[key] for key in ("input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens")}} for key in ("provider", "model"): if call[key] is not None: event[key] = call[key] if call["parent_id"] in selected: event["parent_id"] = call["parent_id"] events.append(event) return sanitize_accounting({"version": 1, "scope_id": scope_id, "events": events}) def validate_manifest(document): code = "invalid_study" keys = {"manifest_id", "evaluator_id", "scope_id", "environment_id", "baseline", "candidate", "tasks"} client._keys(document, keys, keys, code) result = copy.deepcopy(document) for key in ("manifest_id", "evaluator_id", "scope_id", "environment_id"): client._label(result[key], 128, code) configurations = set() for arm in ("baseline", "candidate"): config = result[arm] keys = {"config_id", "revision", "models", "intervention"} client._keys(config, keys, keys, code) for key in ("config_id", "revision"): client._label(config[key], 128, code) client._label(config["intervention"], 1024, code) models = _array(config["models"], 32, code) for model in models: client._label(model, 128, code) if len(set(models)) != len(models) or config["config_id"] in configurations: fail(code, "Configuration IDs and models within a configuration must be unique.") configurations.add(config["config_id"]) tasks = set() expected = 0 for task in _array(result["tasks"], 1000, code, minimum=1): client._keys(task, {"task_id", "task_fingerprint", "content", "trial_ids"}, {"task_id", "trial_ids"}, code) identity = client._label(task["task_id"], 128, code) if identity in tasks: fail(code, "Manifest task IDs must be unique.") tasks.add(identity) _local_hash(task, "content", "task_fingerprint", code) _fingerprint(task.get("task_fingerprint"), code) trials = _array(task["trial_ids"], 1000, code, minimum=1) for trial in trials: client._label(trial, 128, code) if len(set(trials)) != len(trials): fail(code, "Manifest trial IDs must be unique within each task.") expected += len(trials) if expected > 1000: fail(code, "A study may declare at most 1,000 task/trial pairs per arm.") return _bounded(result) def validate_record(document): code = "invalid_study" keys = PROVENANCE | {"status", "outcome", "metrics"} client._keys(document, keys, keys, code) for key in PROVENANCE - {"task_fingerprint"}: client._label(document[key], 128, code) _fingerprint(document["task_fingerprint"], code) _enum(document["status"], RECORD_STATUSES, code) _enum(document["outcome"], {"passed", "failed", "unknown"}, code) client._validate_metrics(document["metrics"], code) return copy.deepcopy(document) def validate_study(document): code = "invalid_study" client._keys(document, {"version", "objective", "max_regression_pct", "manifest", "records"}, {"version", "objective", "manifest", "records"}, code) if type(document["version"]) is not int or document["version"] != 1: fail(code, "Only study version 1 is supported.") _enum(document["objective"], client.METRICS, code) client._number(document.get("max_regression_pct", 0), 1000, code) result = copy.deepcopy(document) manifest = result["manifest"] = validate_manifest(result["manifest"]) configs = {manifest[arm]["config_id"] for arm in ("baseline", "candidate")} pairs = {(task["task_id"], trial) for task in manifest["tasks"] for trial in task["trial_ids"]} seen = set() for record in _array(result["records"], 2000, code): validate_record(record) if record["config_id"] not in configs or (record["task_id"], record["trial_id"]) not in pairs: fail("unexpected_record", "A record is outside the declared configurations or task/trial pairs.") if record["manifest_id"] != manifest["manifest_id"]: fail("unexpected_record", "A record refers to a different manifest.") key = (record["config_id"], record["task_id"], record["trial_id"]) if key in seen: fail("duplicate_record", "A configuration/task/trial record already exists.") seen.add(key) return _bounded(result) def make_record(report, provenance, *, status, outcome): """Attach external evaluation and execution claims without inferring success.""" client._keys(provenance, PROVENANCE, PROVENANCE, "invalid_study") if (not isinstance(report, dict) or report.get("kind") != "accounting_report" or type(report.get("version")) is not int or report["version"] != 1 or "measurement_metrics" not in report): fail("invalid_report", "A version-1 accounting report with measurement_metrics is required.") report_scope = client._label(report.get("scope_id"), 128, "invalid_report") if report_scope != provenance["scope_id"]: fail("invalid_report", "The accounting report scope must match the execution provenance scope.") result = dict(provenance, status=status, outcome=outcome, metrics=client._validate_metrics(report["measurement_metrics"], "invalid_report")) return validate_record(result) def _state_path(path): path = os.path.abspath(os.fspath(path)) return os.path.join(os.path.realpath(os.path.dirname(path)), os.path.basename(path)) def _safe_stat(path): try: info = os.lstat(path) except FileNotFoundError: return None if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: fail("unsafe_file", "Study and lock paths must be ordinary, unlinked files, not symbolic links.") if hasattr(os, "geteuid") and info.st_uid != os.geteuid(): fail("unsafe_file", "Study and lock files must belong to the current user.") return info @contextlib.contextmanager def _study_lock(path): """Stable sibling lock survives atomic study replacement and concurrent writers.""" descriptor = None locked = False try: _safe_stat(path) lock_path = path + ".lock" _safe_stat(lock_path) descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600) info = os.fstat(descriptor) if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: fail("unsafe_file", "The study lock is not an ordinary private file.") os.fchmod(descriptor, 0o600) if os.name == "nt": import msvcrt if info.st_size == 0: os.write(descriptor, b"\0") os.lseek(descriptor, 0, os.SEEK_SET) acquire = lambda: msvcrt.locking(descriptor, msvcrt.LK_NBLCK, 1) release = lambda: msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) else: import fcntl acquire = lambda: fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) release = lambda: fcntl.flock(descriptor, fcntl.LOCK_UN) deadline = time.monotonic() + 10 while True: try: acquire() locked = True break except OSError: if time.monotonic() >= deadline: fail("study_locked", "The study is busy; retry after the current writer finishes.") time.sleep(0.05) _safe_stat(path) yield except OSError: fail("file_error", "The private study or its lock could not be accessed.") finally: if descriptor is not None: if locked: release() os.close(descriptor) def _read_study(path): info = _safe_stat(path) if info is None: fail("input_error", "Initialize a study before recording executions.") if stat.S_IMODE(info.st_mode) & 0o077: fail("unsafe_file", "Study files must have owner-only permissions.") if info.st_size > client.MAX_REQUEST + 2048: fail("input_too_large", "The local study exceeds its bounded size.") descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(descriptor, "rb") as stream: data = stream.read(client.MAX_REQUEST + 2049) if len(data) > client.MAX_REQUEST + 2048: fail("input_too_large", "The local study exceeds its bounded size.") stored = client._loads(data) keys = {"format", "manifest_sha256", "study"} client._keys(stored, keys, keys, "invalid_study") if stored["format"] != LOCAL_FORMAT: fail("invalid_study", "Expected a locally initialized study file.") study = client._object(stored["study"], "invalid_study") if "manifest" not in study or stored["manifest_sha256"] != fingerprint(client._json_bytes(study["manifest"])): fail("manifest_changed", "The expected manifest changed after study initialization.") return validate_study(study) def _write_study(path, study): client._write(path, {"format": LOCAL_FORMAT, "manifest_sha256": fingerprint(client._json_bytes(study["manifest"])), "study": study}) def init_study(path, manifest, *, objective, max_regression_pct=0): """Save the expected manifest with no execution records; never overwrite it.""" study = validate_study({"version": 1, "objective": objective, "max_regression_pct": max_regression_pct, "manifest": manifest, "records": []}) path = _state_path(path) with _study_lock(path): if os.path.lexists(path): fail("study_exists", "The study already exists; initialize a new path for another experiment.") _write_study(path, study) def load_study(path): path = _state_path(path) with _study_lock(path): return _read_study(path) def append_record(path, record): record = validate_record(record) path = _state_path(path) with _study_lock(path): study = _read_study(path) study["records"].append(record) _write_study(path, validate_study(study)) def _parser(): parser = client._Parser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) for name, help_text in (("record", "Hash local event content and arguments; no network"), ("accounting", "Submit sanitized accounting to an explicit server"), ("compare-study", "Submit a checked local study or explicit wire study")): command = commands.add_parser(name, help=help_text) command.add_argument("input") command.add_argument("--output", required=True) if name != "record": command.add_argument("--server", required=True) if name == "compare-study": command.add_argument("--check", action="store_true") command.add_argument("--wire", action="store_true", help="Read an explicit wire study, without local manifest registration") command = commands.add_parser("normalize", help="Import selected OTLP model calls; no network") command.add_argument("input") command.add_argument("--format", choices=("genai", "openinference"), required=True) command.add_argument("--scope-id", required=True) for flag in ("trace-id", "service", "scope"): command.add_argument("--" + flag) command.add_argument("--output", required=True) command = commands.add_parser("study-init", help="Save the expected manifest before executions; no network") command.add_argument("manifest") command.add_argument("--objective", choices=sorted(client.METRICS), required=True) command.add_argument("--max-regression-pct", type=float, default=0) command.add_argument("--output", required=True) command = commands.add_parser("study-record", help="Append an explicit execution record under a local lock") command.add_argument("study") command.add_argument("record") command = commands.add_parser("study-export", help="Export the checked wire study; no network") command.add_argument("study") command.add_argument("--output", required=True) command = commands.add_parser("measure", help="Attach explicit external provenance/outcome to accounting metrics") command.add_argument("input") command.add_argument("--provenance", required=True) command.add_argument("--status", choices=sorted(RECORD_STATUSES), required=True) command.add_argument("--outcome", choices=("passed", "failed", "unknown"), required=True) command.add_argument("--output", required=True) return parser def main(argv=None): try: args = _parser().parse_args(argv) exit_code = 0 if args.command in ("record", "accounting"): result = sanitize_accounting(client._loads(client._read(args.input))) if args.command == "accounting": result = client.post_json(args.server, "/api/v1/accounting", result) if "measurement_metrics" not in result: fail("invalid_report", "The accounting response lacks measurement_metrics.") client._validate_metrics(result["measurement_metrics"], "invalid_report") elif args.command == "normalize": result = normalize_bytes(client._read(args.input), args.format, scope_id=args.scope_id, trace_id=args.trace_id, service=args.service, scope=args.scope) elif args.command == "study-init": if args.output == "-": fail("usage_error", "Study initialization requires a private file path.") init_study(args.output, client._loads(client._read(args.manifest)), objective=args.objective, max_regression_pct=args.max_regression_pct) return 0 elif args.command == "study-record": append_record(args.study, client._loads(client._read(args.record))) return 0 elif args.command == "study-export": result = load_study(args.study) elif args.command == "measure": result = make_record(client._loads(client._read(args.input)), client._loads(client._read(args.provenance)), status=args.status, outcome=args.outcome) else: study = (validate_study(client._loads(client._read(args.input))) if args.wire else load_study(args.input)) result = client.post_json(args.server, "/api/v1/studies/compare", study) verdict = result.get("verdict") if verdict not in ("within_budget", "regression", "insufficient_evidence"): fail("invalid_response", "The study response lacks a supported verdict.") if args.check: exit_code = {"within_budget": 0, "regression": 1, "insufficient_evidence": 2}[verdict] client._write(args.output, result) return exit_code except ClientError as error: sys.stderr.write(json.dumps({"error": {"code": error.code, "message": error.message}}) + "\n") return 2 except (UnicodeError, RecursionError, TypeError, OverflowError, ValueError): sys.stderr.write('{"error":{"code":"invalid_document","message":"Unsupported or excessively nested document."}}\n') return 2 except OSError: sys.stderr.write('{"error":{"code":"file_error","message":"The local evidence file could not be accessed."}}\n') return 2 except KeyboardInterrupt: sys.stderr.write('{"error":{"code":"interrupted","message":"The operation was interrupted."}}\n') return 2 if __name__ == "__main__": sys.exit(main())