From d3b9eaba85e1d2471e2bf34bff2b43415a07086b Mon Sep 17 00:00:00 2001 From: sarathms-kv Date: Mon, 9 Nov 2026 11:22:16 +0530 Subject: [PATCH 1/3] feat: Implement prompt tracing functionality with context management and span tagging --- README.md | 15 ++++ netra/processors/session_span_processor.py | 3 + netra/prompts/api.py | 3 + netra/prompts/context.py | 56 +++++++++++++++ tests/test_prompts_context.py | 79 ++++++++++++++++++++++ 5 files changed, 156 insertions(+) create mode 100644 netra/prompts/context.py create mode 100644 tests/test_prompts_context.py diff --git a/README.md b/README.md index fc25f25a..dfe9f919 100644 --- a/README.md +++ b/README.md @@ -360,6 +360,21 @@ prompt = Netra.prompts.get_prompt("my-prompt", use_cache=True, cache_ttl=300) Netra.prompts.clear_cache() ``` +### Linking prompts to traces + +Wrap your LLM calls in `use_prompt` to tag every span started inside the block with the prompt's identity (`netra.prompt.id`, `netra.prompt.name`, `netra.prompt.version`, `netra.prompt.label`): + +```python +prompt = Netra.prompts.get_prompt("my-prompt", label="production") + +with Netra.prompts.use_prompt(prompt): + client.chat.completions.create(model="gpt-4o", messages=prompt["messages"]) +``` + +- Pass the value returned by `get_prompt` as-is. A failed fetch (empty result) links nothing and does not raise. +- Context is carried by `contextvars`, so spans in threads started with `threading.Thread` are not linked. `asyncio` tasks and `ThreadPoolExecutor` with copied context are. +- Requires a Netra backend that returns prompt identity from `get_prompt`; older backends link nothing. + Caching notes: - `use_cache` defaults to `False`; enable it per call when you want caching. diff --git a/netra/processors/session_span_processor.py b/netra/processors/session_span_processor.py index 6d80049f..cfe9ee29 100644 --- a/netra/processors/session_span_processor.py +++ b/netra/processors/session_span_processor.py @@ -7,6 +7,7 @@ from netra.config import Config from netra.netra_logger import get_logger +from netra.prompts.context import get_active_prompt_attributes from netra.session_manager import ATTR_SESSION_ID, ATTR_TENANT_ID, ATTR_USER_ID, SessionManager logger = get_logger(__name__) @@ -78,6 +79,8 @@ def on_start(self, span: trace.Span, parent_context: Optional[otel_context.Conte span.set_attribute("library.name", Config.LIBRARY_NAME) span.set_attribute("library.version", Config.LIBRARY_VERSION) span.set_attribute("sdk.name", Config.SDK_NAME) + for key, value in get_active_prompt_attributes().items(): + span.set_attribute(key, value) if session_id: span.set_attribute(ATTR_SESSION_ID, session_id) diff --git a/netra/prompts/api.py b/netra/prompts/api.py index f1f80b5e..39988a45 100644 --- a/netra/prompts/api.py +++ b/netra/prompts/api.py @@ -4,6 +4,7 @@ from netra.config import Config from netra.netra_logger import NoticeLevel, get_logger from netra.prompts.client import PromptsHttpClient +from netra.prompts.context import use_prompt logger = get_logger(__name__) @@ -26,6 +27,8 @@ def __init__(self, cfg: Config) -> None: self._client = PromptsHttpClient(cfg) self._cache: TTLCache[Any] = TTLCache(default_ttl=PROMPT_CACHE_TTL_SECONDS) + use_prompt = staticmethod(use_prompt) + def clear_cache(self) -> None: """Clear all cached prompt entries.""" self._cache.clear() diff --git a/netra/prompts/context.py b/netra/prompts/context.py new file mode 100644 index 00000000..99ddcd4f --- /dev/null +++ b/netra/prompts/context.py @@ -0,0 +1,56 @@ +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Any, Dict, Iterator, Mapping, Optional + +from netra.config import Config + +ATTR_PROMPT_ID = f"{Config.LIBRARY_NAME}.prompt.id" +ATTR_PROMPT_NAME = f"{Config.LIBRARY_NAME}.prompt.name" +ATTR_PROMPT_VERSION = f"{Config.LIBRARY_NAME}.prompt.version" +ATTR_PROMPT_LABEL = f"{Config.LIBRARY_NAME}.prompt.label" + +# Prompt identity tagged onto every span started inside `use_prompt`. +_active_prompt: ContextVar[Optional[Mapping[str, Any]]] = ContextVar("netra_active_prompt", default=None) + + +def get_active_prompt_attributes() -> Dict[str, Any]: + """ + Build span attributes for the prompt set by `use_prompt`. + + Returns: + Span attributes, or an empty dict when no prompt is active or it carries no identity + """ + prompt = _active_prompt.get() + if not isinstance(prompt, Mapping): + return {} + + prompt_id = prompt.get("promptId") + version = prompt.get("versionNumber") + if not prompt_id or version is None: + return {} + + attributes: Dict[str, Any] = {ATTR_PROMPT_ID: prompt_id, ATTR_PROMPT_VERSION: version} + if prompt.get("promptName"): + attributes[ATTR_PROMPT_NAME] = prompt["promptName"] + if prompt.get("label"): + attributes[ATTR_PROMPT_LABEL] = prompt["label"] + return attributes + + +@contextmanager +def use_prompt(prompt: Optional[Mapping[str, Any]]) -> Iterator[None]: + """ + Link every span started inside the block to a prompt version. + + Context variables do not cross into threads started with `threading.Thread`; + spans created there are not linked. + + Args: + prompt: The value returned by `Netra.prompts.get_prompt`. A missing or + empty value is accepted and links nothing. + """ + token = _active_prompt.set(prompt) + try: + yield + finally: + _active_prompt.reset(token) diff --git a/tests/test_prompts_context.py b/tests/test_prompts_context.py new file mode 100644 index 00000000..ecf50d5d --- /dev/null +++ b/tests/test_prompts_context.py @@ -0,0 +1,79 @@ +import pytest +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from netra.processors.session_span_processor import SessionSpanProcessor +from netra.prompts.context import ( + ATTR_PROMPT_ID, + ATTR_PROMPT_LABEL, + ATTR_PROMPT_NAME, + ATTR_PROMPT_VERSION, + get_active_prompt_attributes, + use_prompt, +) + +PROMPT = { + "messages": [{"role": "user", "content": "hi"}], + "promptId": "p1", + "promptName": "welcome", + "versionNumber": 3, + "label": "production", +} + + +class TestUsePrompt: + def test_no_active_prompt_yields_no_attributes(self) -> None: + assert get_active_prompt_attributes() == {} + + def test_active_prompt_yields_identity_attributes(self) -> None: + with use_prompt(PROMPT): + assert get_active_prompt_attributes() == { + ATTR_PROMPT_ID: "p1", + ATTR_PROMPT_NAME: "welcome", + ATTR_PROMPT_VERSION: 3, + ATTR_PROMPT_LABEL: "production", + } + assert get_active_prompt_attributes() == {} + + @pytest.mark.parametrize("prompt", [None, {}, {"promptId": "p1"}, {"versionNumber": 1}, "not-a-mapping"]) + def test_untrackable_prompt_links_nothing(self, prompt: object) -> None: + with use_prompt(prompt): # type: ignore[arg-type] + assert get_active_prompt_attributes() == {} + + def test_version_zero_is_kept(self) -> None: + with use_prompt({"promptId": "p1", "versionNumber": 0}): + assert get_active_prompt_attributes()[ATTR_PROMPT_VERSION] == 0 + + def test_nested_blocks_restore_outer_prompt(self) -> None: + inner = {**PROMPT, "promptId": "p2"} + with use_prompt(PROMPT): + with use_prompt(inner): + assert get_active_prompt_attributes()[ATTR_PROMPT_ID] == "p2" + assert get_active_prompt_attributes()[ATTR_PROMPT_ID] == "p1" + + def test_prompt_reset_when_block_raises(self) -> None: + with pytest.raises(RuntimeError): + with use_prompt(PROMPT): + raise RuntimeError("boom") + assert get_active_prompt_attributes() == {} + + +class TestSpanTagging: + def test_spans_inside_block_are_tagged_and_outside_are_not(self) -> None: + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SessionSpanProcessor()) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("test") + + with use_prompt(PROMPT): + with tracer.start_as_current_span("inside"): + pass + with tracer.start_as_current_span("outside"): + pass + + spans = {s.name: s.attributes for s in exporter.get_finished_spans()} + assert spans["inside"][ATTR_PROMPT_ID] == "p1" + assert spans["inside"][ATTR_PROMPT_VERSION] == 3 + assert ATTR_PROMPT_ID not in spans["outside"] From d8251d295edccc059bfb903961a10e7859232092 Mon Sep 17 00:00:00 2001 From: sarathms-kv Date: Tue, 10 Nov 2026 10:14:53 +0530 Subject: [PATCH 2/3] feat: Enhance prompt context management for threading with OpenTelemetry integration --- netra/prompts/context.py | 25 +++++++++++++++++-------- tests/test_prompts_context.py | 14 ++++++++++++++ 2 files changed, 31 insertions(+), 8 deletions(-) diff --git a/netra/prompts/context.py b/netra/prompts/context.py index 99ddcd4f..6f8cfff6 100644 --- a/netra/prompts/context.py +++ b/netra/prompts/context.py @@ -1,16 +1,22 @@ +import logging from contextlib import contextmanager -from contextvars import ContextVar from typing import Any, Dict, Iterator, Mapping, Optional +from opentelemetry import context as context_api + from netra.config import Config +logger = logging.getLogger(__name__) + ATTR_PROMPT_ID = f"{Config.LIBRARY_NAME}.prompt.id" ATTR_PROMPT_NAME = f"{Config.LIBRARY_NAME}.prompt.name" ATTR_PROMPT_VERSION = f"{Config.LIBRARY_NAME}.prompt.version" ATTR_PROMPT_LABEL = f"{Config.LIBRARY_NAME}.prompt.label" -# Prompt identity tagged onto every span started inside `use_prompt`. -_active_prompt: ContextVar[Optional[Mapping[str, Any]]] = ContextVar("netra_active_prompt", default=None) +# Prompt identity tagged onto every span started inside `use_prompt`. Stored in the +# OTel context (not a bare ContextVar) so the threading instrumentation carries it +# into ThreadPoolExecutor workers. +_ACTIVE_PROMPT_KEY = "netra.active_prompt" def get_active_prompt_attributes() -> Dict[str, Any]: @@ -20,13 +26,14 @@ def get_active_prompt_attributes() -> Dict[str, Any]: Returns: Span attributes, or an empty dict when no prompt is active or it carries no identity """ - prompt = _active_prompt.get() + prompt = context_api.get_value(_ACTIVE_PROMPT_KEY) if not isinstance(prompt, Mapping): return {} prompt_id = prompt.get("promptId") version = prompt.get("versionNumber") if not prompt_id or version is None: + logger.debug("Active prompt has no promptId/versionNumber; spans will not be linked: %s", list(prompt)) return {} attributes: Dict[str, Any] = {ATTR_PROMPT_ID: prompt_id, ATTR_PROMPT_VERSION: version} @@ -42,15 +49,17 @@ def use_prompt(prompt: Optional[Mapping[str, Any]]) -> Iterator[None]: """ Link every span started inside the block to a prompt version. - Context variables do not cross into threads started with `threading.Thread`; - spans created there are not linked. + The prompt lives in the OTel context, so it reaches worker threads only when + the threading instrumentation propagates that context (e.g. `ThreadPoolExecutor`). Args: prompt: The value returned by `Netra.prompts.get_prompt`. A missing or empty value is accepted and links nothing. """ - token = _active_prompt.set(prompt) + if not isinstance(prompt, Mapping): + logger.debug("use_prompt called without a prompt mapping (%s); nothing linked", type(prompt).__name__) + token = context_api.attach(context_api.set_value(_ACTIVE_PROMPT_KEY, prompt)) try: yield finally: - _active_prompt.reset(token) + context_api.detach(token) diff --git a/tests/test_prompts_context.py b/tests/test_prompts_context.py index ecf50d5d..2ece96ca 100644 --- a/tests/test_prompts_context.py +++ b/tests/test_prompts_context.py @@ -77,3 +77,17 @@ def test_spans_inside_block_are_tagged_and_outside_are_not(self) -> None: assert spans["inside"][ATTR_PROMPT_ID] == "p1" assert spans["inside"][ATTR_PROMPT_VERSION] == 3 assert ATTR_PROMPT_ID not in spans["outside"] + + +def test_prompt_propagates_into_threadpool_with_threading_instrumentation() -> None: + from concurrent.futures import ThreadPoolExecutor + + from opentelemetry.instrumentation.threading import ThreadingInstrumentor + + ThreadingInstrumentor().instrument() + try: + with use_prompt(PROMPT), ThreadPoolExecutor(1) as pool: + attrs = pool.submit(get_active_prompt_attributes).result() + assert attrs[ATTR_PROMPT_ID] == "p1" + finally: + ThreadingInstrumentor().uninstrument() From d3dd1e7d97b64323b5261fc72e706aa5b050a0d2 Mon Sep 17 00:00:00 2001 From: sarathms-kv Date: Tue, 10 Nov 2026 10:33:28 +0530 Subject: [PATCH 3/3] feat: Enhance threading support for prompt propagation in use_prompt context --- README.md | 2 +- tests/test_prompts_context.py | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index dfe9f919..0ce0582e 100644 --- a/README.md +++ b/README.md @@ -372,7 +372,7 @@ with Netra.prompts.use_prompt(prompt): ``` - Pass the value returned by `get_prompt` as-is. A failed fetch (empty result) links nothing and does not raise. -- Context is carried by `contextvars`, so spans in threads started with `threading.Thread` are not linked. `asyncio` tasks and `ThreadPoolExecutor` with copied context are. +- Spans in `threading.Thread`, `Timer` and `ThreadPoolExecutor` workers are linked while Netra's threading instrumentation is enabled (the default). With it disabled, only `asyncio` tasks inherit the prompt. - Requires a Netra backend that returns prompt identity from `get_prompt`; older backends link nothing. Caching notes: diff --git a/tests/test_prompts_context.py b/tests/test_prompts_context.py index 2ece96ca..fdb29c2b 100644 --- a/tests/test_prompts_context.py +++ b/tests/test_prompts_context.py @@ -89,5 +89,14 @@ def test_prompt_propagates_into_threadpool_with_threading_instrumentation() -> N with use_prompt(PROMPT), ThreadPoolExecutor(1) as pool: attrs = pool.submit(get_active_prompt_attributes).result() assert attrs[ATTR_PROMPT_ID] == "p1" + + import threading + + out: dict = {} + with use_prompt(PROMPT): + t = threading.Thread(target=lambda: out.update(get_active_prompt_attributes())) + t.start() + t.join() + assert out[ATTR_PROMPT_ID] == "p1" finally: ThreadingInstrumentor().uninstrument()