diff --git a/docs/sdk-conformance.md b/docs/sdk-conformance.md index 4c0ba3a..09583dc 100644 --- a/docs/sdk-conformance.md +++ b/docs/sdk-conformance.md @@ -137,9 +137,11 @@ Concurrent calls share one fetch. returns the copy at once and starts one fetch in the background. Only a call with no copy waits for a fetch. - Timeout: a call waits for a fetch for 10 seconds at most, and then gets `timeout`. The - TypeScript SDK aborts the fetch at that time. The Python SDK fetches in a daemon thread, because - the urllib timeout applies to each socket operation and not to the address lookup or to the full - response. Thus the Python fetch can continue after the call stops waiting. + TypeScript SDK aborts the fetch at that time. The Python SDK fetches in a daemon thread and shuts + its connection down at that time, because the urllib timeout applies to each socket operation + and not to the full response. Only an address lookup can continue after the call stops waiting, + until the system resolver gives up. Both SDKs measure the TTL and Retry-After in elapsed time, so + a change of the system clock does not move them. - Failure: when a fetch fails, the SDKs keep the last copy for another TTL and send the error to `onError` / `on_error`. With no copy, the call gets the fallback. With no fallback, the call gets a `PromptError`. With no copy, the calls of the next 10 seconds (or the TTL, if it is shorter) diff --git a/packages/sdk/README.md b/packages/sdk/README.md index 986ba7c..afd5ef1 100644 --- a/packages/sdk/README.md +++ b/packages/sdk/README.md @@ -179,7 +179,8 @@ compiles its own. Model calls are the spans of type `generation` (operations `ch `text_completion`, and `generate_content`), including the spans of the provider integrations that use this SDK: OpenAI, Anthropic, Amazon Bedrock, Google GenAI, and OpenRouter. Spans from the AI SDK and TanStack AI integrations, and from other OpenTelemetry tracers, do not get the attributes. -Messages from `compile()` that you pass to a model call outside `withPrompt` are not recorded. +A model call outside `withPrompt` does not get the attributes, even when it sends messages from +`compile()`; with input capture on, the call still records those messages as its input. ## API diff --git a/packages/sdk/src/prompts.ts b/packages/sdk/src/prompts.ts index 682bb49..2e27b17 100644 --- a/packages/sdk/src/prompts.ts +++ b/packages/sdk/src/prompts.ts @@ -180,9 +180,9 @@ interface CacheEntry { response?: PromptResponse; /** Why the last fetch failed, when there was no response to keep. */ error?: unknown; - /** When the last fetch ended. */ + /** When the last fetch ended, in `performance.now()` time, so that a clock change does not move it. */ checkedAt: number; - /** No fetch before this time: the Retry-After of the last 429 or 503 response. */ + /** No fetch before this `performance.now()` time: the Retry-After of the last 429 or 503 response. */ retryAt: number; pending?: Promise; } @@ -406,12 +406,12 @@ async function load( if (!entry) { if (cache.size >= MAX_CACHE_ENTRIES) cache.delete(cache.keys().next().value!); - entry = { checkedAt: 0, retryAt: 0 }; + entry = { checkedAt: Number.NEGATIVE_INFINITY, retryAt: 0 }; cache.set(key, entry); } const fresh = - Date.now() < + performance.now() < Math.max( entry.checkedAt + (entry.response ? ttlMs : Math.min(ttlMs, ERROR_TTL_MS)), entry.retryAt, @@ -443,7 +443,7 @@ function refresh( }, (error: unknown) => { const delay = error instanceof PromptError ? retryAfter.get(error) : undefined; - entry.retryAt = delay === undefined ? 0 : Date.now() + delay; + entry.retryAt = delay === undefined ? 0 : performance.now() + delay; if (!entry.response) { entry.error = error; @@ -457,7 +457,7 @@ function refresh( ) .finally(() => { // A failed fetch counts too, so that calls do not each start a fetch while it fails. - entry.checkedAt = Date.now(); + entry.checkedAt = performance.now(); entry.pending = undefined; }); } diff --git a/packages/sdk/tests/prompts.test.ts b/packages/sdk/tests/prompts.test.ts index 1689392..6048d16 100644 --- a/packages/sdk/tests/prompts.test.ts +++ b/packages/sdk/tests/prompts.test.ts @@ -209,7 +209,7 @@ test.each([ ); test("getPrompt keeps a prompt for the TTL, then refreshes it in the background", async () => { - vi.useFakeTimers({ toFake: ["Date"] }); + vi.useFakeTimers({ toFake: ["Date", "performance"] }); const refresh = deferred(); const { fetch, calls } = fakeFetch(reply(), refresh.promise); const options = { apiKey: "td_live_a", baseUrl: newBaseUrl(), fetch }; @@ -238,6 +238,19 @@ test("getPrompt keeps a prompt for the TTL, then refreshes it in the background" expect(calls).toHaveLength(3); }); +test("getPrompt measures the TTL in elapsed time, not by the wall clock", async () => { + vi.useFakeTimers({ toFake: ["Date", "performance"], now: Date.parse("2026-10-29T12:00:00Z") }); + const { fetch, calls } = fakeFetch(reply()); + const options = { apiKey: "td_live_a", baseUrl: newBaseUrl(), fetch }; + + await getPrompt("support/reply", options); + // The system clock moves back an hour; 60 seconds still pass. + vi.setSystemTime(Date.parse("2026-10-29T11:00:00Z")); + vi.advanceTimersByTime(60_000); + await getPrompt("support/reply", options); + expect(calls).toHaveLength(2); +}); + test("concurrent calls share one fetch", async () => { const { fetch, calls } = fakeFetch(reply()); const options = { apiKey: "td_live_a", baseUrl: newBaseUrl(), fetch }; @@ -253,7 +266,7 @@ test("concurrent calls share one fetch", async () => { }); test("a failed refresh returns the last copy and reports the error", async () => { - vi.useFakeTimers({ toFake: ["Date"] }); + vi.useFakeTimers({ toFake: ["Date", "performance"] }); const errors: Error[] = []; vi.spyOn(console, "error").mockImplementation(() => {}); @@ -283,7 +296,7 @@ test("a failed refresh returns the last copy and reports the error", async () => }); test("with no copy, a failed fetch returns the fallback, and so do calls for 10 seconds", async () => { - vi.useFakeTimers({ toFake: ["Date"] }); + vi.useFakeTimers({ toFake: ["Date", "performance"] }); vi.spyOn(console, "error").mockImplementation(() => {}); const { fetch, calls } = fakeFetch( apiError(404, "label_not_found", "support/reply has no production label"), @@ -375,7 +388,7 @@ test.each([ ])( "with no copy, getPrompt does not fetch before the Retry-After ($status, $retryAfter)", async ({ status, retryAfter, wait }) => { - vi.useFakeTimers({ toFake: ["Date"], now: Date.parse("2026-10-29T12:00:00Z") }); + vi.useFakeTimers({ toFake: ["Date", "performance"], now: Date.parse("2026-10-29T12:00:00Z") }); vi.spyOn(console, "error").mockImplementation(() => {}); const header = retryAfter === "date" ? new Date(Date.now() + 120_000).toUTCString() : retryAfter; @@ -406,7 +419,7 @@ test.each([ ); test("a refresh that gets a Retry-After keeps the copy until then", async () => { - vi.useFakeTimers({ toFake: ["Date"] }); + vi.useFakeTimers({ toFake: ["Date", "performance"] }); vi.spyOn(console, "error").mockImplementation(() => {}); const { fetch, calls } = fakeFetch( diff --git a/sdks/python/README.md b/sdks/python/README.md index fe2a76c..b785d10 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -251,8 +251,9 @@ ran in it, or before it started in the code that started it; with none, it recor for code that has not compiled in its own task. Model calls are the spans of type `generation` (operations `chat`, `text_completion`, and `generate_content`), including the spans of the provider integrations: OpenAI, Anthropic, Amazon Bedrock, Google GenAI, LiteLLM, and OpenRouter. Spans from -other OpenTelemetry tracers do not get the attributes. Messages from `compile()` that you pass to a -model call outside `with_prompt` are not recorded. +other OpenTelemetry tracers do not get the attributes. A model call outside `with_prompt` does not +get the attributes, even when it sends messages from `compile()`; with input capture on, the call +still records those messages as its input. ## Auto-metrics diff --git a/sdks/python/src/telemetry_dev/_prompts.py b/sdks/python/src/telemetry_dev/_prompts.py index 5cf790d..9673740 100644 --- a/sdks/python/src/telemetry_dev/_prompts.py +++ b/sdks/python/src/telemetry_dev/_prompts.py @@ -1,11 +1,13 @@ from __future__ import annotations import asyncio +import contextlib import copy import hashlib import json import random import re +import socket import threading import urllib.error import urllib.parse @@ -274,7 +276,8 @@ def prompt_attributes( client: Client, context: Context | None, operation: object, capture_input: bool ) -> dict[str, AttributeValue]: """The td.prompt.* attributes for a span with this operation that starts in ``context``.""" - if operation not in _MODEL_CALL_OPERATIONS: + # An attribute can be a list, which a frozenset cannot hold. + if not isinstance(operation, str) or operation not in _MODEL_CALL_OPERATIONS: return {} scope = otel_context.get_value(_PROMPT_KEY, context) or otel_context.get_value(_PROMPT_KEY) if not isinstance(scope, _PromptScope): @@ -467,10 +470,9 @@ def _load( on_error: Callable[[BaseException], None] | None, ) -> dict[str, Any]: """Return the kept copy of a prompt, and fetch a new copy in a background thread when it is - older than the TTL. Only a call with no copy waits for a fetch, until the fetch's deadline: the - timeout of urllib applies to each socket operation, not to the whole fetch. A failed fetch with - no copy fails the calls of the next ``_ERROR_TTL_S`` (or the TTL, if it is shorter) without a - fetch.""" + older than the TTL. Only a call with no copy waits for a fetch, until the fetch's deadline. A + failed fetch with no copy fails the calls of the next ``_ERROR_TTL_S`` (or the TTL, if it is + shorter) without a fetch.""" with _cache_lock: entry = _cache.get(key) if entry is None: @@ -489,7 +491,7 @@ def _load( return entry.response pending, deadline = entry.pending, entry.deadline if pending is not None and not pending.wait(deadline - monotonic()): - raise PromptError("timeout", f"Fetching the prompt took more than {_FETCH_TIMEOUT_S:g} s") + raise _timeout() with _cache_lock: response, error = entry.response, entry.error if response is not None: @@ -548,31 +550,141 @@ def redirect_request( return None -_opener = urllib.request.build_opener(_NoRedirect) +class _Deadline: + """Ends one prompt fetch after ``_FETCH_TIMEOUT_S``. The timeout of urllib applies to each + socket operation, so a server that sends a byte now and then could keep a fetch reading forever, + and each address of a host gets the full timeout to connect. A connection attempt gets only the + time that is left, and at the deadline this shuts the connection down, which ends a blocked + read. It cannot stop an address lookup; the system resolver bounds that.""" + + def __init__(self) -> None: + self.expired = False + self._end = monotonic() + _FETCH_TIMEOUT_S + self._lock = threading.Lock() + self._sockets: list[socket.socket] = [] + self._timer = threading.Timer(_FETCH_TIMEOUT_S, self._expire) + self._timer.daemon = True + self._timer.start() + + def connect( + self, + address: tuple[str, int], + timeout: float | None, + source_address: tuple[str, int] | None = None, + ) -> socket.socket: + """``socket.create_connection``, with each attempt bounded by the deadline.""" + host, port = address + error: OSError = OSError(f"no address for {host}") + for family, kind, proto, _, target in socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM): + left = self._end - monotonic() + if left <= 0: + break + sock = socket.socket(family, kind, proto) + try: + sock.settimeout(left if timeout is None else min(left, timeout)) + if source_address is not None: + sock.bind(source_address) + sock.connect(target) + sock.settimeout(timeout) + except OSError as exc: + sock.close() + error = exc + continue + with self._lock: + if not self.expired: + try: + # A TLS wrap takes the file descriptor of sock, so keep a copy to shut down. + self._sockets.append(sock.dup()) + except OSError: + sock.close() + raise + return sock + sock.close() + break + if self.expired or self._end <= monotonic(): + raise TimeoutError("the prompt fetch passed its deadline") + raise error + + def _expire(self) -> None: + with self._lock: + self.expired = True + for sock in self._sockets: + with contextlib.suppress(OSError): + sock.shutdown(socket.SHUT_RDWR) + + def close(self) -> None: + self._timer.cancel() + with self._lock: + for sock in self._sockets: + sock.close() + self._sockets.clear() + + +class _Request(urllib.request.Request): + def __init__(self, url: str, headers: dict[str, str], deadline: _Deadline) -> None: + super().__init__(url, headers=headers, method="GET") + self.deadline = deadline + + +def _on_deadline(http_class: Any, request: urllib.request.Request) -> Any: + """``http_class``, with the sockets of a ``_Request`` under its deadline.""" + + def connection(host: str, **kwargs: Any) -> Any: + conn = http_class(host, **kwargs) + if isinstance(request, _Request): + conn._create_connection = request.deadline.connect + return conn + + return connection + + +class _HTTPHandler(urllib.request.HTTPHandler): + def do_open(self, http_class: Any, req: urllib.request.Request, **http_conn_args: Any) -> Any: + return super().do_open(_on_deadline(http_class, req), req, **http_conn_args) + + +class _HTTPSHandler(urllib.request.HTTPSHandler): + def do_open(self, http_class: Any, req: urllib.request.Request, **http_conn_args: Any) -> Any: + return super().do_open(_on_deadline(http_class, req), req, **http_conn_args) + + +_opener = urllib.request.build_opener(_NoRedirect, _HTTPHandler, _HTTPSHandler) def _fetch(url: str, headers: dict[str, str]) -> dict[str, Any]: - request = urllib.request.Request(url, headers=headers, method="GET") + deadline = _Deadline() + retry_after: str | None = None try: - with _opener.open(request, timeout=_FETCH_TIMEOUT_S) as response: - body: bytes = response.read() - except urllib.error.HTTPError as exc: - with exc: - error = _api_error(exc.code, exc.read()) - if exc.code in (429, 503): - retry_after = _retry_after(exc.headers.get("Retry-After")) - error._retry_after = retry_after # pyright: ignore[reportPrivateUsage] - raise error from None - except TimeoutError as exc: - raise PromptError( - "timeout", f"Fetching the prompt took more than {_FETCH_TIMEOUT_S:g} s" - ) from exc - except (urllib.error.URLError, OSError) as exc: - if isinstance(getattr(exc, "reason", None), TimeoutError): - raise PromptError( - "timeout", f"Fetching the prompt took more than {_FETCH_TIMEOUT_S:g} s" - ) from exc - raise PromptError("network_error", "Could not reach telemetry.dev") from exc + try: + with _opener.open( + _Request(url, headers, deadline), timeout=_FETCH_TIMEOUT_S + ) as response: + status: int = response.status + body: bytes = response.read() + except urllib.error.HTTPError as exc: + with exc: + status, body = exc.code, exc.read() + retry_after = exc.headers.get("Retry-After") + except Exception as exc: + if ( + deadline.expired + or isinstance(exc, TimeoutError) + or isinstance(getattr(exc, "reason", None), TimeoutError) + ): + raise _timeout() from exc + if isinstance(exc, (urllib.error.URLError, OSError)): + raise PromptError("network_error", "Could not reach telemetry.dev") from exc + raise + finally: + deadline.close() + # A read that the deadline cut short can end without an error, with part of the body. + if deadline.expired: + raise _timeout() + if not 200 <= status < 300: + error = _api_error(status, body) + if status in (429, 503): + error._retry_after = _retry_after(retry_after) # pyright: ignore[reportPrivateUsage] + raise error try: parsed = _parse_response(json.loads(body)) except ValueError: @@ -582,6 +694,10 @@ def _fetch(url: str, headers: dict[str, str]) -> dict[str, Any]: return parsed +def _timeout() -> PromptError: + return PromptError("timeout", f"Fetching the prompt took more than {_FETCH_TIMEOUT_S:g} s") + + def _api_error(status: int, body: bytes) -> PromptError: try: error = json.loads(body).get("error") diff --git a/sdks/python/tests/test_prompts.py b/sdks/python/tests/test_prompts.py index ec70c68..892d676 100644 --- a/sdks/python/tests/test_prompts.py +++ b/sdks/python/tests/test_prompts.py @@ -11,6 +11,7 @@ from collections.abc import Callable, Generator from datetime import datetime, timedelta, timezone from email.utils import format_datetime +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import SimpleNamespace from typing import Any, NoReturn from urllib.parse import parse_qsl, urlsplit @@ -446,6 +447,84 @@ def slow_lookup(*args: Any, **kwargs: Any) -> Any: assert time.monotonic() - started < 0.8 +def test_a_fetch_stops_reading_a_trickled_reply_at_its_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(prompts_module, "_FETCH_TIMEOUT_S", 0.2) + disconnected = threading.Event() + + # Each byte comes within the socket timeout, so only a deadline for the whole fetch ends it. + class Trickle(BaseHTTPRequestHandler): + def do_GET(self) -> None: + self.send_response(200) + self.send_header("content-length", "1000") + self.end_headers() + try: + for _ in range(60): + self.wfile.write(b" ") + self.wfile.flush() + time.sleep(0.05) + except OSError: + disconnected.set() + + def log_message(self, format: str, *args: object) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Trickle) + # server_close then waits for the reply thread, so that it does not write after the test. + server.daemon_threads = False + thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}) + thread.start() + try: + with pytest.raises(PromptError) as raised: + get_prompt( + "support/reply", + api_key="td_" + "live_a", + base_url=f"http://127.0.0.1:{server.server_address[1]}", + ) + assert raised.value.code == "timeout" + # The fetch closed the connection, rather than reading for the 3 s of the reply. + assert disconnected.wait(1) + finally: + server.shutdown() + server.server_close() + thread.join() + + +def test_a_fetch_starts_no_connection_after_its_deadline(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(prompts_module, "_FETCH_TIMEOUT_S", 0.2) + timeouts: list[float | None] = [] + + # Two addresses whose connect waits for the whole timeout, as an unreachable host does. + def lookup(*args: Any, **kwargs: Any) -> list[Any]: + address = (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.0.2.1", 80)) + return [address, address] + + class Unreachable: + def __init__(self, *args: Any) -> None: + self.timeout: float | None = None + + def settimeout(self, timeout: float | None) -> None: + self.timeout = timeout + + def connect(self, address: Any) -> NoReturn: + timeouts.append(self.timeout) + time.sleep(self.timeout or 0) + raise TimeoutError + + def close(self) -> None: + pass + + monkeypatch.setattr(socket, "getaddrinfo", lookup) + monkeypatch.setattr(socket, "socket", Unreachable) + with pytest.raises(PromptError) as raised: + get_prompt("support/reply", api_key="td_" + "live_a", base_url="http://prompts.test") + assert raised.value.code == "timeout" + time.sleep(0.3) + assert len(timeouts) == 1 + assert timeouts[0] is not None and timeouts[0] <= 0.2 + + def test_get_prompt_rejects_a_prompt_of_the_other_type_and_bad_options( server: CaptureServer, ) -> None: @@ -567,6 +646,7 @@ def test_with_prompt_stamps_model_call_spans_in_scope_and_no_others( start_span("plain").end() start_span("embed", type="embedding").end() start_span("explicit parent", type="generation", parent=outside_context).end() + start_span("list operation", attributes={"gen_ai.operation.name": ["chat"]}).end() outside.end() with start_span("after", type="generation"): @@ -583,7 +663,7 @@ def test_with_prompt_stamps_model_call_spans_in_scope_and_no_others( by_name[name]["td.prompt.variables"] == '{"customer_message":"Where is #4?","tone":"friendly"}' ) - for name in ("before", "outside", "tool", "plain", "embed", "after"): + for name in ("before", "outside", "tool", "plain", "embed", "list operation", "after"): assert "td.prompt.name" not in by_name[name]