#!/usr/bin/env python3 """Send sanitized OTLP/HTTP JSON scenarios to a local Perseval project.""" from __future__ import annotations import argparse import hashlib import json import time import urllib.error import urllib.request def string_attr(key: str, value: str) -> dict: return {"key": key, "value": {"stringValue": value}} def bool_attr(key: str, value: bool) -> dict: return {"key": key, "value": {"boolValue": value}} def stable_hex(label: str, length: int) -> str: return hashlib.sha256(label.encode()).hexdigest()[:length] def span( trace_id: str, scenario: str, name: str, role: str, operation: str, start_ns: int, duration_ms: int, *, parent: str | None = None, error: bool = False, attributes: list[dict] | None = None, links: list[dict] | None = None, events: list[dict] | None = None, ) -> dict: span_id = stable_hex(f"{scenario}:{name}", 16) result = { "traceId": trace_id, "spanId": span_id, "name": name, "kind": "SPAN_KIND_INTERNAL", "startTimeUnixNano": str(start_ns), "endTimeUnixNano": str(start_ns + duration_ms * 1_000_000), "attributes": [ string_attr("agent.role", role), string_attr("gen_ai.agent.id", f"fixture-{role}"), string_attr("agent.operation", operation), *(attributes or []), ], "status": { "code": "STATUS_CODE_ERROR" if error else "STATUS_CODE_OK", "message": "sanitized fixture failure" if error else "", }, } if parent: result["parentSpanId"] = parent if links: result["links"] = links if events: result["events"] = events return result def resource_spans(project: str, scenario: str, build: str, spans: list[dict]) -> dict: return { "resource": { "attributes": [ string_attr("perseval.project.id", project), string_attr("service.name", "perseval-docs-fixture"), string_attr("service.version", build), string_attr("deployment.environment.name", "docs-local"), string_attr("gen_ai.conversation.id", f"docs-{scenario}"), ] }, "scopeSpans": [ { "scope": {"name": "perseval.docs.fixture", "version": "1"}, "spans": spans, } ], } def root_and_children( project: str, scenario: str, build: str, children: list[dict], *, root_error: bool, ) -> dict: now = time.time_ns() trace_id = stable_hex(scenario, 32) root_id = stable_hex(f"{scenario}:root", 16) root = span( trace_id, scenario, "orchestrator.run", "orchestrator", "run_agent", now, 10_000, error=root_error, attributes=[ string_attr("agent.final.status", "failed" if root_error else "completed"), string_attr("gen_ai.conversation.id", f"docs-{scenario}"), ], ) root["spanId"] = root_id materialized = [] for index, child in enumerate(children, start=1): materialized.append( span( trace_id, scenario, child["name"], child["role"], child["operation"], now + index * 1_000_000_000, child.get("duration_ms", 800), parent=root_id, error=child.get("error", False), attributes=child.get("attributes"), links=child.get("links"), events=child.get("events"), ) ) return resource_spans(project, scenario, build, [root, *materialized]) def scenarios(project: str, selected: str, occurrence: int) -> list[dict]: items: dict[str, list[dict]] = {} def occurrence_id(scenario: str) -> str: return f"{scenario}:occurrence:{occurrence}" items["uncertain-mutation"] = [ root_and_children( project, occurrence_id("uncertain-mutation"), "0.9.0-uncertain", [ { "name": "planner.plan_return", "role": "planner", "operation": "plan_return", }, { "name": "publisher.submit_return", "role": "publisher", "operation": "submit_return", "error": True, "attributes": [ string_attr("gen_ai.tool.name", "returns_api"), string_attr("agent.tool.requirement", "required"), string_attr("agent.tool.status", "timed_out"), bool_attr("tool.result.success", False), string_attr("agent.operation.effect", "mutating"), string_attr("agent.operation.retry_safety", "unknown"), string_attr("agent.state.predicate", "return_created"), string_attr("agent.state.observation", "ambiguous"), ], "events": [ { "timeUnixNano": str(time.time_ns()), "name": "exception", "attributes": [string_attr("exception.type", "TimeoutError")], } ], }, ], root_error=True, ) ] items["recovered-optional"] = [ root_and_children( project, occurrence_id("recovered-optional"), "1.0.0-optional-fallback", [ { "name": "browser.lookup_optional_docs", "role": "browser", "operation": "lookup_optional_docs", "error": True, "attributes": [ string_attr("gen_ai.tool.name", "browser"), string_attr("agent.tool.requirement", "optional"), string_attr("agent.tool.status", "failed"), bool_attr("tool.result.success", False), string_attr("agent.operation.effect", "read_only"), string_attr("agent.operation.retry_safety", "idempotent"), ], }, { "name": "retriever.use_local_fallback", "role": "retriever", "operation": "use_local_fallback", "attributes": [ string_attr("agent.tool.requirement", "required"), string_attr("agent.tool.status", "succeeded"), bool_attr("tool.result.success", True), ], }, ], root_error=False, ) ] handoff_scenario = occurrence_id("spanlink-handoff") handoff_trace = stable_hex(handoff_scenario, 32) browser_id = stable_hex(f"{handoff_scenario}:browser.collect_evidence", 16) items[handoff_scenario] = [ root_and_children( project, handoff_scenario, "1.0.0-handoff", [ { "name": "planner.plan_research", "role": "planner", "operation": "plan_research", }, { "name": "browser.collect_evidence", "role": "browser", "operation": "collect_evidence", "duration_ms": 1800, }, { "name": "verifier.verify_evidence", "role": "verifier", "operation": "verify_evidence", "attributes": [string_attr("agent.handoff", "evidence")], "links": [ { "traceId": handoff_trace, "spanId": browser_id, "attributes": [string_attr("agent.handoff", "evidence")], } ], }, ], root_error=False, ) ] baseline = root_and_children( project, occurrence_id("baseline-regression"), "0.9.0-baseline", [ {"name": "planner.plan_release", "role": "planner", "operation": "plan_release"}, { "name": "test_runner.run_release_tests", "role": "test_runner", "operation": "run_release_tests", "error": True, "attributes": [ string_attr("gen_ai.tool.name", "release_test_runner"), string_attr("agent.tool.requirement", "required"), string_attr("agent.tool.status", "failed"), bool_attr("tool.result.success", False), string_attr("agent.operation.effect", "read_only"), string_attr("agent.operation.retry_safety", "idempotent"), ], }, { "name": "release_verifier.verify_release", "role": "release_verifier", "operation": "verify_release", "error": True, "attributes": [ string_attr("gen_ai.tool.name", "release_verifier"), string_attr("agent.tool.requirement", "required"), string_attr("agent.tool.status", "failed"), bool_attr("tool.result.success", False), string_attr("agent.operation.effect", "verifying"), string_attr("agent.operation.retry_safety", "idempotent"), ], }, ], root_error=True, ) repaired = root_and_children( project, occurrence_id("repaired-regression"), "1.0.0-repaired", [ {"name": "planner.plan_release", "role": "planner", "operation": "plan_release"}, { "name": "test_runner.run_release_tests", "role": "test_runner", "operation": "run_release_tests", "attributes": [ string_attr("gen_ai.tool.name", "release_test_runner"), string_attr("agent.tool.requirement", "required"), string_attr("agent.tool.status", "succeeded"), bool_attr("tool.result.success", True), ], }, { "name": "release_verifier.verify_release", "role": "release_verifier", "operation": "verify_release", "attributes": [ string_attr("gen_ai.tool.name", "release_verifier"), string_attr("agent.tool.requirement", "required"), string_attr("agent.tool.status", "succeeded"), bool_attr("tool.result.success", True), ], }, ], root_error=False, ) items["baseline-repaired"] = [baseline, repaired] items["missing-telemetry"] = [ root_and_children( project, occurrence_id("missing-telemetry"), "0.9.0-missing-facts", [ { "name": "test_runner.run_uninstrumented_tests", "role": "test_runner", "operation": "run_uninstrumented_tests", "error": True, "attributes": [string_attr("gen_ai.tool.name", "test_runner")], } ], root_error=True, ) ] if selected == "all": return [item for name in items for item in items[name]] return items[selected] def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--endpoint", required=True, help="Full endpoint copied from Sources") parser.add_argument("--project", required=True, help="Project ID copied from Sources") parser.add_argument( "--scenario", default="all", choices=[ "all", "uncertain-mutation", "recovered-optional", "spanlink-handoff", "baseline-repaired", "missing-telemetry", ], ) parser.add_argument("--write", help="Write JSON to this path instead of sending it") parser.add_argument( "--occurrences", type=int, default=1, help="Create this many distinct trace occurrences per selected scenario (default: 1)", ) args = parser.parse_args() if args.occurrences < 1: parser.error("--occurrences must be at least 1") payload = { "resourceSpans": [ resource_span for occurrence in range(1, args.occurrences + 1) for resource_span in scenarios(args.project, args.scenario, occurrence) ] } encoded = json.dumps(payload, separators=(",", ":")).encode() if args.write: with open(args.write, "wb") as destination: destination.write(encoded) print(f"wrote {len(encoded)} bytes to {args.write}") return 0 request = urllib.request.Request( args.endpoint, data=encoded, method="POST", headers={"Content-Type": "application/json"}, ) try: with urllib.request.urlopen(request, timeout=30) as response: body = response.read().decode() or "{}" print(f"HTTP {response.status}: {body}") except urllib.error.HTTPError as error: print(f"HTTP {error.code}: {error.read().decode()}") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())