From ff8866a9cc6e2b7c39c5c203231bdbca0e10a5ab Mon Sep 17 00:00:00 2001 From: Aikins Laryea Date: Fri, 30 Oct 2026 00:17:16 +0000 Subject: [PATCH] fix(sdk): bound prompt fetches and time them by elapsed time The TypeScript prompt cache and Retry-After wait use performance.now(), so a change of the system clock no longer shortens or extends them. A Python prompt fetch now ends at its 10 second deadline. Each connect attempt gets only the time left, none starts after the deadline, and the connection is shut down at the deadline, so a reply that trickles in a byte at a time cannot keep the fetch thread reading. prompt_attributes in Python ignores an operation attribute that is not a string; a list value raised TypeError and dropped the span. Both READMEs now say that a model call outside withPrompt/with_prompt still records the compiled messages as input when capture is on; it only lacks the td.prompt.* attributes. --- docs/sdk-conformance.md | 8 +- packages/sdk/README.md | 3 +- packages/sdk/src/prompts.ts | 12 +- packages/sdk/tests/prompts.test.ts | 23 ++- sdks/python/README.md | 5 +- sdks/python/src/telemetry_dev/_prompts.py | 170 ++++++++++++++++++---- sdks/python/tests/test_prompts.py | 82 ++++++++++- 7 files changed, 258 insertions(+), 45 deletions(-) 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]