#!/usr/bin/env python3 """Original LockedIn Labs first-trace exercise. Python 3.11+; synthetic data only. Install fde-first-trace-requirements.txt in a virtual environment, then: python fde-first-trace.py test python fde-first-trace.py compare python fde-first-trace.py run --case success --omit-lookup-span Real local OpenTelemetry spans; no model, collector or network connection. """ import argparse from contextlib import ExitStack, contextmanager, nullcontext from datetime import datetime, timezone from importlib.metadata import version import json import time import unittest from unittest.mock import patch SCHEMA = "lockedin.first-trace/v1" CASES = ("success", "missing", "failure") LABELS = { "success": "Guide found", "missing": "Guide missing", "failure": "Lookup failed", } GUIDES = {"P1": "Synthetic instructions: open Settings, then choose Notifications."} @contextmanager def without_network(): """Exercise commands fail if code attempts a socket connection or DNS lookup.""" with ExitStack() as stack: for target in ("socket.create_connection", "socket.getaddrinfo", "socket.socket.connect"): stack.enter_context(patch(target, side_effect=RuntimeError("network_disabled_in_lab"))) yield class LookupUnavailable(Exception): pass def lookup_guide(case): # Deliberate teaching delay, not a measured database/service latency. time.sleep(0.030) if case == "failure": raise LookupUnavailable("SYNTHETIC_PRIVATE_DETAIL_DO_NOT_EXPORT") return GUIDES.get("P404" if case == "missing" else "P1") def format_reply(guide): time.sleep(0.015) # Deliberate local delay, not a model call. return {"route": "guide", "text": guide} def choose_fallback(): time.sleep(0.005) return {"route": "handoff", "text": "A person should review this request."} def trace_case(case, omit_lookup_span=False): if case not in CASES or type(omit_lookup_span) is not bool: raise ValueError("invalid_lab_case") from opentelemetry.context import Context from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import SpanLimits, TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from opentelemetry.sdk.trace.sampling import ALWAYS_ON from opentelemetry.trace import Status, StatusCode exporter = InMemorySpanExporter() provider = TracerProvider( resource=Resource({"service.name": "lockedin-first-trace", "training.origin": "synthetic"}), sampler=ALWAYS_ON, shutdown_on_exit=False, span_limits=SpanLimits(max_attributes=4, max_events=2, max_links=0, max_event_attributes=2, max_attribute_length=60), ) provider.add_span_processor(SimpleSpanProcessor(exporter)) tracer = provider.get_tracer("lockedin.first-trace", "1") def operation(name, **kwargs): # Do not let exception messages or stack traces become automatic events. return tracer.start_as_current_span(name, record_exception=False, set_status_on_exception=False, **kwargs) try: with operation("support.request", context=Context()) as root: root.set_attribute("training.case", case) root.add_event("request.started") lookup_context = nullcontext(None) if omit_lookup_span else operation("guide.lookup") with lookup_context as lookup: try: guide = lookup_guide(case) except LookupUnavailable: guide = None if lookup is not None: lookup.set_status(Status(StatusCode.ERROR)) lookup.add_event("lookup.failed", {"error.type": "lookup_unavailable"}) else: if lookup is not None: lookup.set_attribute("training.found", guide is not None) if guide is None: with operation("handoff.choose"): reply = choose_fallback() else: with operation("reply.format"): reply = format_reply(guide) root.set_attribute("training.route", reply["route"]) root.add_event("route.selected", {"training.route": reply["route"]}) provider.force_flush() finished = list(exporter.get_finished_spans()) finally: provider.shutdown() roots = [span for span in finished if span.parent is None] if len(roots) != 1 or len(finished) != (2 if omit_lookup_span else 3): raise RuntimeError("trace_incomplete_check_sdk_configuration") start = roots[0].start_time rows = [] for span in sorted(finished, key=lambda item: item.start_time): rows.append({ "name": span.name, "trace_id": format(span.context.trace_id, "032x"), "span_id": format(span.context.span_id, "016x"), "parent_id": format(span.parent.span_id, "016x") if span.parent else None, "start_ms": round((span.start_time - start) / 1_000_000, 6), "duration_ms": round((span.end_time - span.start_time) / 1_000_000, 6), "status": span.status.status_code.name, "attributes": dict(span.attributes), "events": [{"name": event.name, "at_ms": round((event.timestamp - start) / 1_000_000, 6), "attributes": dict(event.attributes)} for event in span.events], }) return { "id": "gap" if omit_lookup_span and case == "success" else case, "label": "Lookup span omitted" if omit_lookup_span else LABELS[case], "case": case, "lookup_instrumented": not omit_lookup_span, "result": reply["route"], # Keep response text out of telemetry and its report. "spans": rows, } def report(): return { "schema": SCHEMA, "recorded_at": datetime.now(timezone.utc).isoformat(), "packages": {name: version(name) for name in ("opentelemetry-api", "opentelemetry-sdk")}, "provenance": "Actual local OpenTelemetry spans over original synthetic Python functions. Deliberate sleeps make timing visible; these are not service or model benchmarks. IDs and durations change each run. No model, collector, remote service or production telemetry.", "runs": [trace_case(case) for case in CASES] + [trace_case("success", True)], } class FirstTraceTests(unittest.TestCase): def test_parent_child_identity_and_intervals(self): result = trace_case("success") root, lookup, reply = result["spans"] self.assertEqual(result["result"], "guide") self.assertEqual([r["name"] for r in result["spans"]], ["support.request", "guide.lookup", "reply.format"]) self.assertIsNone(root["parent_id"]) self.assertEqual(len({r["span_id"] for r in result["spans"]}), 3) for child in (lookup, reply): self.assertEqual(child["parent_id"], root["span_id"]) self.assertEqual(child["trace_id"], root["trace_id"]) self.assertGreater(child["duration_ms"], 0) self.assertLessEqual(child["start_ms"] + child["duration_ms"], root["duration_ms"] + 0.000002) self.assertLessEqual(lookup["start_ms"] + lookup["duration_ms"], reply["start_ms"] + 0.000002) def test_missing_guide_is_not_an_exception(self): result = trace_case("missing") self.assertEqual(result["result"], "handoff") self.assertTrue(all(row["status"] == "UNSET" for row in result["spans"])) self.assertEqual(result["spans"][1]["attributes"], {"training.found": False}) def test_failure_is_visible_and_fallback_preserves_application_result(self): result = trace_case("failure") root, lookup, fallback = result["spans"] self.assertEqual(result["result"], "handoff") self.assertEqual(lookup["status"], "ERROR") self.assertEqual(lookup["events"][0]["attributes"], {"error.type": "lookup_unavailable"}) self.assertEqual((root["status"], fallback["status"]), ("UNSET", "UNSET")) self.assertNotIn("SYNTHETIC_PRIVATE_DETAIL", json.dumps(result)) def test_omitted_span_does_not_omit_the_work(self): with patch(__name__ + ".lookup_guide", wraps=lookup_guide) as lookup: result = trace_case("success", True) lookup.assert_called_once_with("success") self.assertEqual(result["result"], "guide") self.assertEqual([row["name"] for row in result["spans"]], ["support.request", "reply.format"]) def test_no_network_or_global_provider_change(self): from opentelemetry import trace original = trace.get_tracer_provider() with without_network(), patch.dict("os.environ", { "OTEL_RESOURCE_ATTRIBUTES": "private.label=SYNTHETIC_SECRET", "OTEL_EXPORTER_OTLP_ENDPOINT": "https://collector.invalid", }): result = report() self.assertIs(trace.get_tracer_provider(), original) self.assertNotIn("SYNTHETIC_SECRET", json.dumps(result)) self.assertNotIn(GUIDES["P1"], json.dumps(result)) def test_disabled_sdk_does_not_claim_export_success(self): with patch.dict("os.environ", {"OTEL_SDK_DISABLED": "true"}): with self.assertRaisesRegex(RuntimeError, "trace_incomplete"): trace_case("success") def test_unknown_cases_and_ambiguous_flags_are_rejected(self): for case, omitted in [("other", False), ("success", "yes")]: with self.assertRaises(ValueError): trace_case(case, omitted) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=["test", "compare", "run"]) parser.add_argument("--case", choices=CASES, default="success") parser.add_argument("--omit-lookup-span", action="store_true") args = parser.parse_args() try: with without_network(): if args.command == "test": suite = unittest.defaultTestLoader.loadTestsFromTestCase(FirstTraceTests) return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1 result = report() if args.command == "compare" else trace_case(args.case, args.omit_lookup_span) print(json.dumps(result, indent=2)) return 0 except ImportError: print(json.dumps({"status": "setup_required", "reason": "install_fde_first_trace_requirements"})) return 1 except (RuntimeError, ValueError): print(json.dumps({"status": "incomplete", "reason": "check_case_and_sdk_configuration"})) return 1 if __name__ == "__main__": raise SystemExit(main())