#!/usr/bin/env python3 """TurnScope's dependency-free, local telemetry importer and explicit API client. Run ``python3 client.py --help``. Normalization never accesses the network. Only analyze/compare contact the server supplied by the caller. """ import argparse import copy import datetime import email.utils import hashlib import http.client import ipaddress import json import math import os import re import sys import tempfile import time import unicodedata import urllib.error import urllib.parse import urllib.request MAX_INPUT = 16 * 1024 * 1024 MAX_REQUEST = 2 * 1024 * 1024 MAX_RESPONSE = 16 * 1024 * 1024 MAX_SPANS = 20_000 MAX_CALLS = 1_000 MAX_TOKENS = 1_000_000_000 MAX_METRIC = 1_000_000_000_000 MAX_ATTEMPTS = 3 MAX_RETRY_DELAY = 10 TIMEOUT = 10 INFERENCE_OPERATIONS = {"chat", "generate_content", "text_completion"} METRICS = {"input_tokens", "output_tokens", "turns", "duration_ms", "cost"} IMPORT_LIMITATIONS = [ "Accounting covers selected spans present in this export; completeness of the agent run is unknown.", "Input totals include cache reads and writes. Missing cache details remain unknown.", "Exact context-block tokens and complete tool selection are unavailable; neither is inferred from text.", "Calls are ordered by exact start timestamp, then span ID; this does not establish causal order.", "No pricing is imported. Cache-write premiums, tool fees, and omitted calls are not modeled.", "Use the same evaluation and accounting scope for paired measurements; labels and outcomes are caller claims.", ] class ClientError(Exception): """A bounded, static message safe for JSON error output.""" def __init__(self, code, message): super().__init__(message) self.code = code self.message = message def fail(code, message): raise ClientError(code, message) def _object(value, code="invalid_document"): if not isinstance(value, dict): fail(code, "Expected a JSON object.") return value def _array(value, code="invalid_document"): if not isinstance(value, list): fail(code, "Expected a JSON array.") return value def _keys(value, allowed, required=(), code="invalid_document"): _object(value, code) if set(value) - set(allowed) or set(required) - set(value): fail(code, "Required fields are missing or unsupported fields are present.") def _label(value, limit=128, code="invalid_measurement"): if (not isinstance(value, str) or not value.strip() or len(value.encode("utf-8")) > limit or any(unicodedata.category(ch) == "Cc" for ch in value)): fail(code, "Identifiers must be nonblank, within the byte limit, and contain no controls.") return value def _integer(value, maximum, code, wire=False): if wire and isinstance(value, str) and re.fullmatch(r"[0-9]+", value): # Prevent integer conversion of arbitrarily long digit strings. if len(value) > 24: fail(code, "Integer value exceeds the supported bound.") value = int(value) if type(value) is not int or not 0 <= value <= maximum: fail(code, "Expected a nonnegative integer within the supported bound.") return value def _number(value, maximum, code): if (type(value) not in (int, float) or not math.isfinite(value) or not 0 <= value <= maximum): fail(code, "Expected a finite nonnegative number within the supported bound.") return value def _json_pairs(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("duplicate key") result[key] = value return result def _invalid_constant(_value): raise ValueError("nonfinite number") def _loads(data, code="invalid_json"): try: return json.loads(data, object_pairs_hook=_json_pairs, parse_constant=_invalid_constant) except (ValueError, UnicodeError, RecursionError): fail(code, "Expected valid UTF-8 JSON with unique keys and finite numbers.") def _json_bytes(value): try: return json.dumps(value, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":")).encode("utf-8") except (ValueError, TypeError, UnicodeError, RecursionError): fail("invalid_json", "The document cannot be encoded as finite UTF-8 JSON.") def _attributes(value): result = {} for entry in _array(value, "invalid_attributes"): _object(entry, "invalid_attributes") key = entry.get("key") if not isinstance(key, str) or not key or key in result: fail("invalid_attributes", "OTLP attributes must have unique nonempty string keys.") result[key] = _object(entry.get("value"), "invalid_attributes") return result def _string_attribute(attributes, key, limit=256): value = attributes.get(key) if value is None: return None if set(value) != {"stringValue"}: fail("invalid_attributes", "An accounting attribute must have a stringValue.") return _label(value["stringValue"], limit, "invalid_attributes") def _usage(attributes, key, required=False, maximum=MAX_TOKENS): value = attributes.get(key) if value is None: if required: fail("missing_usage", "A selected inference span lacks reported input or output usage.") return None if set(value) != {"intValue"}: fail("invalid_usage", "Reported usage must use an OTLP intValue.") return _integer(value["intValue"], maximum, "invalid_usage", wire=True) def _span_id(value, width, code="invalid_span"): if not isinstance(value, str) or not re.fullmatch(r"[0-9a-fA-F]{%d}" % width, value) or int(value, 16) == 0: fail(code, "OTLP trace and span IDs must be nonzero hexadecimal values of the required length.") return value.lower() def _is_inference(attributes, format): if format == "openinference": return _string_attribute(attributes, "openinference.span.kind") == "LLM" return _string_attribute(attributes, "gen_ai.operation.name") in INFERENCE_OPERATIONS def _choose(records, field, selected, ambiguous_code): values = {r[field] for r in records} if selected is None: if len(values) != 1: fail(ambiguous_code, "Multiple accounting scopes are present; supply an explicit selector.") selected = next(iter(values)) result = [r for r in records if r[field] == selected] if not result: fail("selection_empty", "The supplied accounting selector matches no inference spans.") return result, selected def _validate_parent_graph(records): parents = {(r["trace_id"], r["span_id"]): (r["trace_id"], r["parent_id"]) for r in records} done = set() for node in parents: visiting = set() while node in parents and node not in done: if node in visiting: fail("invalid_parent_graph", "The exported parent-span graph contains a cycle.") visiting.add(node) node = parents[node] done.update(visiting) return parents def normalize_bytes(data, format, trace_id=None, service=None, scope=None, *, include_call_metadata=False): """Decode an OTLP JSON/JSONL export without any network or filesystem access.""" if len(data) > MAX_INPUT: fail("input_too_large", "Telemetry input exceeds the 16 MiB limit.") if format not in {"genai", "openinference"}: fail("unsupported_format", "Select genai or openinference explicitly.") if trace_id is not None: trace_id = _span_id(trace_id, 32) for selector in (service, scope): if selector is not None: _label(selector, 256, "invalid_selector") try: text = data.decode("utf-8") except UnicodeError: fail("invalid_json", "Telemetry must be UTF-8 JSON or JSONL.") try: documents = [_loads(text)] except ClientError: lines = [line for line in text.splitlines() if line.strip()] if len(lines) < 2: raise documents = [_loads(line) for line in lines] records = [] seen = {} source_spans = duplicates = 0 for document in documents: _object(document, "invalid_otlp") if "resourceSpans" not in document: fail("invalid_otlp", "Expected an OTLP resourceSpans envelope.") for resource in _array(document["resourceSpans"], "invalid_otlp"): _object(resource, "invalid_otlp") resource_data = _object(resource.get("resource", {}), "invalid_otlp") resource_attributes = _attributes(resource_data.get("attributes", [])) service_name = _string_attribute(resource_attributes, "service.name") for scope_spans in _array(resource.get("scopeSpans", []), "invalid_otlp"): _object(scope_spans, "invalid_otlp") scope_data = _object(scope_spans.get("scope", {}), "invalid_otlp") scope_name = scope_data.get("name") if scope_name is not None: _label(scope_name, 256, "invalid_otlp") for raw_span in _array(scope_spans.get("spans", []), "invalid_otlp"): source_spans += 1 if source_spans > MAX_SPANS: fail("too_many_spans", "Telemetry input exceeds 20,000 spans.") _object(raw_span, "invalid_span") sid = _span_id(raw_span.get("spanId"), 16) tid = _span_id(raw_span.get("traceId"), 32) parent = raw_span.get("parentSpanId") parent = _span_id(parent, 16) if parent not in (None, "") else None # Include resource and scope in identity checks; never export this digest. digest = hashlib.sha256(_json_bytes([resource_data, scope_data, raw_span])).digest() key = (tid, sid) if key in seen: if seen[key] != digest: fail("conflicting_span", "Repeated trace/span IDs have conflicting exported records.") duplicates += 1 continue seen[key] = digest attributes = _attributes(raw_span.get("attributes", [])) records.append({"trace_id": tid, "span_id": sid, "parent_id": parent, "service": service_name, "scope": scope_name, "raw": raw_span, "attributes": attributes, "inference": _is_inference(attributes, format)}) inference = [r for r in records if r["inference"]] if not inference: fail("selection_empty", "No supported inference spans were found in the export.") selected, trace_id = _choose(inference, "trace_id", trace_id, "ambiguous_trace") selected, service = _choose(selected, "service", service, "ambiguous_service") if scope is not None: selected = [r for r in selected if r["scope"] == scope] if not selected: fail("selection_empty", "The instrumentation scope matches no inference spans.") if len(selected) > MAX_CALLS: fail("too_many_calls", "The selected accounting scope exceeds 1,000 inference calls.") parents = _validate_parent_graph(records) selected_ids = {(r["trace_id"], r["span_id"]) for r in selected} for key in selected_ids: ancestor = parents.get(key) while ancestor in parents: if ancestor in selected_ids: fail("nested_inference", "Nested inference spans are ambiguous; select one instrumentation scope or fix the export.") ancestor = parents[ancestor] turns = [] calls = [] writes = [] models = set() missing_model = False dropped = False for record in selected: raw = record["raw"] # SERVER spans can mirror client usage. Only client/internal inference is supported. if type(raw.get("kind")) is not int or raw["kind"] not in (1, 3): fail("unsupported_span_kind", "Selected inference spans must have CLIENT or INTERNAL numeric kind.") start = _integer(raw.get("startTimeUnixNano"), 2**64 - 1, "invalid_timestamp", wire=True) end = _integer(raw.get("endTimeUnixNano"), 2**64 - 1, "invalid_timestamp", wire=True) if end < start: fail("invalid_timestamp", "An inference span ends before it starts.") attrs = record["attributes"] if format == "genai": input_key, output_key = "gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens" read_key, write_key = "gen_ai.usage.cache_read.input_tokens", "gen_ai.usage.cache_write.input_tokens" model = _string_attribute(attrs, "gen_ai.response.model") or _string_attribute(attrs, "gen_ai.request.model") else: input_key, output_key = "llm.token_count.prompt", "llm.token_count.completion" read_key, write_key = "llm.token_count.prompt_details.cache_read", "llm.token_count.prompt_details.cache_write" model = (_string_attribute(attrs, "llm.response.model_name") or _string_attribute(attrs, "llm.model_name") or _string_attribute(attrs, "llm.request.model_name")) inputs, outputs = _usage(attrs, input_key, True), _usage(attrs, output_key, True) if format == "openinference": total = _usage(attrs, "llm.token_count.total", maximum=2 * MAX_TOKENS) if total is not None and total != inputs + outputs: fail("contradictory_usage", "The reported token total disagrees with input plus output usage.") reads, written = _usage(attrs, read_key), _usage(attrs, write_key) if format == "genai": legacy = _usage(attrs, "gen_ai.usage.cache_creation.input_tokens") if written is not None and legacy is not None and written != legacy: fail("conflicting_cache_alias", "Current and legacy cache-write attributes disagree.") if written is None: written = legacy if (reads or 0) + (written or 0) > inputs: fail("contradictory_usage", "Reported cache subsets exceed total input usage.") turns.append((start, record["span_id"], { "id": record["span_id"], "input_tokens": inputs, "output_tokens": outputs, "cached_input_tokens": reads, "context_blocks": [], "tool_calls": None, })) if include_call_metadata: provider_key = "gen_ai.provider.name" if format == "genai" else "llm.provider" provider = _string_attribute(attrs, provider_key) status = _object(raw.get("status", {}), "invalid_span").get("code", 0) if type(status) is not int or status not in (0, 1, 2): fail("invalid_span", "OTLP status must have an unset, OK, or error numeric code.") calls.append({"id": record["span_id"], "parent_id": record["parent_id"], "provider": provider, "model": model, "start_time_unix_nano": str(start), "end_time_unix_nano": str(end), "status": {0: "unknown", 1: "completed", 2: "failed"}[status], "input_tokens": inputs, "output_tokens": outputs, "cache_read_tokens": reads, "cache_write_tokens": written}) if written is not None: writes.append({"turn_id": record["span_id"], "tokens": written}) if model is not None: models.add(model) else: missing_model = True dropped = dropped or bool(raw.get("droppedAttributesCount", 0) or raw.get("droppedEventsCount", 0)) turns.sort(key=lambda item: (item[0], item[1])) trace = {"version": 1, "name": "OTLP " + trace_id, "turns": [t[2] for t in turns]} if len(models) == 1 and not missing_model: trace["model"] = next(iter(models)) warnings = [] if dropped: warnings.append("dropped_telemetry") if service is None: warnings.append("service_name_missing") if len(models) != 1 or missing_model: warnings.append("model_identity_incomplete_or_mixed") if any(t[2]["cached_input_tokens"] is None for t in turns): warnings.append("cache_read_coverage_incomplete") return {"format": "turnscope.import.v1", "trace": trace, "import": { "source_format": "otlp." + format, "accounting_scope": {"trace_id": trace_id, "service": service, "instrumentation_scope": scope, "completeness": "selected_export_only"}, "source_spans": source_spans, "duplicate_spans": duplicates, "selected_inference_spans": len(selected), "excluded_inference_spans": len(inference) - len(selected), "dropped_non_inference_spans": len(records) - len(inference), "cache_write_observations": sorted(writes, key=lambda item: item["turn_id"]), "models": sorted(models), "warnings": warnings, "limitations": list(IMPORT_LIMITATIONS), **({"calls": sorted(calls, key=lambda call: (int(call["start_time_unix_nano"]), call["id"]))} if include_call_metadata else {}), }} def sanitize_trace(trace): """Validate the v1 upload surface and replace context content with its hash.""" code = "invalid_trace" _keys(trace, {"version", "name", "model", "pricing", "turns"}, {"version", "turns"}, code) if type(trace["version"]) is not int or trace["version"] != 1: fail(code, "Only version-1 normalized traces can be analyzed.") result = copy.deepcopy(trace) for key in ("name", "model"): if result.get(key) is not None: _label(result[key], 256, code) if result.get("pricing") is not None: pricing = result["pricing"] keys = {"currency", "input_per_million", "cached_input_per_million", "output_per_million"} _keys(pricing, keys, keys, code) if not isinstance(pricing["currency"], str) or not re.fullmatch(r"[A-Z]{3}", pricing["currency"]): fail(code, "Pricing currency must be three uppercase letters.") for key in keys - {"currency"}: _number(pricing[key], 1_000_000, code) turns = _array(result["turns"], code) if not 1 <= len(turns) <= MAX_CALLS: fail(code, "A trace must contain between one and 1,000 turns.") seen_turns = set() block_count = call_count = 0 for turn in turns: _keys(turn, {"id", "input_tokens", "output_tokens", "cached_input_tokens", "context_blocks", "tool_calls"}, {"id", "input_tokens", "output_tokens"}, code) tid = _label(turn["id"], 256, code) if tid in seen_turns: fail(code, "Turn IDs must be unique.") seen_turns.add(tid) inputs = _integer(turn["input_tokens"], MAX_TOKENS, code) _integer(turn["output_tokens"], MAX_TOKENS, code) if turn.get("cached_input_tokens") is not None: _integer(turn["cached_input_tokens"], inputs, code) blocks = _array(turn.get("context_blocks", []), code) calls = _array([] if turn.get("tool_calls") is None else turn["tool_calls"], code) if len(blocks) > 256 or len(calls) > 256: fail(code, "Per-turn context or tool-call limits were exceeded.") inventory = 0 block_ids = set() for block in blocks: _keys(block, {"id", "kind", "tokens", "fingerprint", "content", "tool_name"}, {"id", "kind", "tokens"}, code) bid = _label(block["id"], 256, code) if bid in block_ids: fail(code, "Block IDs must be unique within each turn.") block_ids.add(bid) if block["kind"] not in {"static_document", "tool_result", "tool_schema", "history", "other"}: fail(code, "Unsupported context-block kind.") inventory += _integer(block["tokens"], MAX_TOKENS, code) if block.get("tool_name") is not None: _label(block["tool_name"], 128, code) content, fingerprint = block.get("content"), block.get("fingerprint") if fingerprint is not None and (not isinstance(fingerprint, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", fingerprint)): fail(code, "Context fingerprints must be lowercase SHA-256 values.") if content is not None: if not isinstance(content, str) or len(content.encode("utf-8")) > 32768: fail(code, "Context content exceeds the supported UTF-8 byte limit.") computed = "sha256:" + hashlib.sha256(content.encode("utf-8")).hexdigest() if fingerprint is not None and fingerprint != computed: fail("fingerprint_mismatch", "Context content does not match its supplied fingerprint.") block["fingerprint"] = computed elif fingerprint is None: fail(code, "Every context block requires a fingerprint or content.") block.pop("content", None) if inventory > inputs: fail(code, "Context inventory exceeds total reported input.") for call in calls: _keys(call, {"name", "kind", "target"}, {"name", "kind"}, code) _label(call["name"], 128, code) if call["kind"] not in {"discovery", "read", "search", "other"}: fail(code, "Unsupported selected tool-call kind.") if call.get("target") is not None: _label(call["target"], 2048, code) block_count += len(blocks) call_count += len(calls) if block_count > 20_000 or call_count > 20_000: fail(code, "Trace-wide context or tool-call limits were exceeded.") return result def _validate_metrics(metrics, code="invalid_measurement"): _keys(metrics, METRICS, code=code) for key, value in metrics.items(): if value is None: continue if key in {"input_tokens", "output_tokens", "turns"}: _integer(value, MAX_METRIC, code) elif key == "duration_ms": _number(value, MAX_METRIC, code) else: keys = {"lower", "upper", "currency", "basis", "assumptions_id"} _keys(value, keys, keys, code) lower = _number(value["lower"], MAX_METRIC, code) upper = _number(value["upper"], MAX_METRIC, code) if lower > upper: fail(code, "The cost lower bound exceeds its upper bound.") if not isinstance(value["currency"], str) or not re.fullmatch(r"[A-Z]{3}", value["currency"]): fail(code, "Cost currency must be three uppercase letters.") if value["basis"] not in ("modeled", "reported"): fail(code, "Cost basis must be modeled or reported.") _label(value["assumptions_id"], 128, code) return copy.deepcopy(metrics) def make_measurement(report, *, task_id, trial_id, evaluator_id, outcome, scope_id): """Attach explicit external evaluation claims to reported accounting metrics.""" if not isinstance(report, dict) or "measurement_metrics" not in report: fail("invalid_report", "The report lacks measurement_metrics; use a compatible analyze endpoint.") metrics = _validate_metrics(report["measurement_metrics"], "invalid_report") record = {"task_id": task_id, "trial_id": trial_id, "evaluator_id": evaluator_id, "outcome": outcome, "scope_id": scope_id, "metrics": metrics} return _validate_measurements(record)[0] def _validate_measurements(document): records = document if isinstance(document, list) else [document] if not 1 <= len(records) <= 1000: fail("invalid_measurement", "Each comparison side requires one to 1,000 measurements.") keys = {"task_id", "trial_id", "evaluator_id", "outcome", "scope_id", "metrics"} seen = set() for record in records: _keys(record, keys, keys, "invalid_measurement") for key in ("task_id", "trial_id", "evaluator_id", "scope_id"): _label(record[key]) if record["outcome"] not in ("passed", "failed", "unknown"): fail("invalid_measurement", "Outcome must be passed, failed, or unknown.") pair = (record["task_id"], record["trial_id"]) if pair in seen: fail("invalid_measurement", "Task/trial pairs must be unique within each comparison side.") seen.add(pair) _validate_metrics(record["metrics"]) return copy.deepcopy(records) def _server_url(server, endpoint): if (not isinstance(server, str) or any(ch.isspace() or unicodedata.category(ch) == "Cc" for ch in server) or "\\" in server): fail("invalid_server", "Supply an HTTPS URL or a loopback HTTP URL without credentials, query, or fragment.") try: parsed = urllib.parse.urlsplit(server) hostname = parsed.hostname port = parsed.port if (parsed.scheme not in {"https", "http"} or not hostname or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or "?" in server or "#" in server or "%" in hostname or (port is not None and port == 0)): raise ValueError() if parsed.scheme == "http": if hostname.lower() != "localhost" and not ipaddress.ip_address(hostname).is_loopback: raise ValueError() return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, parsed.path.rstrip("/") + endpoint, "", "")) except (ValueError, UnicodeError): fail("invalid_server", "Supply an HTTPS URL or a loopback HTTP URL without credentials, query, or fragment.") class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def _retry_after(value): if value is None: return 1.0 if re.fullmatch(r"[0-9]+", value.strip()): return float(value) if len(value.strip()) < 12 else math.inf try: when = email.utils.parsedate_to_datetime(value) if when.tzinfo is None: when = when.replace(tzinfo=datetime.timezone.utc) return max(0, when.timestamp() - time.time()) except (ValueError, TypeError, OverflowError): return 1.0 def post_json(server, endpoint, payload): """POST JSON with no redirects, ambient proxy, or unbounded retry/body.""" url = _server_url(server, endpoint) body = _json_bytes(payload) if len(body) > MAX_REQUEST: fail("request_too_large", "The sanitized API request exceeds the 2 MiB limit.") opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect()) spent_wait = 0.0 for attempt in range(MAX_ATTEMPTS): request = urllib.request.Request(url, data=body, method="POST", headers={ "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "TurnScope-local-client/0.2.0", }) deadline = time.monotonic() + TIMEOUT try: with opener.open(request, timeout=TIMEOUT) as response: content_length = response.headers.get("Content-Length") declared_length = None if content_length is not None: if not re.fullmatch(r"[0-9]+", content_length.strip()): fail("invalid_response", "The API returned an invalid response length.") if len(content_length.strip()) > 20 or int(content_length) > MAX_RESPONSE: fail("response_too_large", "The API response exceeds the 16 MiB limit.") declared_length = int(content_length) data = bytearray() while True: if time.monotonic() >= deadline: fail("network_error", "The API response exceeded the overall read deadline.") chunk = response.read1(min(65536, MAX_RESPONSE + 1 - len(data))) if time.monotonic() >= deadline: fail("network_error", "The API response exceeded the overall read deadline.") if not chunk: break data.extend(chunk) if len(data) > MAX_RESPONSE: fail("response_too_large", "The API response exceeds the 16 MiB limit.") if not response.chunked and declared_length is not None and len(data) != declared_length: fail("invalid_response", "The API response ended before its declared length.") result = _loads(data, "invalid_response") return _object(result, "invalid_response") except urllib.error.HTTPError as error: status = error.code delay = _retry_after(error.headers.get("Retry-After")) if status == 429 else None error.close() if status != 429: fail("http_error", "The server rejected the request or attempted a redirect (HTTP %d)." % status) if attempt + 1 == MAX_ATTEMPTS or delay > MAX_RETRY_DELAY - spent_wait: fail("rate_limited", "The server is busy; retry later. The bounded retry budget was exhausted.") time.sleep(delay) spent_wait += delay except (urllib.error.URLError, OSError, ValueError, http.client.HTTPException): fail("network_error", "The server request failed; no response body or connection details are included.") fail("rate_limited", "The bounded retry budget was exhausted.") def _read(path): try: if path == "-": data = sys.stdin.buffer.read(MAX_INPUT + 1) else: with open(path, "rb") as stream: data = stream.read(MAX_INPUT + 1) except OSError: fail("input_error", "The input file could not be read.") if len(data) > MAX_INPUT: fail("input_too_large", "Input exceeds the 16 MiB limit.") return data def _write(path, result): data = _json_bytes(result) + b"\n" temporary = None try: if path == "-": sys.stdout.write(data.decode("utf-8")) return directory = os.path.dirname(os.path.abspath(path)) with tempfile.NamedTemporaryFile(mode="wb", dir=directory, prefix=".turnscope-", delete=False) as stream: temporary = stream.name stream.write(data) os.replace(temporary, path) temporary = None except OSError: fail("output_error", "The output file could not be written.") finally: if temporary is not None: try: os.unlink(temporary) except OSError: pass def _import_metadata(metadata): """Keep only the bounded first-party accounting metadata contract.""" allowed = {"source_format", "accounting_scope", "source_spans", "duplicate_spans", "selected_inference_spans", "excluded_inference_spans", "dropped_non_inference_spans", "cache_write_observations", "models", "warnings", "limitations"} _keys(metadata, allowed | {"calls"}, allowed, "invalid_import") if metadata["source_format"] not in ("otlp.genai", "otlp.openinference"): fail("invalid_import", "Unsupported import metadata format.") accounting = metadata["accounting_scope"] keys = {"trace_id", "service", "instrumentation_scope", "completeness"} _keys(accounting, keys, keys, "invalid_import") _span_id(accounting["trace_id"], 32, "invalid_import") for key in ("service", "instrumentation_scope"): if accounting[key] is not None: _label(accounting[key], 256, "invalid_import") if accounting["completeness"] != "selected_export_only": fail("invalid_import", "Import metadata must retain its limited accounting scope.") for key in ("source_spans", "duplicate_spans", "selected_inference_spans", "excluded_inference_spans", "dropped_non_inference_spans"): _integer(metadata[key], MAX_SPANS, "invalid_import") writes = _array(metadata["cache_write_observations"], "invalid_import") models = _array(metadata["models"], "invalid_import") warnings = _array(metadata["warnings"], "invalid_import") if len(writes) > MAX_CALLS or len(models) > MAX_CALLS or len(warnings) > 4: fail("invalid_import", "Import metadata exceeds its supported bounds.") for entry in writes: _keys(entry, {"turn_id", "tokens"}, {"turn_id", "tokens"}, "invalid_import") _span_id(entry["turn_id"], 16, "invalid_import") _integer(entry["tokens"], MAX_TOKENS, "invalid_import") for model in models: _label(model, 256, "invalid_import") for warning in warnings: if warning not in ("dropped_telemetry", "service_name_missing", "model_identity_incomplete_or_mixed", "cache_read_coverage_incomplete"): fail("invalid_import", "Unsupported import warning.") if metadata["limitations"] != IMPORT_LIMITATIONS: fail("invalid_import", "The import limitations are missing or have been changed.") if "calls" in metadata: calls = _array(metadata["calls"], "invalid_import") if len(calls) != metadata["selected_inference_spans"] or len(calls) > MAX_CALLS: fail("invalid_import", "Per-call metadata must match the selected inference count.") seen = set() for call in calls: keys = {"id", "parent_id", "provider", "model", "start_time_unix_nano", "end_time_unix_nano", "status", "input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens"} _keys(call, keys, keys, "invalid_import") sid = _span_id(call["id"], 16, "invalid_import") if sid in seen: fail("invalid_import", "Per-call metadata IDs must be unique.") seen.add(sid) if call["parent_id"] is not None: _span_id(call["parent_id"], 16, "invalid_import") for key in ("provider", "model"): if call[key] is not None: _label(call[key], 256, "invalid_import") start = _integer(call["start_time_unix_nano"], 2**64 - 1, "invalid_import", wire=True) end = _integer(call["end_time_unix_nano"], 2**64 - 1, "invalid_import", wire=True) if end < start or call["status"] not in ("unknown", "completed", "failed"): fail("invalid_import", "Per-call timing or execution status is invalid.") inputs = _integer(call["input_tokens"], MAX_TOKENS, "invalid_import") _integer(call["output_tokens"], MAX_TOKENS, "invalid_import") for key in ("cache_read_tokens", "cache_write_tokens"): if call[key] is not None: _integer(call[key], inputs, "invalid_import") if (call["cache_read_tokens"] or 0) + (call["cache_write_tokens"] or 0) > inputs: fail("invalid_import", "Per-call cache subsets exceed inclusive input usage.") return copy.deepcopy(metadata) class _Parser(argparse.ArgumentParser): def error(self, _message): fail("usage_error", "Invalid command arguments; run the command with --help for its required options.") def _parser(): parser = _Parser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) normalize = commands.add_parser("normalize", help="Normalize local OTLP JSON/JSONL; no network access") normalize.add_argument("input") normalize.add_argument("--format", choices=("genai", "openinference"), required=True) for flag in ("trace-id", "service", "scope"): normalize.add_argument("--" + flag) normalize.add_argument("--output", required=True) analyze = commands.add_parser("analyze", help="Send a sanitized v1 trace to an explicit server") analyze.add_argument("input") analyze.add_argument("--server", required=True) analyze.add_argument("--output", required=True) measure = commands.add_parser("measure", help="Attach caller-supplied evaluation claims to an analyzed report") measure.add_argument("input") for flag in ("task-id", "trial-id", "evaluator-id", "scope-id"): measure.add_argument("--" + flag, required=True) measure.add_argument("--outcome", choices=("passed", "failed", "unknown"), required=True) measure.add_argument("--output", required=True) compare = commands.add_parser("compare", help="Compare supplied measurements through an explicit server") compare.add_argument("baseline") compare.add_argument("candidate") compare.add_argument("--server", required=True) compare.add_argument("--objective", choices=sorted(METRICS), required=True) compare.add_argument("--max-regression-pct", type=float, default=0) compare.add_argument("--check", action="store_true") compare.add_argument("--output", required=True) return parser def main(argv=None): try: args = _parser().parse_args(argv) exit_code = 0 if args.command == "normalize": result = normalize_bytes(_read(args.input), args.format, args.trace_id, args.service, args.scope) elif args.command == "analyze": document = _loads(_read(args.input)) _object(document) metadata = None if document.get("format") == "turnscope.import.v1": _keys(document, {"format", "trace", "import"}, {"format", "trace", "import"}, "invalid_import") metadata = _import_metadata(document["import"]) document = document["trace"] trace = sanitize_trace(document) result = post_json(args.server, "/api/v1/analyze", trace) if "measurement_metrics" not in result: fail("invalid_report", "The server response lacks measurement_metrics.") _validate_metrics(result["measurement_metrics"], "invalid_report") if metadata is not None: result["import"] = metadata elif args.command == "measure": result = make_measurement(_loads(_read(args.input)), task_id=args.task_id, trial_id=args.trial_id, evaluator_id=args.evaluator_id, outcome=args.outcome, scope_id=args.scope_id) else: _number(args.max_regression_pct, 1000, "invalid_measurement") payload = {"version": 1, "objective": args.objective, "max_regression_pct": args.max_regression_pct, "baseline": _validate_measurements(_loads(_read(args.baseline))), "candidate": _validate_measurements(_loads(_read(args.candidate)))} result = post_json(args.server, "/api/v1/compare", payload) verdict = result.get("verdict") if verdict not in ("within_budget", "regression", "insufficient_evidence"): fail("invalid_response", "The comparison response lacks a supported verdict.") if args.check: exit_code = {"within_budget": 0, "regression": 1, "insufficient_evidence": 2}[verdict] _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): sys.stderr.write('{"error":{"code":"invalid_document","message":"Unsupported or excessively nested document."}}\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())