From abf4b90654d9552475f5256f6e8f89b2c1733b89 Mon Sep 17 00:00:00 2001 From: JS Date: Wed, 21 Oct 2026 11:27:48 +0800 Subject: [PATCH 1/2] feat(tracing): boot journey + process device for out-of-request calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every SDK call is a user request — human or agent (ruling, campus-api-python#114) — so calls outside any request context (startup, background threads) now carry: - X-Journey-ID: one uid-journey-… per process boot, lazily minted on first use, fixed for the process's lifetime (format matches campus.common.utils.uid.generate_category_uid without importing the server package). - X-Journey-Name: only when the app names the journey (set_process_journey(name=...)), riding the campus#952 adoption. - X-Campus-Device: the process device, defaulting to RAILWAY_DEPLOYMENT_ID, so boot spans attribute to the deploy that produced them. Strict MVP scope (ruled): in-request calls never inherit the boot journey — a journeyed request forwards its own, a journey-less request stays unjourneyed (the trace waterfall already correlates it) — and the process device only fills the gap when the request stashed none. Opt-out, both ruled in: CAMPUS_PROCESS_JOURNEY=off (also false/0/no; a non-off value pins an explicit id) for pytest/CI, and disable_process_journey() in code; set_process_journey() re-enables and set_process_device() overrides. Fixes #114. --- campus_python/tracing.py | 212 +++++++++++++++++++++++--- tests/unit/test_trace_propagation.py | 214 +++++++++++++++++++++++++++ 2 files changed, 405 insertions(+), 21 deletions(-) diff --git a/campus_python/tracing.py b/campus_python/tracing.py index e9398cb..1a4dafa 100644 --- a/campus_python/tracing.py +++ b/campus_python/tracing.py @@ -14,11 +14,24 @@ names, same flask.g contract (trace_id / span_id stashed by the middleware's before_request hook, and the same _campus_trace_instrumented idempotency marker). It is deliberately self-contained so the SDK does -not import the server-side audit client stack just to emit two headers — +not import the server-side audit client stack just to emit headers — keep the two in lockstep; W3C traceparent interop would extend both. + +Boot journeys (#114): every SDK call is a user request, whether that +user is human or agent — so calls made outside any request context +(startup, background threads) carry a process journey, one +uid-journey-… per process boot, plus a process device tag (default: +RAILWAY_DEPLOYMENT_ID) so boot spans attribute to the deploy that +produced them. Inside a request the caller's own journey always wins; +strict fallback only fills the out-of-request gap (campus-api-python#114 +ruling, 2026-10-21). Set CAMPUS_PROCESS_JOURNEY=off to suppress (tests, +operators); set_process_journey()/set_process_device() override in code. """ +import os +import threading import typing +import uuid import requests @@ -39,6 +52,144 @@ # end to end. Keep the two in lockstep. JOURNEY_NAME_HEADER = "X-Journey-Name" +# Environment for the boot journey (#114): off/false/0/no disables it +# (tests, operators), any other non-empty value pins an explicit +# uid-journey-… id. Unset: auto-mint one per process boot. +PROCESS_JOURNEY_ENV = "CAMPUS_PROCESS_JOURNEY" +_PROCESS_JOURNEY_OFF = frozenset({"off", "false", "0", "no"}) + +# Railway injects the deployment id into every container: the default +# process device tag (#114), so boot spans attribute to the deploy that +# produced them (receiver-side preference is unchanged, campus#837). +PROCESS_DEVICE_ENV = "RAILWAY_DEPLOYMENT_ID" + +# Boot-journey state (#114): None until first consultation, then fixed +# for the process's lifetime. A lock guards init/mint; reads take the +# fast path (initialized flag) lock-free. +_process_lock = threading.Lock() +_process_state: dict = { + "initialized": False, + "disabled": False, + "journey_id": None, + "journey_name": None, + "device_id": None, +} + + +def _mint_journey_id() -> str: + """uid-journey-… matching campus.common.utils.uid.generate_category_uid + (first 16 chars of the hyphenated uuid4 string) — replicated rather + than imported so the SDK stays decoupled from the server package.""" + return f"uid-journey-{str(uuid.uuid4())[:16]}" + + +def _init_process_state() -> None: + """Initialize from the environment once (caller holds the lock).""" + raw = (os.environ.get(PROCESS_JOURNEY_ENV) or "").strip() + if raw.lower() in _PROCESS_JOURNEY_OFF: + _process_state["disabled"] = True + elif raw: + _process_state["journey_id"] = raw + _process_state["device_id"] = os.environ.get(PROCESS_DEVICE_ENV) or None + _process_state["initialized"] = True + + +def _process_device() -> str | None: + """The process device tag after lazy env init (no journey minting).""" + if not _process_state["initialized"]: + with _process_lock: + if not _process_state["initialized"]: + _init_process_state() + return _process_state["device_id"] + + +def _process_snapshot() -> tuple[str | None, str | None, str | None]: + """(journey_id, journey_name, device_id) after lazy init and mint. + + The boot journey mints on first use and stays fixed for the + process's lifetime (one journey per boot, #114). Hosts that only + read the device (in-request gap-fill) never mint via + _process_device(). + """ + if not _process_state["initialized"]: + with _process_lock: + if not _process_state["initialized"]: + _init_process_state() + if _process_state["journey_id"] is None and not _process_state["disabled"]: + with _process_lock: + if _process_state["journey_id"] is None: + _process_state["journey_id"] = _mint_journey_id() + return ( + _process_state["journey_id"], + _process_state["journey_name"], + _process_state["device_id"], + ) + + +def set_process_journey( + journey_id: str | None = None, + *, + name: str | None = None, +) -> str: + """Set this process's boot journey explicitly (#114). + + Overrides CAMPUS_PROCESS_JOURNEY and re-enables journeys after + disable_process_journey(). Mints a uid-journey-… when journey_id is + omitted (keeping any previously minted/explicit id on later calls); + a name labels the journey's cards (forwarded as X-Journey-Name, + campus#952). + + Returns: + The process journey id now in effect. + """ + with _process_lock: + if not _process_state["initialized"]: + _init_process_state() + _process_state["disabled"] = False + if _process_state["journey_id"] is None: + _process_state["journey_id"] = journey_id or _mint_journey_id() + elif journey_id is not None: + _process_state["journey_id"] = journey_id + if name is not None: + _process_state["journey_name"] = name + return _process_state["journey_id"] + + +def disable_process_journey() -> None: + """Suppress the boot journey on out-of-request calls (#114). + + For tests and operators who don't want the correlation; the process + device tag is unaffected. set_process_journey() re-enables. + """ + with _process_lock: + if not _process_state["initialized"]: + _init_process_state() + _process_state["disabled"] = True + + +def set_process_device(device_id: str) -> None: + """Override this process's device tag (default: RAILWAY_DEPLOYMENT_ID). + + Used as a fallback only — a request context with its own stashed + device keeps it (campus#825/#837 semantics are unchanged). + """ + with _process_lock: + if not _process_state["initialized"]: + _init_process_state() + _process_state["device_id"] = device_id + + +def _reset_process_journey() -> None: + """Reset all boot-journey state (test hook; also re-reads env).""" + with _process_lock: + _process_state.update( + initialized=False, + disabled=False, + journey_id=None, + journey_name=None, + device_id=None, + ) + # Mirror of campus.config.DEVICE_ID_HEADER (campus#837): forwarded so a # child service's span carries the caller's device identity when the # host stashed one (flask_campus's push_context sets g.device from the @@ -75,41 +226,59 @@ def current_context() -> tuple[str, str] | None: def propagation_headers() -> dict[str, str]: - """Headers to attach to an outbound SDK call from the active request. + """Headers to attach to an outbound SDK call from the active context. - Empty outside a request context. The receiving campus service's + Empty outside a request context until #114 — now: out-of-request + calls carry the process boot journey (X-Journey-ID/X-Journey-Name) + and the process device (X-Campus-Device, default RAILWAY_DEPLOYMENT_ID), + so startup and background calls correlate under one journey per boot. + Inside a request, the caller's own journey always wins — a journey-less + request stays unjourneyed (strict MVP: the trace waterfall already + correlates the chain) — and the process device only fills the gap + when the request stashed none. The receiving campus service's tracing middleware turns the trace headers into a child span of the caller's span (#794, campus#816); the journeys middleware adopts X-Journey-ID so the child span joins the caller's action journey (campus#828), with X-Journey-Name carrying its human label (campus#952); the tracing middleware adopts X-Campus-Device for the child span's device tag when it knows nothing better (campus#837). - The three are independent: trace headers require span state on - flask.g (tracing middleware ran), the journey headers only require - an active journey, the device header only requires a stashed - g.device. """ headers: dict[str, str] = {} try: import flask except ImportError: # pragma: no cover - host without flask + flask = None + in_request = flask is not None and flask.has_request_context() + + if in_request: + trace_id = getattr(flask.g, "trace_id", None) + span_id = getattr(flask.g, "span_id", None) + if trace_id and span_id: + headers[TRACE_ID_HEADER] = trace_id + headers[PARENT_SPAN_ID_HEADER] = span_id + journey_id = getattr(flask.g, "journey_id", None) + if journey_id: + headers[JOURNEY_ID_HEADER] = journey_id + journey_name = getattr(flask.g, "journey_name", None) + if journey_name: + headers[JOURNEY_NAME_HEADER] = journey_name + device_id = getattr(flask.g, "device", None) + if not device_id: + # Gap-fill only: a request with its own device keeps it. + device_id = _process_device() + if device_id: + headers[DEVICE_ID_HEADER] = str(device_id) return headers - if not flask.has_request_context(): - return headers - trace_id = getattr(flask.g, "trace_id", None) - span_id = getattr(flask.g, "span_id", None) - if trace_id and span_id: - headers[TRACE_ID_HEADER] = trace_id - headers[PARENT_SPAN_ID_HEADER] = span_id - journey_id = getattr(flask.g, "journey_id", None) + + # Out of any request context (startup, background threads): the + # boot journey and process device are the only correlation there is. + journey_id, journey_name, device_id = _process_snapshot() if journey_id: headers[JOURNEY_ID_HEADER] = journey_id - journey_name = getattr(flask.g, "journey_name", None) if journey_name: headers[JOURNEY_NAME_HEADER] = journey_name - device_id = getattr(flask.g, "device", None) if device_id: - headers[DEVICE_ID_HEADER] = str(device_id) + headers[DEVICE_ID_HEADER] = device_id return headers @@ -117,9 +286,10 @@ def instrument_requests_session(session: requests.Session) -> bool: """Wrap a requests.Session so its calls carry campus trace context. Headers are computed at call time from the host's active request - context, so a shared session stays correct under concurrent requests, - and calls made outside a traced request (startup, background threads) - are left untouched. + context, so a shared session stays correct under concurrent requests. + Calls outside a request context (startup, background threads) carry + the process boot journey and device (#114); calls inside a request + never inherit it (the caller's journey, or none, wins). Idempotent: re-instrumenting a session is a no-op. diff --git a/tests/unit/test_trace_propagation.py b/tests/unit/test_trace_propagation.py index 5937300..084850f 100644 --- a/tests/unit/test_trace_propagation.py +++ b/tests/unit/test_trace_propagation.py @@ -7,6 +7,7 @@ nothing when there is no active traced request. """ +import os import unittest import flask @@ -227,6 +228,219 @@ def test_name_lockstep_with_campus_middleware(self): self.assertEqual(tracing.JOURNEY_NAME_HEADER, "X-Journey-Name") +class TestProcessJourney(unittest.TestCase): + """Boot journey + process device on out-of-request calls (#114). + + Ruling (2026-10-21): every SDK call is a user request — human or + agent — so out-of-request calls (startup, background threads) + correlate under one journey per process boot, attributed to the + operator. Strict MVP: in-request calls never inherit the journey. + """ + + def setUp(self): + tracing._reset_process_journey() + self.addCleanup(tracing._reset_process_journey) + # Snapshot and clear both env knobs so tests are deterministic + # regardless of the host environment. + self._saved_env = { + key: os.environ.get(key) + for key in (tracing.PROCESS_JOURNEY_ENV, tracing.PROCESS_DEVICE_ENV) + } + for key in self._saved_env: + os.environ.pop(key, None) + self.addCleanup(self._restore_env) + + def _restore_env(self): + for key, value in self._saved_env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + def _client(self): + client = _make_client() + self.captured = _capture_send(client) + return client + + def test_out_of_request_call_carries_boot_journey(self): + client = self._client() + + client.get("/ping") + + journey_id = self.captured["headers"].get("X-Journey-ID") + self.assertTrue(journey_id.startswith("uid-journey-")) + # Auto-minted: no name until the app sets one. + self.assertNotIn("X-Journey-Name", self.captured["headers"]) + + def test_one_journey_per_process(self): + """The minted id stays fixed across calls — one journey per boot.""" + client = self._client() + + client.get("/ping") + first = self.captured["headers"].get("X-Journey-ID") + client.get("/ping") + + self.assertEqual(self.captured["headers"].get("X-Journey-ID"), first) + + def test_explicit_id_and_name_via_api(self): + tracing.set_process_journey("uid-journey-custom01", name="campus.api.boot") + client = self._client() + + client.get("/ping") + + self.assertEqual( + self.captured["headers"].get("X-Journey-ID"), "uid-journey-custom01" + ) + self.assertEqual( + self.captured["headers"].get("X-Journey-Name"), "campus.api.boot" + ) + + def test_set_process_journey_mints_and_returns_id(self): + journey_id = tracing.set_process_journey(name="campus.yapper.boot") + + self.assertTrue(journey_id.startswith("uid-journey-")) + client = self._client() + client.get("/ping") + self.assertEqual(self.captured["headers"].get("X-Journey-ID"), journey_id) + self.assertEqual( + self.captured["headers"].get("X-Journey-Name"), "campus.yapper.boot" + ) + + def test_env_opt_out_suppresses_journey(self): + os.environ[tracing.PROCESS_JOURNEY_ENV] = "off" + client = self._client() + + client.get("/ping") + + self.assertNotIn("X-Journey-ID", self.captured["headers"]) + self.assertNotIn("X-Journey-Name", self.captured["headers"]) + + def test_disable_api_suppresses_journey(self): + tracing.disable_process_journey() + client = self._client() + + client.get("/ping") + + self.assertNotIn("X-Journey-ID", self.captured["headers"]) + + def test_set_process_journey_reenables_after_disable(self): + tracing.disable_process_journey() + tracing.set_process_journey(name="campus.api.boot") + client = self._client() + + client.get("/ping") + + self.assertTrue(self.captured["headers"].get("X-Journey-ID", "").startswith( + "uid-journey-" + )) + + def test_env_explicit_id_pinned(self): + """A non-off CAMPUS_PROCESS_JOURNEY value pins that id (e.g. to + correlate a known deploy) instead of auto-minting.""" + os.environ[tracing.PROCESS_JOURNEY_ENV] = "uid-journey-deploy42" + client = self._client() + + client.get("/ping") + + self.assertEqual( + self.captured["headers"].get("X-Journey-ID"), "uid-journey-deploy42" + ) + + def test_process_device_from_railway_env(self): + os.environ[tracing.PROCESS_DEVICE_ENV] = "deploy-abc123" + client = self._client() + + client.get("/ping") + + self.assertEqual( + self.captured["headers"].get("X-Campus-Device"), "deploy-abc123" + ) + + def test_in_request_journeyless_call_stays_unjourneyed(self): + """Strict MVP: no journey header inside a request that has none — + the trace waterfall already correlates the chain.""" + app = flask.Flask(__name__) + client = self._client() + tracing.set_process_journey(name="campus.api.boot") + + with app.test_request_context("/"): + client.get("/ping") + + self.assertNotIn("X-Journey-ID", self.captured["headers"]) + self.assertNotIn("X-Journey-Name", self.captured["headers"]) + + def test_in_request_journey_wins_over_process(self): + """A journeyed request forwards its own journey, never the boot's.""" + app = flask.Flask(__name__) + client = self._client() + boot_id = tracing.set_process_journey() + + with app.test_request_context("/"): + flask.g.journey_id = "uid-journey-request1" + flask.g.journey_name = "campus.submissions.submit" + client.get("/ping") + + self.assertEqual( + self.captured["headers"].get("X-Journey-ID"), "uid-journey-request1" + ) + self.assertEqual( + self.captured["headers"].get("X-Journey-Name"), + "campus.submissions.submit", + ) + self.assertNotEqual( + self.captured["headers"].get("X-Journey-ID"), boot_id + ) + + def test_in_request_device_gap_filled_by_process(self): + """A request with no stashed device attributes to the deploy.""" + app = flask.Flask(__name__) + client = self._client() + os.environ[tracing.PROCESS_DEVICE_ENV] = "deploy-abc123" + + with app.test_request_context("/"): + flask.g.trace_id = TRACE_ID + flask.g.span_id = SPAN_ID + client.get("/ping") + + self.assertEqual( + self.captured["headers"].get("X-Campus-Device"), "deploy-abc123" + ) + + def test_in_request_stashed_device_wins(self): + app = flask.Flask(__name__) + client = self._client() + os.environ[tracing.PROCESS_DEVICE_ENV] = "deploy-abc123" + + with app.test_request_context("/"): + flask.g.device = "uid-device-session1" + client.get("/ping") + + self.assertEqual( + self.captured["headers"].get("X-Campus-Device"), "uid-device-session1" + ) + + def test_set_process_device_overrides_env(self): + os.environ[tracing.PROCESS_DEVICE_ENV] = "deploy-abc123" + tracing.set_process_device("deploy-xyz789") + client = self._client() + + client.get("/ping") + + self.assertEqual( + self.captured["headers"].get("X-Campus-Device"), "deploy-xyz789" + ) + + def test_minted_id_matches_campus_format(self): + """uid-journey-, matching + campus.common.utils.uid.generate_category_uid("journey").""" + journey_id = tracing._mint_journey_id() + + self.assertTrue(journey_id.startswith("uid-journey-")) + suffix = journey_id.removeprefix("uid-journey-") + self.assertEqual(len(suffix), 16) + self.assertEqual(suffix.count("-"), 2) + + class TestDeviceForwarding(unittest.TestCase): """Device forwarding on SDK calls (campus#837). From 9b8208754e296bb68db3c971d90947b4b1833ff7 Mon Sep 17 00:00:00 2001 From: JS Date: Wed, 21 Oct 2026 11:31:25 +0800 Subject: [PATCH 2/2] fix(tracing): narrow flask import directly so pyright sees non-None g The in_request flag broke type narrowing: pyright could not connect the bool to flask being non-None inside the request branch. Branch on flask None-ness directly and return from each path. --- campus_python/tracing.py | 37 ++++++++++++++++++------------------- 1 file changed, 18 insertions(+), 19 deletions(-) diff --git a/campus_python/tracing.py b/campus_python/tracing.py index 1a4dafa..52feb37 100644 --- a/campus_python/tracing.py +++ b/campus_python/tracing.py @@ -248,37 +248,36 @@ def propagation_headers() -> dict[str, str]: import flask except ImportError: # pragma: no cover - host without flask flask = None - in_request = flask is not None and flask.has_request_context() - - if in_request: - trace_id = getattr(flask.g, "trace_id", None) - span_id = getattr(flask.g, "span_id", None) - if trace_id and span_id: - headers[TRACE_ID_HEADER] = trace_id - headers[PARENT_SPAN_ID_HEADER] = span_id - journey_id = getattr(flask.g, "journey_id", None) + + if flask is None or not flask.has_request_context(): + # Out of any request context (startup, background threads): the + # boot journey and process device are the only correlation there is. + journey_id, journey_name, device_id = _process_snapshot() if journey_id: headers[JOURNEY_ID_HEADER] = journey_id - journey_name = getattr(flask.g, "journey_name", None) if journey_name: headers[JOURNEY_NAME_HEADER] = journey_name - device_id = getattr(flask.g, "device", None) - if not device_id: - # Gap-fill only: a request with its own device keeps it. - device_id = _process_device() if device_id: - headers[DEVICE_ID_HEADER] = str(device_id) + headers[DEVICE_ID_HEADER] = device_id return headers - # Out of any request context (startup, background threads): the - # boot journey and process device are the only correlation there is. - journey_id, journey_name, device_id = _process_snapshot() + trace_id = getattr(flask.g, "trace_id", None) + span_id = getattr(flask.g, "span_id", None) + if trace_id and span_id: + headers[TRACE_ID_HEADER] = trace_id + headers[PARENT_SPAN_ID_HEADER] = span_id + journey_id = getattr(flask.g, "journey_id", None) if journey_id: headers[JOURNEY_ID_HEADER] = journey_id + journey_name = getattr(flask.g, "journey_name", None) if journey_name: headers[JOURNEY_NAME_HEADER] = journey_name + device_id = getattr(flask.g, "device", None) + if not device_id: + # Gap-fill only: a request with its own device keeps it. + device_id = _process_device() if device_id: - headers[DEVICE_ID_HEADER] = device_id + headers[DEVICE_ID_HEADER] = str(device_id) return headers