From b6a25f9f455258dabb8d13b13978123239327376 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 10 Sep 2026 16:05:04 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(client):=20Agent=20Skills=20=E2=80=94?= =?UTF-8?q?=20the=20FDv2=20delivery=20transport?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FDv2SkillStore puts LaunchDarkly's SDK-facing FDv2 channel underneath the protocol layer: GET /sdk/poll and GET /sdk/stream, authenticated with the environment's server-side SDK key, streaming by default. It carries basis across requests, sends If-None-Match and treats 304 as a current answer, retries with capped jittered backoff, honours Retry-After only up to max_backoff, gives up after a bounded run of consecutive failures where a committed payload resets the count, and keeps serving last known good through every failure. A mobile key or client-side environment ID is refused in the constructor. Standard library only. close interrupts the socket rather than only setting a flag, because the delivery thread lives in a read no flag can reach; without that every shutdown of a healthy stream waited out the full join timeout. The no-store message now names FDv2SkillStore first, and watch_skills points at it as the store with a delivery transport. Co-Authored-By: Claude Fable 5.1 --- packages/client/README.md | 62 + packages/client/agents.md | 25 +- .../src/launchdarkly_ai_server/__init__.py | 5 +- .../src/launchdarkly_ai_server/skills_core.py | 13 +- .../src/launchdarkly_ai_server/skills_fdv2.py | 748 ++++++++- .../launchdarkly_ai_server/skills_watch.py | 2 +- packages/client/tests/test_skills.py | 12 + packages/client/tests/test_skills_fdv2.py | 1332 ++++++++++++++++- packages/client/tests/test_skills_watch.py | 5 +- 9 files changed, 2162 insertions(+), 42 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index 41d5c6b..42b1f40 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -445,6 +445,66 @@ which OS ran the write. The keys stay valid everywhere else: an AI Config refere named `aux` parses, and its other fields are unaffected. If you have a skill named for a device, rename it. +#### Receiving skills from LaunchDarkly + +`InMemorySkillStore` is for tests and bring-your-own-content. In production, skill content +arrives through `FDv2SkillStore`, which speaks LaunchDarkly's SDK-facing FDv2 delivery +channel — the same `GET /sdk/poll` and `GET /sdk/stream` endpoints the base SDK's FDv2 data +source uses, authenticated with the environment's server-side SDK key. + +```python +import os + +from launchdarkly_ai_server import FDv2SkillStore, init_client, watch_skills + +store = FDv2SkillStore(os.environ["LD_SDK_KEY"]).start() +store.wait_for_skills(timeout=10) +await init_client(options={"skillStore": store}) + +# Materialize now, and re-materialize whenever delivery changes. +report, watcher = await watch_skills("*", ".claude/skills") +try: + ... +finally: + watcher.close() + store.close() +``` + +**Nothing above the store changes.** The accessors, verification, and `write_skills` see raw +objects through the `SkillStore` interface and cannot tell which store produced them. + +**Server-side only.** Skills are for server-side agent runtimes and skill content is +customer-confidential. A mobile key (`mob-…`) or a client-side environment ID raises from the +constructor. + +**Streaming is the default, and it is what makes revocation fast.** A `delete-object` reaches +a live stream in seconds; with `mode="poll"` it arrives within one `poll_interval`. Paired +with `watch_skills`, a revoked skill's `SKILL.md` leaves the disk without a restart. During an +outage the store keeps serving the last content it received and `write_skills`' default +`on_unavailable="keep"` leaves managed files alone — an outage must not read as "everything +was revoked". + +**One network timeout, and its default depends on the mode.** `read_timeout` bounds every +socket operation of a request, connecting included. In `mode="poll"` it bounds the whole +request and defaults to 10 seconds; in `mode="stream"` it bounds each wait for the next bytes +and defaults to 300 seconds, well beyond LaunchDarkly's heartbeat interval. + +**The connection also carries your flags.** A client cannot request only the skill payload, +so a skills-enabled environment delivers flag and segment objects on the same connection. +They are skipped, not evaluated — this store does no evaluation of any kind — and +`diagnostics.objects_ignored` counts them. + +> **Beta caveats, worth knowing before you deploy.** Payload signing does not exist on this +> channel yet, so delivery is TLS-only and the content hash establishes self-consistency, not +> origin authenticity. The FDv2 protocol is opt-in per account: without it the endpoints +> return HTTP 403, which the store reports as a fatal error explaining what to do. `ld-relay` +> does not speak the FDv2 endpoints, so relay-only deployments cannot receive skills. + +**If every skill comes back empty, check `diagnostics.hashless_objects`.** Verification +withholds any delivered object without a `contentHash`, so a nonzero count means skills are +being withheld rather than that the environment has none. The store also logs an error per +hashless object naming the reason. There is deliberately no fallback that skips verification. + **Total path length is yours to bound, not the SDK's.** The 255-byte bound above is per *component*; the root is your path, so `` + `` + `/SKILL.md` can still exceed Windows' 260-character `MAX_PATH` with a perfectly legal key. Choose a short managed root on @@ -460,7 +520,9 @@ Windows. | `write_skills(skills, root, *, prune=True, timeout=10.0, on_unavailable="keep")` | Materialize skills under `root`, returning a `ReconcileReport`. `prune` removes formerly-managed skills no longer requested. `on_unavailable="raise"` raises instead of reporting when content cannot be retrieved. Raises `ValueError` for an unusable root, a negative `timeout`, or an unrecognised `on_unavailable`. **Performs synchronous filesystem I/O — see the note below.** | | `SkillStore` | The structural interface content arrives through: `get_object(kind, key, version=None)`, `all_objects(kind)`, optional `add_listener(kind, fn)` / `remove_listener(kind, fn)`. | | `InMemorySkillStore(objects=None)` | A dict-backed store with `put(raw)`, for local development and testing. Holds several versions of a key. | +| `FDv2SkillStore(sdk_key, *, base_uri=…, mode="stream", …)` | The delivery transport: a store fed by LaunchDarkly over the SDK-facing FDv2 channel. `start()`, `wait_for_skills(timeout)`, `close()`, `diagnostics`, `failed`; also a context manager. **Server-side only** — a mobile key or client-side environment ID raises. See *Receiving skills from LaunchDarkly* above. | | `watch_skills(skills, root, …)` | `write_skills` plus a re-reconcile on every delivery change. Returns `(initial report, SkillWatcher)`; close the watcher when done. Revocation then takes effect within `debounce` of arriving rather than at the next restart. | +| `StoreDiagnostics` | What the transport has seen: `payloads_transferred`, `skill_objects_received`, `objects_ignored`, `objects_revoked`, `hashless_objects`, `connection_failures`, `last_error`. | Configure the store with `init_client(options={"skillStore": store})`. With none configured, the accessors raise `RuntimeError` explaining what to do and `write_skills` reports the diff --git a/packages/client/agents.md b/packages/client/agents.md index 978dc8a..2a19c27 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -30,7 +30,7 @@ No other `launchdarkly-ai-*` package may define or duplicate these. They import | `src/launchdarkly_ai_server/types_validation.py` | `parse_ai_config` — validates flag variation shape; `is_valid_skill_key` / `is_valid_skill_version` / `skill_key_rejection_reason` (the canonical key-grammar explanation every layer quotes) | | `src/launchdarkly_ai_server/skills.py` | Agent Skills, retrieval half — `skill_refs`, `get_skill`/`get_skills`/`all_skills`, `InMemorySkillStore`, and the store/telemetry injection points `_set_store` / `_set_emitter_for_testing` | | `src/launchdarkly_ai_server/skills_core.py` | Shared skills internals — the `SkillStore` seam, module state, the telemetry seam and its three recorders, integrity verification, and store resolution. Imported by both `skills.py` and the materialization layer; imports neither | -| `src/launchdarkly_ai_server/skills_fdv2.py` | Agent Skills, delivery protocol — the wire-key/`version` translation, the held object set, and the pure `_ProtocolReader` that commits a payload's events at `payload-transferred`. Sits **below** the store interface; nothing in the feature imports it | +| `src/launchdarkly_ai_server/skills_fdv2.py` | Agent Skills, delivery transport — the FDv2 protocol, the wire-key/`version` translation, the held object set, and `FDv2SkillStore`. Sits **below** the store interface; imports `skills_core` only, and nothing imports it | | `src/launchdarkly_ai_server/skills_watch.py` | Agent Skills, eager re-reconcile — `watch_skills` / `SkillWatcher`, wiring the store's change listener to `write_skills`. Sits **above** `skills_fs` and modifies none of it | | `src/launchdarkly_ai_server/skills_fs.py` | Agent Skills, materialization half — `write_skills`, request resolution, the manifest format and on-disk filenames, per-skill reconcile, and pruning | | `src/launchdarkly_ai_server/safe_fs.py` | Descriptor-pinned filesystem primitives — `atomic_write`, `unlink_file`, `pinned_directory`, `open_directory_nofollow`, `open_or_create_directory`, `SymlinkRefused`, and the `*at()` capability probe. Owns the descriptor-vs-path platform split; knows nothing about skills | @@ -221,10 +221,12 @@ of skills, and the `"*"` reconcile, since `//SKILL.md` is a single pa ### The delivery transport, and the one field that will bite you -`skills_fdv2.py` translates LaunchDarkly's FDv2 delivery protocol into raw objects in the -shape `skills_core.SkillStore` documents. It lives below the store interface; **nothing above -that interface knows it exists**. If a transport change ever seems to require editing an -accessor, verification, or `write_skills`, the adapter boundary is wrong. +`FDv2SkillStore` speaks LaunchDarkly's SDK-facing FDv2 channel (`GET /sdk/poll`, +`GET /sdk/stream`, server-side SDK key in `Authorization`, `basis` + `mv` params, +`If-None-Match`/304). It lives below the store interface and produces raw objects in the +shape `skills_core.SkillStore` documents; **nothing above that interface knows it exists**. If a transport +change ever seems to require editing an accessor, verification, or `write_skills`, the adapter +boundary is wrong. **The skill's version is in the object's `key`. `version` is the payload's.** Each version of a skill is its own object on the wire, identified as `:`: @@ -285,6 +287,19 @@ such skill" — and would let a prune delete the last known-good copy on disk. N a hash from the delivered content: that certifies the content against itself and verifies nothing. +**There is one network timeout, not two.** `urllib`'s `timeout` is the socket timeout for the +whole operation, so connect, headers and each read share it, and the module cannot bound the +connect separately without a custom connection class it should not carry. `read_timeout` is +therefore the only knob, and its default is per mode (`DEFAULT_POLL_TIMEOUT` for a whole poll +request, `DEFAULT_STREAM_READ_TIMEOUT` for the gap between reads on a stream). Do not add a +parameter that the standard library cannot honour; `TestTimeouts` measures the bound against a +socket that accepts and never answers. + +**`close` interrupts the socket, it does not just set a flag.** The delivery thread spends its +life blocked in a read that no flag can reach, and closing a response from another thread does +not unblock CPython's buffered reader. `_interrupt_read` shuts the socket down underneath it. +Without that, every shutdown of a *healthy* stream blocks for the full join timeout. + ### The reported outcome vocabulary, and the `Resolution` mapping `get_skill` returns `Skill | None`; `get_skill_result` returns a frozen `SkillOutcome` diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index f863d84..0f98711 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -46,6 +46,7 @@ skill_refs, ) from .skills_core import SkillStore +from .skills_fdv2 import FDv2SkillStore, StoreDiagnostics from .skills_fs import ( MANIFEST_FILENAME, MANIFEST_VERSION, @@ -238,7 +239,9 @@ "write_skills", "SkillStore", "InMemorySkillStore", - # skills — the eager re-reconcile + # skills — the FDv2 delivery transport, and the eager re-reconcile it enables + "FDv2SkillStore", + "StoreDiagnostics", "watch_skills", "SkillWatcher", # skills — the three closed-set unions a typed consumer needs to name diff --git a/packages/client/src/launchdarkly_ai_server/skills_core.py b/packages/client/src/launchdarkly_ai_server/skills_core.py index d2d870c..b804301 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_core.py +++ b/packages/client/src/launchdarkly_ai_server/skills_core.py @@ -134,9 +134,18 @@ NO_STORE_MESSAGE = ( "No skill store is configured, so skill content cannot be retrieved. Configure " - 'one with init_client(options={"skillStore": store}) — InMemorySkillStore is ' - "available for local development and testing." + 'one with init_client(options={"skillStore": store}) — FDv2SkillStore receives ' + "content from LaunchDarkly, and InMemorySkillStore is available for local " + "development and testing." ) +""" +The first thing a user sees when no store is configured, so it names both stores. + +``FDv2SkillStore`` comes first because it is the answer in production, and a +message that offered only ``InMemorySkillStore`` would point a deployment at the +development store. Callers match on "skill store"; keep that phrase if the +wording changes. +""" # --------------------------------------------------------------------------- diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index d20a0e2..da84a3b 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -1,15 +1,10 @@ """ -Agent Skills — the FDv2 delivery protocol. +Agent Skills — the FDv2 delivery transport. -The half of the delivery transport that has no I/O: identifying skill objects -on the wire, translating them into the raw object shape the ``SkillStore`` -interface defines, holding them by ``(key, version)``, and applying a -payload's events as one consistent commit. ``FDv2SkillStore``, the store that -puts a network connection underneath this, follows in a separate change. - -It sits *below* the ``SkillStore`` interface, and everything above — the -accessors, integrity verification, the ``Skill`` dataclass, materialization — -is unaware of it. +The store implementation that talks to LaunchDarkly. It sits *below* the +``SkillStore`` interface: it produces raw wire objects in the shape +``skills_core`` documents, and everything above — the accessors, integrity +verification, the ``Skill`` dataclass, materialization — is unaware of it. Layering:: @@ -20,12 +15,12 @@ GET /sdk/poll, GET /sdk/stream, authenticated with the environment's server-side SDK key -Dependencies run one way: this module imports nothing from the feature beyond -the version validator in ``types_validation``, and nothing in the feature -imports it. It uses only the standard library, so it adds no dependency +Dependencies run one way: this module imports ``skills_core`` for the +interface's kind constant and nothing else from the feature, and nothing in the +feature imports it. It uses only the standard library, so it adds no dependency to a package whose sole runtime dependency is ``opentelemetry-api``. -Three things this layer does *not* do, on purpose: +Three things this module does *not* do, on purpose: - **It does not verify content.** Verification lives at the accessor boundary in ``skills_core`` so that it applies to every store equally, including a @@ -36,24 +31,29 @@ - **It does not evaluate anything.** Flag and segment objects that share the connection are skipped and counted, nothing more. -One assumption it *does* make, and states: **the payload intent it reads is the -payload skills arrive on.** Delivery sends one payload per credential and the -protocol tells a client to read only the first payload intent, so today those are -the same payload. ``_ProtocolReader`` keeps the pair apart anyway, because the -cost of conflating them is an emptied skill set. - The design rationale — why the skill's version is read from the object's ``key`` and never from ``version``, why changes commit at -``payload-transferred`` — is in ``agents.md`` under *The delivery transport*. +``payload-transferred``, why there is one network timeout — is in +``agents.md`` under *The delivery transport*. """ from __future__ import annotations +import json import logging +import math +import random import re +import socket +import threading +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Callable from dataclasses import dataclass, field -from typing import Any +from typing import Any, Literal +from .skills_core import SKILL_OBJECT_KIND from .types_validation import is_valid_skill_version logger = logging.getLogger(__name__) @@ -86,6 +86,20 @@ has exactly one. """ +DEFAULT_BASE_URI = "https://sdk.launchdarkly.com" +"""Where the SDK-facing FDv2 endpoints live. Overridable for Federal and private +instances.""" + +POLL_PATH = "/sdk/poll" +STREAM_PATH = "/sdk/stream" + +DEFAULT_POLL_TIMEOUT = 10.0 +"""Default ``read_timeout`` in ``"poll"`` mode: the bound on one whole request.""" + +DEFAULT_STREAM_READ_TIMEOUT = 300.0 +"""Default ``read_timeout`` in ``"stream"`` mode: the longest gap tolerated +between two reads. LaunchDarkly's heartbeats arrive well inside this.""" + _EVENT_SERVER_INTENT = "server-intent" _EVENT_PUT_OBJECT = "put-object" _EVENT_DELETE_OBJECT = "delete-object" @@ -115,6 +129,8 @@ no ``id``. """ +Mode = Literal["stream", "poll"] + _MOBILE_KEY_PREFIX = "mob-" _SERVER_KEY_PREFIX = "sdk-" _CLIENT_SIDE_ID = re.compile(r"\A[0-9a-f]{20,}\Z") @@ -504,7 +520,6 @@ def __init__(self, committed: _SkillObjectSet) -> None: self.diagnostics = StoreDiagnostics() # Identities already reported by ``_warn_hashless``. Per reader, so a # recreated store reports again and two stores never quieten each other. - # No lock: ``handle`` runs only on its owner's single delivery thread. self._warned_hashless: set[tuple[str, Any]] = set() # The payload the current intent describes, and the payload skills have # actually arrived on. One payload per connection makes these the same @@ -609,8 +624,8 @@ def _delete_object(self, data: Any) -> _TransferOutcome: self.diagnostics.objects_revoked += 1 # A revocation identifies the payload as ours just as a put does. self._skills_in_payload += 1 - # A tombstone carries identity and no content, so a listener that reads - # content must check for ``content`` rather than assume it. + # A tombstone carries identity and no content; see + # ``FDv2SkillStore.add_listener`` for what listeners should expect. self._changes.append( {"key": tombstone.key, "version": tombstone.object_version} ) @@ -780,3 +795,686 @@ def _warn_if_nothing_can_verify(committed: _SkillObjectSet) -> None: len(held), _HASHLESS_ADVICE, ) + + +# --------------------------------------------------------------------------- +# HTTP +# --------------------------------------------------------------------------- + + +class _FatalTransportError(Exception): + """A failure retrying cannot fix: bad credential, forbidden, wrong URI.""" + + +class _RecoverableTransportError(Exception): + """A failure worth retrying. Carries a server-requested delay when given one.""" + + def __init__(self, message: str, retry_after: float | None = None) -> None: + super().__init__(message) + self.retry_after = retry_after + + +_FORBIDDEN_ADVICE = ( + "The FDv2 protocol is opt-in per LaunchDarkly account and is served as HTTP " + "403 while it is off. Skill delivery needs it enabled; contact LaunchDarkly " + "support to enable it for your account." +) + + +def _retry_after_seconds(headers: Any) -> float | None: + """ + ``Retry-After`` in seconds, when the server sent a usable one. + + The HTTP-date form, and non-finite values such as ``inf`` or ``1e309`` that + ``float`` accepts, fall back to our own backoff: none of them is a delay, + and an infinite one would overflow the wait that honours it. + """ + if headers is None: + return None + try: + raw = headers.get("Retry-After") + except AttributeError: + return None + if raw is None: + return None + try: + seconds: float = float(str(raw).strip()) + except ValueError: + return None + if not math.isfinite(seconds): + return None + return max(0.0, seconds) + + +def _classify_status(status: int, headers: Any) -> Exception: + """Turns an HTTP error status into the right exception type.""" + if status == 401: + return _FatalTransportError( + "LaunchDarkly rejected the SDK key (HTTP 401). Skill delivery cannot " + "start. Check that the key is the environment's server-side SDK key." + ) + if status == 403: + return _FatalTransportError( + f"LaunchDarkly returned HTTP 403. {_FORBIDDEN_ADVICE}" + ) + if status in (400, 405, 406, 414, 501): + return _FatalTransportError( + f"LaunchDarkly returned HTTP {status}, which retrying will not fix. " + "The request this adapter sent was not understood. It carries only " + "the SDK key and, after the first payload, a 'basis' selector, so " + "check the base URI and that the endpoint speaks FDv2." + ) + return _RecoverableTransportError( + f"LaunchDarkly returned HTTP {status}", _retry_after_seconds(headers) + ) + + +def _interrupt_read(response: Any) -> None: + """ + Best-effort interruption of a read blocked on *response*, from another thread. + + Closing the response is not enough: CPython's buffered reader stays parked in + ``readline`` until bytes arrive. Shutting the *socket* down underneath it + unblocks it immediately. Reaching the socket means walking urllib's private + attribute chain, so every step is guarded and failure is silent: the + delivery thread is a daemon and ``close``'s join timeout is the backstop. + """ + for path in (("fp", "raw", "_sock"), ("fp", "_sock"), ("_sock",)): + found: Any = response + for name in path: + found = getattr(found, name, None) + if found is None: + break + if found is not None and hasattr(found, "shutdown"): + try: + found.shutdown(socket.SHUT_RDWR) + except OSError: + pass + return + + +class _StreamConnection: + """ + One open streaming connection: an event iterator plus a way to interrupt it + from another thread, which is what ``FDv2SkillStore.close`` needs. + """ + + def __init__(self, response: Any) -> None: + self._response = response + self.events = _iter_sse(response) + + def close(self) -> None: + """Interrupts the read. Safe to call from any thread, and twice.""" + _interrupt_read(self._response) + try: + self._response.close() + except Exception: + pass + + +@dataclass(frozen=True) +class _PollResult: + not_modified: bool + events: list[tuple[str, Any]] + etag: str | None + + +class _Requester: + """ + The only place this module opens a socket. Standard library only, on purpose. + + *read_timeout* is applied to every socket operation of a request. ``urllib`` + has no separate connect timeout: its ``timeout`` becomes the socket timeout + for the whole operation, so connecting, waiting for headers and each body + read are all bounded by the same value. + """ + + def __init__( + self, + sdk_key: str, + base_uri: str, + *, + read_timeout: float, + opener: Any = None, + ) -> None: + self._sdk_key = sdk_key + self._base_uri = base_uri.rstrip("/") + self._read_timeout = read_timeout + # Injectable so tests can drive a fake endpoint without a socket. + self._opener = opener or urllib.request.build_opener() + + def _url(self, path: str, basis: str | None) -> str: + """ + The request URL: the path, plus ``basis`` once a payload has committed. + + Deliberately no ``mv`` (data model version). That parameter selects the + *flag* data model and the connection rejects any value but the flag + default; the agent-skill payload is generic, is served regardless of it, + and has no model version of its own to ask for. + """ + if not basis: + return f"{self._base_uri}{path}" + return f"{self._base_uri}{path}?{urllib.parse.urlencode({'basis': basis})}" + + def _request( + self, path: str, basis: str | None, headers: dict[str, str] + ) -> urllib.request.Request: + all_headers = {"Authorization": self._sdk_key, **headers} + return urllib.request.Request( + self._url(path, basis), headers=all_headers, method="GET" + ) + + def poll(self, basis: str | None, etag: str | None) -> _PollResult: + """One ``GET /sdk/poll``. A 304 is a first-class outcome, not an error.""" + headers = {"Accept": "application/json"} + if etag: + headers["If-None-Match"] = etag + request = self._request(POLL_PATH, basis, headers) + try: + with self._opener.open(request, timeout=self._read_timeout) as response: + status = getattr(response, "status", None) or response.getcode() + if status == 304: + return _PollResult(not_modified=True, events=[], etag=etag) + body = response.read() + new_etag = response.headers.get("ETag") or etag + except urllib.error.HTTPError as exc: + if exc.code == 304: + # urllib raises on 304 when no redirect handler swallows it. + return _PollResult(not_modified=True, events=[], etag=etag) + raise _classify_status(exc.code, exc.headers) from exc + except Exception as exc: + raise _RecoverableTransportError( + f"polling request failed: {type(exc).__name__}: {exc}" + ) from exc + + return _PollResult( + not_modified=False, events=_decode_poll_body(body), etag=new_etag + ) + + def stream(self, basis: str | None) -> _StreamConnection: + """Opens ``GET /sdk/stream``.""" + request = self._request( + STREAM_PATH, + basis, + {"Accept": "text/event-stream", "Cache-Control": "no-cache"}, + ) + try: + response = self._opener.open(request, timeout=self._read_timeout) + except urllib.error.HTTPError as exc: + raise _classify_status(exc.code, exc.headers) from exc + except Exception as exc: + raise _RecoverableTransportError( + f"streaming request failed: {type(exc).__name__}: {exc}" + ) from exc + return _StreamConnection(response) + + +def _decode_poll_body(body: bytes) -> list[tuple[str, Any]]: + """ + Unwraps ``{"events": [...]}``. Polling and streaming carry identical event + objects, which is why the protocol reader is shared between the two modes. + """ + try: + parsed = json.loads(body.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise _RecoverableTransportError( + f"polling response was not valid JSON: {exc}" + ) from exc + if not isinstance(parsed, dict) or not isinstance(parsed.get("events"), list): + raise _RecoverableTransportError("polling response had no 'events' array") + events: list[tuple[str, Any]] = [] + for entry in parsed["events"]: + if not isinstance(entry, dict): + continue + name = entry.get("event") + if isinstance(name, str): + events.append((name, entry.get("data"))) + return events + + +def _iter_sse(response: Any) -> Any: + """ + Decodes an SSE body into ``(event name, data)`` pairs. + + Minimal on purpose: ``event:``/``data:`` fields, multi-line ``data`` joined + with newlines, a blank line dispatching, and ``:`` comments skipped. + """ + try: + name: str | None = None + data_lines: list[str] = [] + for raw_line in response: + line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n") + if line == "": + if name is not None: + payload = "\n".join(data_lines) + try: + parsed = json.loads(payload) if payload else None + except json.JSONDecodeError: + logger.warning( + "Discarding FDv2 '%s' event whose data was not JSON", name + ) + parsed = None + else: + yield name, parsed + name = None + data_lines = [] + continue + if line.startswith(":"): + continue + field_name, _, value = line.partition(":") + value = value[1:] if value.startswith(" ") else value + if field_name == "event": + name = value + elif field_name == "data": + data_lines.append(value) + finally: + try: + response.close() + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Backoff +# --------------------------------------------------------------------------- + + +def _backoff_delay( + attempt: int, *, base: float, maximum: float, jitter: float = 0.5 +) -> float: + """ + Exponential backoff with jitter, capped at *maximum*. + + Jitter is subtractive over the whole range rather than added on top, so the + cap is a real ceiling: a fleet restarted together must not reconnect in + lockstep, and must not exceed the interval the cap promises. + """ + # float(2 ** n): the integer power is untyped to mypy. + ceiling: float = min(maximum, base * float(2 ** max(0, attempt - 1))) + return ceiling * (1.0 - jitter * random.random()) + + +# --------------------------------------------------------------------------- +# The store +# --------------------------------------------------------------------------- + + +class FDv2SkillStore: + """ + A ``SkillStore`` fed by LaunchDarkly's SDK-facing FDv2 delivery channel. + + Constructed with the environment's server-side SDK key, started explicitly, + and passed to ``init_client``:: + + store = FDv2SkillStore(sdk_key=os.environ["LD_SDK_KEY"]) + store.start() + store.wait_for_skills(timeout=10) + await init_client(options={"skillStore": store}) + + skill = await get_skill("pdf-extraction") + ... + store.close() + + It also works as a context manager. + + **Server-side only.** A mobile key or a client-side environment ID is + refused in the constructor. + + **Delivery is in the background; retrieval is not.** A daemon thread owns + the connection and fills memory, and ``get_object`` only ever reads what has + already arrived. A process that calls ``get_skill`` immediately after + ``start()`` may see an empty store; ``wait_for_skills`` orders boot against + the first payload. + + **Last known good survives an outage.** A transport failure never empties + the store and never makes ``get_object`` raise, which is what makes + ``write_skills(on_unavailable="keep")`` correct. ``diagnostics`` and + ``failed`` report the degradation. + + **What arrives is untrusted.** Raw wire objects are held verbatim and + verified at the accessor boundary, not here. In particular an object with no + ``contentHash`` is held and then *withheld*; see + ``StoreDiagnostics.hashless_objects``. + """ + + def __init__( + self, + sdk_key: str, + *, + base_uri: str = DEFAULT_BASE_URI, + mode: Mode = "stream", + poll_interval: float = 30.0, + read_timeout: float | None = None, + initial_backoff: float = 1.0, + max_backoff: float = 30.0, + max_consecutive_failures: int = 10, + _requester: Any = None, + ) -> None: + """ + *mode* is ``"stream"`` by default. Prefer it: a ``delete-object`` reaches + a live stream in seconds. ``"poll"`` exists for environments that cannot + hold a long-lived connection, and revocation there is one + ``poll_interval`` late. + + *read_timeout* is the only network timeout and bounds every socket + operation of a request, so its meaning and default follow the mode: in + ``"poll"`` it bounds the whole request (``DEFAULT_POLL_TIMEOUT``); in + ``"stream"`` it bounds each wait for the next bytes + (``DEFAULT_STREAM_READ_TIMEOUT``). Must be positive when given. + + *max_backoff* caps every delay between retries, including one the server + asks for with ``Retry-After``. + + *max_consecutive_failures* bounds the retry loop. On exceeding it the + transport stops, logs an error, and the store keeps serving last known + good; ``failed`` reports it. Only failures in a row count: a committed + payload resets the count. + """ + _require_server_side_credential(sdk_key) + if mode not in ("stream", "poll"): + raise ValueError(f'mode must be "stream" or "poll", got {mode!r}') + if poll_interval <= 0: + raise ValueError(f"poll_interval must be positive, got {poll_interval!r}") + if read_timeout is None: + read_timeout = ( + DEFAULT_STREAM_READ_TIMEOUT + if mode == "stream" + else DEFAULT_POLL_TIMEOUT + ) + elif not (math.isfinite(read_timeout) and read_timeout > 0): + raise ValueError(f"read_timeout must be positive, got {read_timeout!r}") + + self._mode: Mode = mode + self._poll_interval = poll_interval + self._initial_backoff = initial_backoff + self._max_backoff = max_backoff + self._max_consecutive_failures = max_consecutive_failures + + self._objects = _SkillObjectSet() + self._reader = _ProtocolReader(self._objects) + self._lock = threading.RLock() + self._listeners: dict[str, list[Callable[[dict[str, Any]], Any]]] = {} + + self._basis: str | None = None + self._etag: str | None = None + + self._requester = _requester or _Requester( + sdk_key.strip(), + base_uri, + read_timeout=read_timeout, + ) + + self._stop = threading.Event() + self._first_payload = threading.Event() + self._thread: threading.Thread | None = None + self._failed_reason: str | None = None + # The open streaming connection, so ``close`` can interrupt its read. + self._connection: Any = None + # Recoverable failures since the last committed payload. Reset at the + # commit rather than when a connection returns: a stream only ever ends + # by being dropped, so resetting on return would count every healthy, + # server-recycled connection as a failure. + self._failures = 0 + + # -- lifecycle --------------------------------------------------------- + + def start(self) -> FDv2SkillStore: + """ + Starts the delivery thread. Idempotent; returns ``self`` so it chains. + + Does not block: use ``wait_for_skills`` when boot ordering matters. + """ + with self._lock: + if self._thread is not None and self._thread.is_alive(): + return self + self._stop.clear() + self._thread = threading.Thread( + target=self._run, name="ld-ai-skills-fdv2", daemon=True + ) + self._thread.start() + return self + + def close(self, timeout: float = 5.0) -> None: + """ + Stops delivery. Idempotent, and safe to call from any thread. + + Held content is *not* dropped: a closed store still answers from what it + received. Detaching the store from the accessors is the job of the + package-level ``launchdarkly_ai_server.shutdown()`` coroutine. + """ + self._stop.set() + # The delivery thread is normally blocked in a socket read that no flag + # can reach; without this the join waits out its full timeout. + with self._lock: + connection = self._connection + if connection is not None: + connection.close() + thread = self._thread + if ( + thread is not None + and thread.is_alive() + and thread is not threading.current_thread() + ): + thread.join(timeout=timeout) + + def __enter__(self) -> FDv2SkillStore: + return self.start() + + def __exit__(self, *_exc: Any) -> None: + self.close() + + def wait_for_skills(self, timeout: float = 10.0) -> bool: + """ + Blocks until the first payload has been committed, or *timeout* elapses. + + ``True`` means a payload arrived — not that any skill in it verified, and + not that the environment has any skills. ``diagnostics`` answers the rest. + """ + return self._first_payload.wait(timeout=timeout) + + @property + def failed(self) -> str | None: + """Why delivery stopped for good, or ``None`` while it is running.""" + with self._lock: + return self._failed_reason + + @property + def diagnostics(self) -> StoreDiagnostics: + """A snapshot of what the transport has seen. See ``StoreDiagnostics``.""" + with self._lock: + return StoreDiagnostics(**vars(self._reader.diagnostics)) + + # -- the SkillStore interface ----------------------------------------- + + def get_object( + self, kind: str, key: str, version: int | None = None + ) -> dict[str, Any] | None: + if kind != SKILL_OBJECT_KIND: + return None + with self._lock: + return self._objects.get(key, version) + + def all_objects(self, kind: str) -> dict[str, dict[str, Any]]: + if kind != SKILL_OBJECT_KIND: + return {} + with self._lock: + return self._objects.snapshot() + + def add_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: + """ + Registers *fn* to be called once per changed object, at + ``payload-transferred`` rather than as objects stream in. + + A put notifies with the raw skill object. A revocation notifies with a + ``{"key", "version"}`` tombstone carrying no content, so a listener that + reads content must check for ``content`` rather than assume it. + + *fn* runs on the delivery thread. Keep it cheap and non-blocking. An + exception it raises is logged and swallowed, because a broken listener + must not be able to kill delivery. + """ + with self._lock: + self._listeners.setdefault(kind, []).append(fn) + + def remove_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: + """ + Unregisters *fn* from *kind*. Safe to call from any thread, including + from inside a listener: a removal during one commit takes effect from + the next. + + Removes one occurrence; removing a callable that is not registered is a + no-op, so ``SkillWatcher.close`` can detach unconditionally. + """ + with self._lock: + listeners = self._listeners.get(kind) + if listeners is None: + return + try: + listeners.remove(fn) + except ValueError: + return + + def _notify(self, changes: list[dict[str, Any]]) -> None: + with self._lock: + listeners = list(self._listeners.get(SKILL_OBJECT_KIND, [])) + for raw in changes: + for listener in listeners: + try: + listener(raw) + except Exception: + logger.error( + "A skill store change listener raised; delivery continues", + exc_info=True, + ) + + # -- the delivery loop ------------------------------------------------- + + def _run(self) -> None: + while not self._stop.is_set(): + try: + if self._mode == "stream": + self._stream_once() + else: + self._poll_once() + # A poll that returned is a current answer even when it committed + # nothing (HTTP 304). A stream never returns normally; its + # successes are counted at each commit in ``_apply``. + self._record_success() + except _FatalTransportError as exc: + self._give_up(str(exc)) + return + except _RecoverableTransportError as exc: + with self._lock: + self._failures += 1 + failures = self._failures + self._reader.diagnostics.connection_failures = failures + self._reader.diagnostics.last_error = str(exc) + if failures > self._max_consecutive_failures: + self._give_up( + f"gave up after {failures} consecutive failures; " + f"last error: {exc}" + ) + return + delay = exc.retry_after + if delay is None or not math.isfinite(delay): + delay = _backoff_delay( + failures, base=self._initial_backoff, maximum=self._max_backoff + ) + # ``Retry-After`` is a request and ``max_backoff`` is a promise. + # The header may come from a proxy rather than LaunchDarkly, and + # a value in the hours would park revocation for that long. + delay = min(delay, self._max_backoff) + logger.warning( + "Skill delivery failed (%s); retrying in %.1fs", exc, delay + ) + if self._stop.wait(delay): + return + continue + except Exception as exc: # pragma: no cover - defensive + self._give_up(f"unexpected error in skill delivery: {exc!r}") + logger.error("Unexpected error in skill delivery", exc_info=True) + return + + if self._mode == "poll" and self._stop.wait(self._poll_interval): + return + + def _record_success(self) -> None: + with self._lock: + self._failures = 0 + self._reader.diagnostics.connection_failures = 0 + + def _give_up(self, reason: str) -> None: + with self._lock: + self._failed_reason = reason + self._reader.diagnostics.last_error = reason + logger.error( + "Skill delivery has stopped and will not retry: %s. The store keeps " + "serving the last content it received; skills will not update until " + "the process restarts with a working connection.", + reason, + ) + # Unblock anyone waiting on a first payload that is never coming. + self._first_payload.set() + + def _apply(self, name: str, data: Any) -> None: + """ + Feeds one event to the reader, publishes a commit, and raises the + transport error the event calls for, if any. + """ + with self._lock: + outcome = self._reader.handle(name, data) + if outcome.committed and outcome.basis is not None: + self._basis = outcome.basis + if outcome.committed: + # A commit breaks the row of consecutive failures. + self._record_success() + self._first_payload.set() + if outcome.changes: + self._notify(outcome.changes) + if outcome.fatal: + raise _FatalTransportError(outcome.fatal) + if outcome.disconnect: + raise _RecoverableTransportError(outcome.disconnect) + + def _poll_once(self) -> None: + with self._lock: + basis, etag = self._basis, self._etag + result = self._requester.poll(basis, etag) + with self._lock: + self._etag = result.etag + if result.not_modified: + logger.debug("Skill payload unchanged (HTTP 304)") + # A 304 counts as a first payload, so a boot that reconnects with a + # cached basis is not blocked on a transfer the server will not send. + self._first_payload.set() + return + for name, data in result.events: + self._apply(name, data) + + def _stream_once(self) -> None: + with self._lock: + basis = self._basis + connection = self._requester.stream(basis) + with self._lock: + self._connection = connection + try: + # ``close`` may have run while the connect was in flight and found + # no connection to interrupt; this is the last chance to notice + # before the read below blocks. + if self._stop.is_set(): + return + for name, data in connection.events: + if self._stop.is_set(): + return + self._apply(name, data) + except Exception: + if self._stop.is_set(): + # ``close`` interrupted the read on purpose. + return + raise + finally: + connection.close() + with self._lock: + self._connection = None + # A stream that ends without a goodbye is a dropped connection. + raise _RecoverableTransportError("the FDv2 stream closed unexpectedly") diff --git a/packages/client/src/launchdarkly_ai_server/skills_watch.py b/packages/client/src/launchdarkly_ai_server/skills_watch.py index 0526305..e421df6 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_watch.py +++ b/packages/client/src/launchdarkly_ai_server/skills_watch.py @@ -267,7 +267,7 @@ async def watch_skills( "watch_skills needs a skill store that implements add_listener(kind, " "fn); the configured store does not, so delivery changes cannot be " "observed. Use write_skills for a one-shot reconcile, or configure a " - "store with a delivery transport." + "store with a delivery transport (FDv2SkillStore)." ) if debounce < 0: raise ValueError(f"debounce must not be negative, got {debounce!r}") diff --git a/packages/client/tests/test_skills.py b/packages/client/tests/test_skills.py index d3f8309..a07ea6a 100644 --- a/packages/client/tests/test_skills.py +++ b/packages/client/tests/test_skills.py @@ -603,6 +603,18 @@ async def test_all_skills_raises_actionably_when_no_store(self) -> None: with pytest.raises(RuntimeError, match="skill store"): await all_skills() + async def test_the_no_store_message_names_the_delivery_store_first(self) -> None: + """ + A deployment that hits this message must be pointed at the store that + receives content from LaunchDarkly, not only at the development one. + """ + with pytest.raises(RuntimeError) as reported: + await get_skill("a") + message = str(reported.value) + assert "FDv2SkillStore" in message + assert "InMemorySkillStore" in message + assert message.index("FDv2SkillStore") < message.index("InMemorySkillStore") + async def test_shutdown_clears_the_store( self, make_raw_skill: Any, mock_ld_client: Any ) -> None: diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index b88d14d..0cce2f0 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -1,26 +1,55 @@ """ -Tests for the FDv2 skill delivery protocol. - -Wire semantics — which objects are skills, the skill's version in the wire ``key`` -versus the payload's in ``version``, revocation, mixed payloads, the commit at ``payload-transferred`` — are asserted -against ``_ProtocolReader``, which has no I/O, so each case reads as the contract -it is rather than as a server script. +Tests for the FDv2 skill delivery transport. + +Two layers, deliberately: + +- **A real fake endpoint.** ``_FakeFDv2Endpoint`` is an in-process + ``ThreadingHTTPServer`` that implements the wire contract — the ``basis`` + query parameter, ``Authorization``, ``If-None-Match``/304, the + ``{"events": [...]}`` polling envelope, and SSE for streaming. The store under + test opens real sockets against it, so request construction and header + handling are exercised rather than mocked. +- **The protocol reader driven directly.** Wire semantics — which objects are + skills, the skill's version in the wire ``key`` versus the payload's in + ``version``, revocation, mixed payloads — are + asserted against ``_ProtocolReader``, which has no I/O, so those cases read as + the contract they are instead of as a server script. """ from __future__ import annotations import hashlib +import json +import socket +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, ClassVar +from urllib.parse import parse_qs, urlparse import pytest -from launchdarkly_ai_server import InMemorySkillStore +from launchdarkly_ai_server import ( + FDv2SkillStore, + InMemorySkillStore, + all_skills, + get_skill, + get_skill_result, + init_client, + watch_skills, +) from launchdarkly_ai_server.skills_core import SKILL_OBJECT_KIND from launchdarkly_ai_server.skills_fdv2 import ( + DEFAULT_POLL_TIMEOUT, + DEFAULT_STREAM_READ_TIMEOUT, FDV2_KEY_DELIMITER, FDV2_OBJECT_KIND, + _backoff_delay, _is_skill_event, _ProtocolReader, + _RecoverableTransportError, + _Requester, + _retry_after_seconds, _SkillObjectSet, _store_object_from_put, _tombstone_from_delete, @@ -28,6 +57,7 @@ pytestmark = pytest.mark.usefixtures("reset_skill_state") +SDK_KEY = "sdk-00000000-0000-4000-8000-000000000000" SKILL_BODY = "---\nname: PDF Extraction\n---\nExtract text from PDFs.\n" @@ -143,6 +173,182 @@ def full_payload( ) +# --------------------------------------------------------------------------- +# The fake endpoint +# --------------------------------------------------------------------------- + + +class _FakeFDv2Endpoint: + """ + An in-process server implementing the SDK-facing FDv2 contract. + + Scripted per request: ``queue_poll`` appends a response for the next + ``/sdk/poll``, ``queue_stream`` appends a sequence of SSE events for the next + ``/sdk/stream``. Every request's method, path, query and headers are recorded + in ``requests`` so the tests can assert on what the store actually sent — + which is the only way ``basis`` round-tripping and ``If-None-Match`` can be + checked at all. + """ + + def __init__(self) -> None: + self.requests: list[dict[str, Any]] = [] + self._polls: list[dict[str, Any]] = [] + self._streams: list[list[dict[str, Any]]] = [] + self._lock = threading.Lock() + self.hold_stream_open = False + self._release = threading.Event() + + endpoint = self + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *_args: Any) -> None: + return + + def do_GET(self) -> None: + parsed = urlparse(self.path) + query = {k: v[0] for k, v in parse_qs(parsed.query).items()} + with endpoint._lock: + endpoint.requests.append( + { + "path": parsed.path, + "query": query, + "authorization": self.headers.get("Authorization"), + "if_none_match": self.headers.get("If-None-Match"), + "accept": self.headers.get("Accept"), + } + ) + if parsed.path == "/sdk/poll": + endpoint._serve_poll(self) + elif parsed.path == "/sdk/stream": + endpoint._serve_stream(self) + else: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + + class Server(ThreadingHTTPServer): + # Handler threads are not joined on shutdown: a test that ends while + # a stream is deliberately held open should not pay for the hold. + daemon_threads = True + + self._server = Server(("127.0.0.1", 0), Handler) + # A short poll interval so `shutdown` is prompt: the default 0.5s is + # paid at the teardown of every test that touches the endpoint. + self._thread = threading.Thread( + target=lambda: self._server.serve_forever(poll_interval=0.01), daemon=True + ) + self._thread.start() + + @property + def base_uri(self) -> str: + host, port = self._server.server_address[:2] + return f"http://{host}:{port}" + + # -- scripting --------------------------------------------------------- + + def queue_poll( + self, + payload_events: list[dict[str, Any]] | None = None, + *, + status: int = 200, + etag: str | None = None, + retry_after: str | None = None, + ) -> None: + with self._lock: + self._polls.append( + { + "status": status, + "events": payload_events or [], + "etag": etag, + "retry_after": retry_after, + } + ) + + def queue_stream(self, payload_events: list[dict[str, Any]]) -> None: + with self._lock: + self._streams.append(payload_events) + + # -- serving ----------------------------------------------------------- + + def _serve_poll(self, handler: BaseHTTPRequestHandler) -> None: + with self._lock: + response = ( + self._polls.pop(0) if self._polls else {"status": 304, "events": []} + ) + status = response["status"] + handler.send_response(status) + if response.get("etag"): + handler.send_header("ETag", response["etag"]) + if response.get("retry_after"): + handler.send_header("Retry-After", response["retry_after"]) + if status in (200,): + body = json.dumps({"events": response["events"]}).encode("utf-8") + handler.send_header("Content-Type", "application/json") + handler.send_header("Content-Length", str(len(body))) + handler.end_headers() + handler.wfile.write(body) + return + handler.send_header("Content-Length", "0") + handler.end_headers() + + def _serve_stream(self, handler: BaseHTTPRequestHandler) -> None: + with self._lock: + payload_events = self._streams.pop(0) if self._streams else [] + handler.send_response(200) + handler.send_header("Content-Type", "text/event-stream") + handler.send_header("Cache-Control", "no-cache") + handler.send_header("Transfer-Encoding", "chunked") + handler.end_headers() + for event in payload_events: + chunk = ( + f"event: {event['event']}\ndata: {json.dumps(event.get('data'))}\n\n" + ).encode() + handler.wfile.write(f"{len(chunk):X}\r\n".encode() + chunk + b"\r\n") + handler.wfile.flush() + if self.hold_stream_open: + # Keeps the connection up so a test can assert on the store's state + # without racing the reconnect path. Released on ``close`` so the + # hold costs the suite nothing once the test is done with it. + self._release.wait(timeout=10) + handler.wfile.write(b"0\r\n\r\n") + + def close(self) -> None: + self._release.set() + self._server.shutdown() + self._server.server_close() + + +@pytest.fixture +def endpoint() -> Any: + server = _FakeFDv2Endpoint() + yield server + server.close() + + +def poll_store(endpoint: Any, **kwargs: Any) -> FDv2SkillStore: + return FDv2SkillStore( + SDK_KEY, + base_uri=endpoint.base_uri, + mode="poll", + poll_interval=kwargs.pop("poll_interval", 0.05), + initial_backoff=kwargs.pop("initial_backoff", 0.01), + max_backoff=kwargs.pop("max_backoff", 0.05), + read_timeout=kwargs.pop("read_timeout", 5.0), + **kwargs, + ) + + +def wait_until(predicate: Any, timeout: float = 5.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return predicate() + + # --------------------------------------------------------------------------- # Identifying skill objects, and ignoring everything else # --------------------------------------------------------------------------- @@ -827,6 +1033,608 @@ def test_snapshot_agrees(self) -> None: assert memory.all_objects(SKILL_OBJECT_KIND) == objects.snapshot() +# --------------------------------------------------------------------------- +# The store against the fake endpoint +# --------------------------------------------------------------------------- + + +class TestPollingAgainstTheEndpoint: + def test_a_polled_skill_becomes_retrievable_through_the_accessors( + self, endpoint: Any + ) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=5) is True + raw = store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") + assert raw is not None + assert raw["version"] == 3 + + def test_the_request_carries_the_sdk_key_and_no_data_model_version( + self, endpoint: Any + ) -> None: + """ + No ``mv``: that parameter selects the *flag* data model, the connection + rejects any value but the flag default, and the generic agent-skill + payload is served regardless of it. Sending ``mv=1`` — the skill + payload's own model version — gets the whole connection refused. + """ + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + first = endpoint.requests[0] + assert first["path"] == "/sdk/poll" + assert first["authorization"] == SDK_KEY + assert "mv" not in first["query"] + + def test_the_first_request_sends_no_basis(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + assert "basis" not in endpoint.requests[0]["query"] + + def test_the_basis_from_payload_transferred_is_echoed_on_the_next_request( + self, endpoint: Any + ) -> None: + endpoint.queue_poll( + full_payload(("put-object", put_skill()), state="selector-abc") + ) + endpoint.queue_poll(status=304) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + assert wait_until(lambda: len(endpoint.requests) >= 2) + assert endpoint.requests[1]["query"]["basis"] == "selector-abc" + + def test_the_basis_advances_across_successive_payloads(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()), state="basis-1")) + endpoint.queue_poll( + events( + ("server-intent", server_intent("xfer-changes")), + ("put-object", put_skill("second")), + ("payload-transferred", transferred("basis-2")), + ) + ) + endpoint.queue_poll(status=304) + with poll_store(endpoint): + assert wait_until(lambda: len(endpoint.requests) >= 3) + bases = [r["query"].get("basis") for r in endpoint.requests[:3]] + assert bases == [None, "basis-1", "basis-2"] + + def test_an_etag_is_returned_as_if_none_match(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill())), etag='W/"v1"') + endpoint.queue_poll(status=304) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + assert wait_until(lambda: len(endpoint.requests) >= 2) + assert endpoint.requests[1]["if_none_match"] == 'W/"v1"' + + def test_a_304_keeps_the_held_content(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill())), etag='W/"v1"') + endpoint.queue_poll(status=304) + endpoint.queue_poll(status=304) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + assert wait_until(lambda: len(endpoint.requests) >= 3) + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + assert store.diagnostics.payloads_transferred == 1 + assert store.failed is None + + def test_a_304_before_any_payload_still_releases_wait_for_skills( + self, endpoint: Any + ) -> None: + """A reconnect with a cached basis has nothing to transfer; boot must not + block on a payload the server has no reason to send.""" + endpoint.queue_poll(status=304) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=5) is True + + def test_a_mixed_payload_over_the_wire_yields_only_the_skill( + self, endpoint: Any + ) -> None: + endpoint.queue_poll( + full_payload( + ("put-object", put_flag("flag-a")), + ("put-object", put_segment("beta")), + ("put-object", put_skill("pdf-extraction")), + ("put-object", put_flag("flag-b")), + ) + ) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + held = store.all_objects(SKILL_OBJECT_KIND) + assert len(held) == 1 + assert next(iter(held.values()))["key"] == "pdf-extraction" + assert store.diagnostics.objects_ignored == 3 + + def test_a_revocation_over_the_wire_removes_the_skill(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll( + events( + ("server-intent", server_intent("xfer-changes")), + ("delete-object", delete_skill()), + ("payload-transferred", transferred("basis-2")), + ) + ) + endpoint.queue_poll(status=304) + with poll_store(endpoint) as store: + assert wait_until( + lambda: ( + store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is None + and store.diagnostics.objects_revoked == 1 + ) + ) + + def test_the_store_asks_for_only_the_kind_it_serves(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + assert store.get_object("flag", "pdf-extraction") is None + assert store.all_objects("flag") == {} + + +class TestStreamingAgainstTheEndpoint: + def test_a_streamed_payload_lands(self, endpoint: Any) -> None: + endpoint.hold_stream_open = True + endpoint.queue_stream(full_payload(("put-object", put_skill()))) + store = FDv2SkillStore( + SDK_KEY, base_uri=endpoint.base_uri, mode="stream", initial_backoff=0.01 + ) + try: + store.start() + assert store.wait_for_skills(timeout=5) is True + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + finally: + store.close() + + def test_the_stream_request_advertises_event_stream(self, endpoint: Any) -> None: + endpoint.hold_stream_open = True + endpoint.queue_stream(full_payload(("put-object", put_skill()))) + store = FDv2SkillStore(SDK_KEY, base_uri=endpoint.base_uri, mode="stream") + try: + store.start() + store.wait_for_skills(timeout=5) + finally: + store.close() + assert endpoint.requests[0]["path"] == "/sdk/stream" + assert endpoint.requests[0]["accept"] == "text/event-stream" + + def test_a_streamed_revocation_arrives_without_a_restart( + self, endpoint: Any + ) -> None: + endpoint.hold_stream_open = True + endpoint.queue_stream( + full_payload(("put-object", put_skill())) + + events( + ("server-intent", server_intent("xfer-changes")), + ("delete-object", delete_skill()), + ("payload-transferred", transferred("basis-2")), + ) + ) + store = FDv2SkillStore(SDK_KEY, base_uri=endpoint.base_uri, mode="stream") + try: + store.start() + assert wait_until( + lambda: ( + store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is None + and store.diagnostics.objects_revoked == 1 + ) + ) + finally: + store.close() + + def test_a_dropped_stream_reconnects_with_the_basis_it_reached( + self, endpoint: Any + ) -> None: + endpoint.queue_stream( + full_payload(("put-object", put_skill()), state="basis-1") + ) + endpoint.queue_stream(events(("heart-beat", None))) + store = FDv2SkillStore( + SDK_KEY, + base_uri=endpoint.base_uri, + mode="stream", + initial_backoff=0.01, + max_backoff=0.05, + ) + try: + store.start() + assert wait_until(lambda: len(endpoint.requests) >= 2) + finally: + store.close() + assert endpoint.requests[1]["query"]["basis"] == "basis-1" + + def test_close_returns_promptly_while_a_stream_is_open(self, endpoint: Any) -> None: + """ + The delivery thread is blocked in a socket read that no stop flag can + reach, so ``close`` closes the connection under it. Without that, every + shutdown of a healthy stream waits out the join timeout. + """ + endpoint.hold_stream_open = True + endpoint.queue_stream(full_payload(("put-object", put_skill()))) + store = FDv2SkillStore(SDK_KEY, base_uri=endpoint.base_uri, mode="stream") + store.start() + assert store.wait_for_skills(timeout=5) is True + started = time.monotonic() + store.close(timeout=5.0) + assert time.monotonic() - started < 1.0 + + def test_an_interrupted_stream_is_not_reported_as_a_failure( + self, endpoint: Any + ) -> None: + endpoint.hold_stream_open = True + endpoint.queue_stream(full_payload(("put-object", put_skill()))) + store = FDv2SkillStore(SDK_KEY, base_uri=endpoint.base_uri, mode="stream") + store.start() + store.wait_for_skills(timeout=5) + store.close() + assert store.failed is None + + def test_content_survives_a_reconnect(self, endpoint: Any) -> None: + endpoint.queue_stream(full_payload(("put-object", put_skill()))) + endpoint.queue_stream(events(("heart-beat", None))) + store = FDv2SkillStore( + SDK_KEY, base_uri=endpoint.base_uri, mode="stream", initial_backoff=0.01 + ) + try: + store.start() + assert wait_until(lambda: len(endpoint.requests) >= 2) + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + finally: + store.close() + + +# --------------------------------------------------------------------------- +# Failure handling +# --------------------------------------------------------------------------- + + +class _ScriptedConnection: + """Stands in for ``_StreamConnection``: an event iterator plus a close.""" + + def __init__(self, payload_events: Any) -> None: + self.events = iter(payload_events) + self.closed = False + + def close(self) -> None: + self.closed = True + + +class _ScriptedRequester: + """Raises a scripted sequence, so backoff is asserted without real sockets.""" + + def __init__(self, *outcomes: Any) -> None: + self.outcomes = list(outcomes) + self.calls: list[tuple[str | None, str | None]] = [] + + def poll(self, basis: str | None, etag: str | None) -> Any: + self.calls.append((basis, etag)) + outcome = ( + self.outcomes.pop(0) if self.outcomes else _RecoverableTransportError("x") + ) + if isinstance(outcome, Exception): + raise outcome + return outcome + + def stream(self, basis: str | None) -> Any: + self.calls.append((basis, None)) + outcome = ( + self.outcomes.pop(0) if self.outcomes else _RecoverableTransportError("x") + ) + if isinstance(outcome, Exception): + raise outcome + return _ScriptedConnection(outcome) + + +class _RecyclingRequester: + """ + A healthy server that recycles connections: every ``stream`` call succeeds, + transfers a full payload, and then ends the connection, as LaunchDarkly and + any proxy in between do to a long-lived stream. + """ + + def __init__(self) -> None: + self.connections = 0 + + def stream(self, basis: str | None) -> Any: + self.connections += 1 + return _ScriptedConnection( + [ + (e["event"], e["data"]) + for e in full_payload( + ("put-object", put_skill()), state=f"basis-{self.connections}" + ) + ] + ) + + +class _BlockingConnection: + """A stream that never produces an event until it is closed.""" + + def __init__(self) -> None: + self._closed = threading.Event() + + @property + def events(self) -> Any: + self._closed.wait() + return iter(()) + + def close(self) -> None: + self._closed.set() + + +class _SlowConnectRequester: + """ + A ``stream`` whose connect does not return until the test releases it, + standing in for a slow TLS handshake, followed by a read that never yields. + """ + + def __init__(self) -> None: + self.entered = threading.Event() + self.release = threading.Event() + + def stream(self, basis: str | None) -> Any: + self.entered.set() + self.release.wait(timeout=10) + return _BlockingConnection() + + +def stream_store(**kwargs: Any) -> FDv2SkillStore: + return FDv2SkillStore( + SDK_KEY, + mode="stream", + initial_backoff=kwargs.pop("initial_backoff", 0.001), + max_backoff=kwargs.pop("max_backoff", 0.002), + **kwargs, + ) + + +class TestFailureHandling: + def test_a_403_stops_delivery_and_explains_why( + self, endpoint: Any, caplog: Any + ) -> None: + endpoint.queue_poll(status=403) + with caplog.at_level("ERROR"): + with poll_store(endpoint) as store: + assert wait_until(lambda: store.failed is not None) + assert "403" in store.failed + assert "opt-in" in store.failed + assert any("opt-in" in r.getMessage() for r in caplog.records) + + def test_a_401_stops_delivery(self, endpoint: Any) -> None: + endpoint.queue_poll(status=401) + with poll_store(endpoint) as store: + assert wait_until(lambda: store.failed is not None) + assert "401" in store.failed + + def test_a_fatal_failure_releases_wait_for_skills_rather_than_hanging( + self, endpoint: Any + ) -> None: + endpoint.queue_poll(status=401) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=5) is True + assert store.failed is not None + + def test_a_fatal_failure_keeps_last_known_good_servable( + self, endpoint: Any + ) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=403) + with poll_store(endpoint) as store: + assert wait_until(lambda: store.failed is not None) + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + + def test_a_500_is_retried(self, endpoint: Any) -> None: + endpoint.queue_poll(status=500) + endpoint.queue_poll(status=503) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=5) is True + assert store.failed is None + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + + def test_a_retry_resets_the_failure_count_on_success(self, endpoint: Any) -> None: + endpoint.queue_poll(status=500) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=304) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=5) + assert wait_until(lambda: store.diagnostics.connection_failures == 0) + + def test_retries_are_bounded(self) -> None: + store = FDv2SkillStore( + SDK_KEY, + mode="poll", + poll_interval=0.01, + initial_backoff=0.001, + max_backoff=0.002, + max_consecutive_failures=3, + _requester=_ScriptedRequester(), + ) + try: + store.start() + assert wait_until(lambda: store.failed is not None) + # Four, not three: the bound is the number of failures *tolerated*, + # so the run that exceeds it is the one that gives up. + assert "gave up after 4 consecutive failures" in store.failed + finally: + store.close() + + def test_recycled_stream_connections_are_not_failures(self) -> None: + # A streaming connection only ever ends by being dropped, so a loop + # that counted every drop as a failure would give up on a healthy + # server after max_consecutive_failures + 1 recycles, and delivery + # (including revocation) would silently stop for the process lifetime. + requester = _RecyclingRequester() + store = stream_store(max_consecutive_failures=3, _requester=requester) + try: + store.start() + assert wait_until(lambda: requester.connections >= 8) + assert store.failed is None + assert store.diagnostics.payloads_transferred >= 8 + # A drop is a failure until the next commit clears it, so the count + # may read 1 mid-reconnect. What it must never do is climb. + assert store.diagnostics.connection_failures <= 1 + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + finally: + store.close() + + def test_a_stream_commit_resets_the_failure_count(self) -> None: + payload = [ + (e["event"], e["data"]) for e in full_payload(("put-object", put_skill())) + ] + requester = _ScriptedRequester( + _RecoverableTransportError("x"), + _RecoverableTransportError("x"), + _RecoverableTransportError("x"), + payload, + ) + store = stream_store(max_consecutive_failures=3, _requester=requester) + try: + store.start() + assert store.wait_for_skills(timeout=5) + # Three failures reach the bound, then a commit, then the exhausted + # requester fails on every reconnect. The count must start again at + # the commit: the stream's own drop is failure one, and three more + # connects are owed before giving up. Carrying the three over would + # give up on the drop itself, with no further connect at all. + assert wait_until(lambda: store.failed is not None) + assert "gave up after 4 consecutive failures" in store.failed + assert "last error: x" in store.failed + assert len(requester.calls) == 7 + finally: + store.close() + + def test_stream_retries_are_bounded(self) -> None: + store = stream_store( + max_consecutive_failures=3, _requester=_ScriptedRequester() + ) + try: + store.start() + assert wait_until(lambda: store.failed is not None) + assert "gave up after 4 consecutive failures" in store.failed + finally: + store.close() + + def test_a_retry_after_header_is_honoured(self) -> None: + requester = _ScriptedRequester( + _RecoverableTransportError("slow down", retry_after=0.25), + ) + store = FDv2SkillStore( + SDK_KEY, + mode="poll", + poll_interval=10.0, + initial_backoff=5.0, + _requester=requester, + ) + try: + started = time.monotonic() + store.start() + assert wait_until(lambda: len(requester.calls) >= 2, timeout=3) + elapsed = time.monotonic() - started + # The server asked for 0.25s; our own backoff would have been 5s. + assert 0.2 <= elapsed < 3.0 + finally: + store.close() + + def test_a_retry_after_header_is_parsed_off_the_wire(self, endpoint: Any) -> None: + endpoint.queue_poll(status=429, retry_after="0") + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint, initial_backoff=5.0) as store: + # If Retry-After were ignored the 5s backoff would blow the timeout. + assert store.wait_for_skills(timeout=3) is True + + @pytest.mark.parametrize("raw", ["inf", "Infinity", "-inf", "nan", "1e309"]) + def test_a_non_finite_retry_after_is_ignored(self, raw: str) -> None: + assert _retry_after_seconds({"Retry-After": raw}) is None + + def test_retry_after_parsing_keeps_its_edges(self) -> None: + assert _retry_after_seconds({"Retry-After": "0"}) == 0.0 + assert _retry_after_seconds({"Retry-After": "-5"}) == 0.0 + assert ( + _retry_after_seconds({"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}) + is None + ) + assert _retry_after_seconds({"Retry-After": "2.5"}) == 2.5 + + @pytest.mark.parametrize("retry_after", [float("inf"), float("nan"), 86400.0]) + def test_an_unreasonable_retry_after_neither_kills_delivery_nor_parks_it( + self, retry_after: float + ) -> None: + # An infinite wait would overflow inside the retry handler and kill the + # thread with `failed` still None; a day-long one would be honoured to + # the second. Both must fall back to the max_backoff cap and carry on. + requester = _ScriptedRequester( + _RecoverableTransportError("slow down", retry_after=retry_after), + [ + (e["event"], e["data"]) + for e in full_payload(("put-object", put_skill())) + ], + ) + store = stream_store(max_backoff=0.05, _requester=requester) + try: + store.start() + assert store.wait_for_skills(timeout=3) is True + assert store.failed is None + assert store._thread is not None and store._thread.is_alive() + finally: + store.close() + + def test_a_non_finite_retry_after_off_the_wire_falls_back_to_backoff( + self, endpoint: Any + ) -> None: + endpoint.queue_poll(status=429, retry_after="inf") + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=3) is True + assert store.failed is None + + def test_backoff_is_exponential_and_capped(self) -> None: + assert _backoff_delay(1, base=1.0, maximum=30.0, jitter=0.0) == 1.0 + assert _backoff_delay(2, base=1.0, maximum=30.0, jitter=0.0) == 2.0 + assert _backoff_delay(3, base=1.0, maximum=30.0, jitter=0.0) == 4.0 + assert _backoff_delay(20, base=1.0, maximum=30.0, jitter=0.0) == 30.0 + + def test_jitter_never_exceeds_the_cap(self) -> None: + for attempt in range(1, 12): + for _ in range(50): + assert 0.0 <= _backoff_delay(attempt, base=1.0, maximum=5.0) <= 5.0 + + def test_a_malformed_polling_envelope_is_recoverable_not_fatal( + self, endpoint: Any + ) -> None: + store = FDv2SkillStore( + SDK_KEY, + mode="poll", + poll_interval=0.01, + initial_backoff=0.001, + _requester=_ScriptedRequester( + _RecoverableTransportError("polling response had no 'events' array") + ), + ) + try: + store.start() + assert wait_until(lambda: store.diagnostics.connection_failures >= 1) + assert store.failed is None + finally: + store.close() + + def test_a_listener_that_raises_does_not_kill_delivery(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill("first")))) + endpoint.queue_poll( + events( + ("server-intent", server_intent("xfer-changes")), + ("put-object", put_skill("second")), + ("payload-transferred", transferred("basis-2")), + ) + ) + endpoint.queue_poll(status=304) + with poll_store(endpoint) as store: + store.add_listener(SKILL_OBJECT_KIND, lambda _raw: 1 / 0) + assert wait_until( + lambda: store.get_object(SKILL_OBJECT_KIND, "second") is not None + ) + assert store.failed is None + + # --------------------------------------------------------------------------- # The contentHash gap # --------------------------------------------------------------------------- @@ -854,6 +1662,63 @@ class TestMissingContentHash: itself and verify nothing. """ + async def test_a_hashless_skill_is_withheld_with_the_right_reason( + self, endpoint: Any + ) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill(omit_hash=True)))) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + await init_client(options={"skillStore": store}, client=object()) + + outcome = await get_skill_result("pdf-extraction") + assert outcome.skill is None + assert outcome.reason == "integrity_failure" + assert await get_skill("pdf-extraction") is None + assert await all_skills() == [] + + async def test_the_object_is_still_held_so_the_outcome_is_not_absent( + self, endpoint: Any + ) -> None: + """ + Holding it is what makes the failure diagnosable. Dropping it at the + transport would report ``absent`` — indistinguishable from "no such + skill" — and would additionally let a prune delete the last known-good + copy already on disk. + """ + endpoint.queue_poll(full_payload(("put-object", put_skill(omit_hash=True)))) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + raw = store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") + assert raw is not None + assert "contentHash" not in raw + await init_client(options={"skillStore": store}, client=object()) + assert (await get_skill_result("pdf-extraction")).reason != "absent" + + def test_the_store_counts_hashless_objects(self, endpoint: Any) -> None: + endpoint.queue_poll( + full_payload( + ("put-object", put_skill("a", omit_hash=True)), + ("put-object", put_skill("b", omit_hash=True)), + ("put-object", put_skill("c")), + ) + ) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + assert store.diagnostics.hashless_objects == 2 + assert store.diagnostics.skill_objects_received == 3 + + def test_a_hashless_object_logs_an_error_naming_the_reason_code( + self, endpoint: Any, caplog: Any + ) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill(omit_hash=True)))) + with caplog.at_level("ERROR"): + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + rendered = "\n".join(r.getMessage() for r in caplog.records) + assert "missing_content_hash" in rendered + assert "pdf-extraction" in rendered + assert "contentHash" in rendered + def test_a_redelivered_hashless_object_logs_once_per_store( self, caplog: Any ) -> None: @@ -895,3 +1760,456 @@ def test_two_live_stores_do_not_suppress_each_other(self, caplog: Any) -> None: drive(one, payload) drive(two, payload) assert len(_per_object_hashless_errors(caplog)) == 2 + + def test_a_wholly_hashless_payload_says_so_once( + self, endpoint: Any, caplog: Any + ) -> None: + endpoint.queue_poll( + full_payload( + ("put-object", put_skill("a", omit_hash=True)), + ("put-object", put_skill("b", omit_hash=True)), + ) + ) + with caplog.at_level("ERROR"): + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + summaries = [ + r + for r in caplog.records + if "No skill content will resolve" in r.getMessage() + ] + assert len(summaries) == 1 + assert "All 2 skill object(s)" in summaries[0].getMessage() + + def test_a_partly_hashed_payload_does_not_claim_total_failure( + self, endpoint: Any, caplog: Any + ) -> None: + endpoint.queue_poll( + full_payload( + ("put-object", put_skill("a", omit_hash=True)), + ("put-object", put_skill("b")), + ) + ) + with caplog.at_level("ERROR"): + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + rendered = "\n".join(r.getMessage() for r in caplog.records) + assert "No skill content will resolve" not in rendered + + async def test_a_hash_that_does_not_match_is_a_different_failure( + self, endpoint: Any + ) -> None: + """``missing_content_hash`` and ``hash_mismatch`` must not collapse: one + means the envelope carried no hash, the other means the content did not + match the hash it carried.""" + endpoint.queue_poll( + full_payload( + ("put-object", put_skill(content_hash=_hash("something else"))) + ) + ) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + assert store.diagnostics.hashless_objects == 0 + await init_client(options={"skillStore": store}, client=object()) + assert ( + await get_skill_result("pdf-extraction") + ).reason == "integrity_failure" + + async def test_a_hashed_skill_resolves_end_to_end(self, endpoint: Any) -> None: + """The positive control: a well-formed envelope resolves end to end.""" + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + await init_client(options={"skillStore": store}, client=object()) + + skill = await get_skill("pdf-extraction") + assert skill is not None + assert skill.key == "pdf-extraction" + assert skill.version == 3 + assert skill.content == SKILL_BODY.encode("utf-8") + assert skill.content_hash == _hash(SKILL_BODY) + assert skill.name == "PDF Extraction" + + async def test_a_pinned_reference_resolves_to_the_pinned_object_version( + self, endpoint: Any + ) -> None: + endpoint.queue_poll( + full_payload( + ("put-object", put_skill(object_version=2, content="v2 body")), + ("put-object", put_skill(object_version=5, content="v5 body")), + ) + ) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + await init_client(options={"skillStore": store}, client=object()) + + pinned = await get_skill("pdf-extraction", version=2) + assert pinned is not None + assert pinned.content == b"v2 body" + newest = await get_skill("pdf-extraction") + assert newest is not None + assert newest.version == 5 + + async def test_the_payload_version_is_not_resolvable_as_a_skill_version( + self, endpoint: Any + ) -> None: + """ + The end-to-end form of the wire-key/``version`` assertion. + + Asking for the payload version resolves nothing — reported ``absent``, + because the store answers "I hold no such version" rather than answering + with the wrong one. The version that *does* resolve is the one after the + delimiter in the object's wire ``key``. + """ + endpoint.queue_poll( + full_payload( + ("put-object", put_skill(object_version=3, payload_version=42)) + ) + ) + with poll_store(endpoint) as store: + store.wait_for_skills(timeout=5) + await init_client(options={"skillStore": store}, client=object()) + by_payload_version = await get_skill_result("pdf-extraction", version=42) + assert by_payload_version.skill is None + assert by_payload_version.reason == "absent" + assert await get_skill("pdf-extraction", version=3) is not None + + +# --------------------------------------------------------------------------- +# Server-side only +# --------------------------------------------------------------------------- + + +class TestServerSideOnly: + def test_a_mobile_key_is_refused(self) -> None: + with pytest.raises(ValueError, match="mobile key"): + FDv2SkillStore("mob-00000000-0000-4000-8000-000000000000") + + def test_a_client_side_environment_id_is_refused(self) -> None: + with pytest.raises(ValueError, match="client-side"): + FDv2SkillStore("0123456789abcdef01234567") + + def test_an_empty_credential_is_refused(self) -> None: + with pytest.raises(ValueError, match="server-side SDK key"): + FDv2SkillStore(" ") + + def test_a_server_side_key_is_accepted(self) -> None: + assert FDv2SkillStore(SDK_KEY) is not None + + def test_an_unrecognised_credential_shape_warns_but_is_allowed( + self, caplog: Any + ) -> None: + """Private instances and test doubles issue keys without the public prefix.""" + with caplog.at_level("WARNING"): + FDv2SkillStore("my-private-instance-credential") + assert any("server-side SDK key" in r.message for r in caplog.records) + + def test_an_unknown_mode_is_refused(self) -> None: + with pytest.raises(ValueError, match="stream"): + FDv2SkillStore(SDK_KEY, mode="mobile") # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# The eager re-reconcile, end to end over the transport +# --------------------------------------------------------------------------- + + +class TestWatchSkillsOverTheTransport: + """ + ``watch_skills`` against a live ``FDv2SkillStore``. The watcher's own + behaviour — debounce, refusal of a store without ``add_listener``, detaching + on close — is covered in ``test_skills_watch.py`` against the in-memory + store; these are the cases that only mean something with a transport + underneath: a wire-level revocation, a new skill version, and an outage. + """ + + async def test_a_revocation_prunes_without_a_restart( + self, endpoint: Any, tmp_path: Any + ) -> None: + """ + The store's change listener drives the reconcile, so the file goes away + seconds after the ``delete-object`` rather than at the next process start. + """ + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll( + events( + ("server-intent", server_intent("xfer-changes")), + ("delete-object", delete_skill()), + ("payload-transferred", transferred("basis-2")), + ) + ) + endpoint.queue_poll(status=304) + + with poll_store(endpoint, poll_interval=0.2) as store: + store.wait_for_skills(timeout=5) + await init_client(options={"skillStore": store}, client=object()) + report, watcher = await watch_skills( + "*", tmp_path / "skills", debounce=0.05 + ) + try: + written = tmp_path / "skills" / "pdf-extraction" / "SKILL.md" + assert written.exists() + assert any(a.action == "written" for a in report.actions) + assert wait_until(lambda: not written.exists(), timeout=10) + finally: + watcher.close() + + async def test_a_new_version_is_rewritten_without_a_restart( + self, endpoint: Any, tmp_path: Any + ) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill(content="first")))) + endpoint.queue_poll( + events( + ("server-intent", server_intent("xfer-full")), + ("put-object", put_skill(object_version=4, content="second")), + ("payload-transferred", transferred("basis-2")), + ) + ) + endpoint.queue_poll(status=304) + + with poll_store(endpoint, poll_interval=0.2) as store: + store.wait_for_skills(timeout=5) + await init_client(options={"skillStore": store}, client=object()) + _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.05) + try: + written = tmp_path / "s" / "pdf-extraction" / "SKILL.md" + assert written.read_text() == "first" + assert wait_until(lambda: written.read_text() == "second", timeout=10) + finally: + watcher.close() + + async def test_the_default_keeps_last_known_good_during_an_outage( + self, endpoint: Any, tmp_path: Any + ) -> None: + """``on_unavailable="keep"`` is the default: an outage must not read as + "everything was revoked".""" + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=500) + with poll_store(endpoint, poll_interval=0.05) as store: + store.wait_for_skills(timeout=5) + await init_client(options={"skillStore": store}, client=object()) + _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.05) + try: + written = tmp_path / "s" / "pdf-extraction" / "SKILL.md" + assert written.exists() + # ``last_error`` rather than ``connection_failures``: the counter + # resets on the next successful poll, so asserting on it races + # the retry that is supposed to happen. + assert wait_until( + lambda: store.diagnostics.last_error is not None, timeout=10 + ) + time.sleep(0.3) + assert written.exists() + finally: + watcher.close() + + +# --------------------------------------------------------------------------- +# Listener registration +# --------------------------------------------------------------------------- + + +class TestListenerRegistration: + @staticmethod + def _skill_listeners(store: Any) -> list[Any]: + return list(store._listeners.get(SKILL_OBJECT_KIND, [])) + + def test_fdv2_remove_listener_of_an_unregistered_callable_is_a_no_op( + self, endpoint: Any + ) -> None: + with poll_store(endpoint) as store: + store.remove_listener(SKILL_OBJECT_KIND, print) + store.add_listener(SKILL_OBJECT_KIND, print) + store.remove_listener("flag", print) + store.remove_listener(SKILL_OBJECT_KIND, print) + store.remove_listener(SKILL_OBJECT_KIND, print) + assert self._skill_listeners(store) == [] + + +# --------------------------------------------------------------------------- +# Lifecycle +# --------------------------------------------------------------------------- + + +class TestLifecycle: + def test_start_is_idempotent(self, endpoint: Any) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + store = poll_store(endpoint) + try: + assert store.start() is store + assert store.start() is store + assert store.wait_for_skills(timeout=5) + finally: + store.close() + + def test_close_is_idempotent(self, endpoint: Any) -> None: + store = poll_store(endpoint) + store.start() + store.close() + store.close() + + def test_close_during_a_slow_connect_returns_promptly(self) -> None: + # Before the connect returns there is no connection for close() to + # interrupt. If the delivery thread then enters the read anyway, close() + # sits out its whole join timeout on a stream that will never speak. + requester = _SlowConnectRequester() + store = stream_store(_requester=requester) + store.start() + assert requester.entered.wait(timeout=5) + threading.Timer(0.1, requester.release.set).start() + started = time.monotonic() + store.close(timeout=5.0) + elapsed = time.monotonic() - started + assert elapsed < 2.0 + assert store._thread is not None + assert not store._thread.is_alive() + + def test_a_closed_store_still_answers_from_what_it_received( + self, endpoint: Any + ) -> None: + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + store = poll_store(endpoint) + store.start() + store.wait_for_skills(timeout=5) + store.close() + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + + def test_wait_for_skills_times_out_rather_than_hanging(self) -> None: + store = FDv2SkillStore( + SDK_KEY, mode="poll", poll_interval=60, _requester=_ScriptedRequester() + ) + try: + assert store.wait_for_skills(timeout=0.05) is False + finally: + store.close() + + def test_the_store_satisfies_the_seam_before_it_starts(self) -> None: + store = FDv2SkillStore(SDK_KEY) + assert store.get_object(SKILL_OBJECT_KIND, "anything") is None + assert store.all_objects(SKILL_OBJECT_KIND) == {} + + +# --------------------------------------------------------------------------- +# Timeouts +# --------------------------------------------------------------------------- + + +class _BlackHole: + """ + A listening socket that accepts connections and never sends a byte. + + This is the host ``read_timeout`` exists for: the TCP handshake completes, so + nothing fails fast, and then no response ever comes. A request against it can + only end by timing out, which makes the elapsed time a direct measurement of + the timeout actually applied. + """ + + def __init__(self) -> None: + self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._listener.bind(("127.0.0.1", 0)) + self._listener.listen(8) + self._accepted: list[socket.socket] = [] + self._stop = threading.Event() + self._thread = threading.Thread(target=self._accept_forever, daemon=True) + self._thread.start() + host, port = self._listener.getsockname() + self.base_uri = f"http://{host}:{port}" + + def _accept_forever(self) -> None: + self._listener.settimeout(0.05) + while not self._stop.is_set(): + try: + conn, _ = self._listener.accept() + except OSError: + continue + self._accepted.append(conn) + + def close(self) -> None: + self._stop.set() + self._thread.join(timeout=2) + for conn in self._accepted: + conn.close() + self._listener.close() + + +@pytest.fixture +def black_hole() -> Any: + server = _BlackHole() + yield server + server.close() + + +class TestTimeouts: + """ + ``read_timeout`` is the only network timeout, and every request honours it. + + The bounds asserted here are loose on purpose: the point is that a request + against an unresponsive host fails in roughly ``read_timeout`` rather than in + minutes, and that a regression back to a much longer default fails this + suite quickly instead of hanging it. + """ + + def test_a_poll_against_an_unresponsive_host_fails_within_read_timeout( + self, black_hole: Any + ) -> None: + requester = _Requester(SDK_KEY, black_hole.base_uri, read_timeout=0.3) + started = time.monotonic() + with pytest.raises(_RecoverableTransportError) as excinfo: + requester.poll(None, None) + elapsed = time.monotonic() - started + assert 0.2 <= elapsed < 2.0 + assert "timed out" in str(excinfo.value) + + def test_a_stream_against_an_unresponsive_host_fails_within_read_timeout( + self, black_hole: Any + ) -> None: + requester = _Requester(SDK_KEY, black_hole.base_uri, read_timeout=0.3) + started = time.monotonic() + with pytest.raises(_RecoverableTransportError): + requester.stream(None) + assert time.monotonic() - started < 2.0 + + def test_the_store_reports_the_timeout_and_keeps_going( + self, black_hole: Any + ) -> None: + store = FDv2SkillStore( + SDK_KEY, + base_uri=black_hole.base_uri, + mode="poll", + poll_interval=0.05, + initial_backoff=0.01, + max_backoff=0.05, + read_timeout=0.3, + ) + try: + store.start() + assert wait_until(lambda: store.diagnostics.connection_failures >= 1) + assert store.failed is None + assert "timed out" in (store.diagnostics.last_error or "") + finally: + store.close() + + def test_the_default_bound_depends_on_the_mode(self) -> None: + assert DEFAULT_POLL_TIMEOUT == 10.0 + assert DEFAULT_STREAM_READ_TIMEOUT == 300.0 + polling = FDv2SkillStore(SDK_KEY, mode="poll") + streaming = FDv2SkillStore(SDK_KEY, mode="stream") + assert polling._requester._read_timeout == DEFAULT_POLL_TIMEOUT + assert streaming._requester._read_timeout == DEFAULT_STREAM_READ_TIMEOUT + + @pytest.mark.parametrize("mode", ["poll", "stream"]) + def test_an_explicit_read_timeout_overrides_the_default(self, mode: Any) -> None: + store = FDv2SkillStore(SDK_KEY, mode=mode, read_timeout=42.0) + assert store._requester._read_timeout == 42.0 + + @pytest.mark.parametrize("value", [0.0, -1.0, float("inf"), float("nan")]) + def test_a_non_positive_read_timeout_is_rejected(self, value: float) -> None: + with pytest.raises(ValueError, match="read_timeout"): + FDv2SkillStore(SDK_KEY, read_timeout=value) + + def test_there_is_no_separate_connect_timeout(self) -> None: + # ``urllib`` cannot bound the connect separately from the reads, so the + # constructor does not offer a parameter that would only pretend to. + with pytest.raises(TypeError): + FDv2SkillStore(SDK_KEY, connect_timeout=2.0) # type: ignore[call-arg] diff --git a/packages/client/tests/test_skills_watch.py b/packages/client/tests/test_skills_watch.py index 48a95f0..b44ffea 100644 --- a/packages/client/tests/test_skills_watch.py +++ b/packages/client/tests/test_skills_watch.py @@ -4,7 +4,10 @@ The watcher is wired to the ``SkillStore`` interface, not to any one transport: it needs a store that implements ``add_listener``, and nothing more. These tests therefore drive it from ``InMemorySkillStore``, whose ``put`` notifies its -listeners synchronously, and from small hand-written store doubles. +listeners synchronously, and from small hand-written store doubles. The +end-to-end path — a ``delete-object`` arriving over a live FDv2 connection and +pruning a skill's files — is exercised in ``test_skills_fdv2.py``, where the fake +endpoint lives. Every test writes only inside pytest's ``tmp_path``. The watcher runs a real worker thread, so tests wait on observable outcomes rather than on fixed sleeps From c285ff533991b3e74a2d0b976a91de1008302596 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Fri, 11 Sep 2026 14:04:26 -0400 Subject: [PATCH 2/2] fix(client): retry an FDv2 stream that dies mid-read _Requester.stream wrapped only the connect as recoverable, so a read timeout, reset or truncated chunk in the body reached the delivery loop as whatever the socket raised. The loop read that as a bug and gave up: delivery stopped for the process lifetime, taking updates and revocations with it, the first time a socket died. read_timeout exists to bound a stream that has gone quiet so the loop can reconnect, and tripping it did the opposite. The body now carries the same promise the connect already did. Wrapping the line source rather than the whole read keeps protocol reader errors out of it: those are raised from the consumer's loop body, where they still surface as the bugs they are. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 21 ++++++- packages/client/tests/test_skills_fdv2.py | 61 +++++++++++++++++++ 2 files changed, 81 insertions(+), 1 deletion(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index da84a3b..18b8184 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -1032,6 +1032,25 @@ def _decode_poll_body(body: bytes) -> list[tuple[str, Any]]: return events +def _iter_stream_lines(response: Any) -> Any: + """ + Yields a streaming body's raw lines, presenting a read failure as retryable. + + A live stream dies mid-body far more often than it refuses to open: a read + timeout on a stream that went quiet, a reset, a truncated chunk. Each of + those arrives as whatever the socket raised, and the delivery loop retries + only the transport errors this module defines — anything else it reads as a + bug and stops for the process lifetime. Connecting is already wrapped in + ``_Requester.stream``; this is the same promise for the body. + """ + try: + yield from response + except Exception as exc: + raise _RecoverableTransportError( + f"reading the FDv2 stream failed: {type(exc).__name__}: {exc}" + ) from exc + + def _iter_sse(response: Any) -> Any: """ Decodes an SSE body into ``(event name, data)`` pairs. @@ -1042,7 +1061,7 @@ def _iter_sse(response: Any) -> Any: try: name: str | None = None data_lines: list[str] = [] - for raw_line in response: + for raw_line in _iter_stream_lines(response): line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n") if line == "": if name is not None: diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 0cce2f0..af5d181 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -23,6 +23,7 @@ import socket import threading import time +from http.client import IncompleteRead from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, ClassVar from urllib.parse import parse_qs, urlparse @@ -52,6 +53,7 @@ _retry_after_seconds, _SkillObjectSet, _store_object_from_put, + _StreamConnection, _tombstone_from_delete, ) @@ -1287,6 +1289,41 @@ def test_content_survives_a_reconnect(self, endpoint: Any) -> None: # --------------------------------------------------------------------------- +class _DyingResponse: + """ + A streaming body that transfers a payload and then fails mid-read. + + This is how a live stream actually ends: not with a clean end of body but + with a read timeout on a stream that went quiet, or a reset from the server + or a proxy in between. + """ + + def __init__(self, exc: BaseException) -> None: + self._exc = exc + + def __iter__(self) -> Any: + for event in full_payload(("put-object", put_skill())): + yield f"event: {event['event']}\n".encode() + yield f"data: {json.dumps(event['data'])}\n".encode() + yield b"\n" + raise self._exc + + def close(self) -> None: + pass + + +class _DyingStreamRequester: + """Every connection transfers a payload, then dies with *exc* mid-read.""" + + def __init__(self, exc: BaseException) -> None: + self.connections = 0 + self._exc = exc + + def stream(self, basis: str | None) -> Any: + self.connections += 1 + return _StreamConnection(_DyingResponse(self._exc)) + + class _ScriptedConnection: """Stands in for ``_StreamConnection``: an event iterator plus a close.""" @@ -1477,6 +1514,30 @@ def test_recycled_stream_connections_are_not_failures(self) -> None: finally: store.close() + @pytest.mark.parametrize( + "exc", + [ + TimeoutError("timed out"), + ConnectionResetError(54, "Connection reset by peer"), + IncompleteRead(b"partial"), + ], + ids=["read timeout", "reset", "truncated body"], + ) + def test_a_stream_that_dies_mid_read_reconnects(self, exc: BaseException) -> None: + # A stream fails in its body far more often than at its connect, and + # ``read_timeout`` exists to bound one that has gone quiet. Treating + # such a failure as unexpected would stop delivery — including + # revocation — for the process lifetime the first time a socket died. + requester = _DyingStreamRequester(exc) + store = stream_store(max_consecutive_failures=3, _requester=requester) + try: + store.start() + assert store.wait_for_skills(timeout=5) is True + assert wait_until(lambda: requester.connections >= 5) + assert store.failed is None + finally: + store.close() + def test_a_stream_commit_resets_the_failure_count(self) -> None: payload = [ (e["event"], e["data"]) for e in full_payload(("put-object", put_skill()))