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..18b8184 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,705 @@ 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_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. + + 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 _iter_stream_lines(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..af5d181 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -1,33 +1,65 @@ """ -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.client import IncompleteRead +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, + _StreamConnection, _tombstone_from_delete, ) 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 +175,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 +1035,667 @@ 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 _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.""" + + 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() + + @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())) + ] + 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 +1723,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 +1821,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