From 6ac1577be4d6024d4f8053fa48785dde1052a287 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 09:56:43 -0400 Subject: [PATCH 01/10] fix(client): point FDv2 streaming at the streaming host MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `FDv2SkillStore` built both `/sdk/poll` and `/sdk/stream` from a single `base_uri`, so the default configuration sent its streaming request to `sdk.launchdarkly.com`. LaunchDarkly serves streaming from a separate host, and `mode="stream"` is this store's default (TESTING.md §3.25), so this was the default path rather than an edge case: the first contact with a real environment would have gone to the wrong host. Both base server-side SDKs ship the two as distinct defaults, which is the evidence this follows: ldclient.config.Config base_uri='https://app.launchdarkly.com' stream_uri='https://stream.launchdarkly.com' @launchdarkly/js-server-sdk-common baseUri: 'https://sdk.launchdarkly.com' streamUri: 'https://stream.launchdarkly.com' `DEFAULT_BASE_URI` here already pointed at the FDv2 polling host (`sdk.launchdarkly.com`, not `ldclient`'s older `app.` value), so only the streaming host is new. Adds `DEFAULT_STREAM_URI` and a `stream_uri` keyword, mirroring `skills-fdv2.ts`: a `base_uri` given on its own still applies to both endpoints, so a relay or a private instance serving both from one host needs only the one option, and naming both overrides them independently. `_require_https_base_uri` becomes `_require_https_uri` and takes the option name, so a cleartext `stream_uri` is refused under its own name. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 94 ++++++++++++++----- packages/client/tests/test_skills_fdv2.py | 76 +++++++++++++++ 2 files changed, 147 insertions(+), 23 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index c0435d1..8036f27 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -86,8 +86,25 @@ """ DEFAULT_BASE_URI = "https://sdk.launchdarkly.com" -"""Where the SDK-facing FDv2 endpoints live. Overridable for Federal and private -instances.""" +"""Where ``GET /sdk/poll`` is served. Overridable for Federal instances, private +instances, and relay deployments.""" + +DEFAULT_STREAM_URI = "https://stream.launchdarkly.com" +""" +Where ``GET /sdk/stream`` is served. + +LaunchDarkly serves streaming from a **different host** than polling, which is +why this is a second default rather than a path under ``DEFAULT_BASE_URI``. Both +base server-side SDKs ship the pair: ``ldclient.config.Config`` defaults +``stream_uri`` to ``https://stream.launchdarkly.com`` alongside its own polling +host, and ``@launchdarkly/js-server-sdk-common`` does the same. ``mode="stream"`` +is this store's default, so a single-host default would have the default +configuration connect to the wrong host on first contact with a real environment. + +A *base_uri* given on its own applies to both endpoints, because a relay or a +private instance serving both from one host should need only one option; see +``FDv2SkillStore.__init__``. +""" POLL_PATH = "/sdk/poll" STREAM_PATH = "/sdk/stream" @@ -202,9 +219,9 @@ def _require_server_side_credential(sdk_key: str) -> None: """The only hosts a plain ``http://`` base URI may name: a local test double.""" -def _require_https_base_uri(base_uri: str) -> None: +def _require_https_uri(base_uri: str, option: str = "base_uri") -> None: """ - Refuses a base URI that would send the SDK key in cleartext. + Refuses a URI that would send the SDK key in cleartext. Every request carries the environment's server-side SDK key in ``Authorization``, so the transport is ``https://`` only. The one exception @@ -215,7 +232,7 @@ def _require_https_base_uri(base_uri: str) -> None: """ if not isinstance(base_uri, str) or not base_uri.strip(): raise ValueError( - "FDv2SkillStore requires an https:// base URI; none was given." + f"FDv2SkillStore requires an https:// URI for {option}; none was given." ) parts = urllib.parse.urlsplit(base_uri.strip()) if parts.scheme == "https" and parts.hostname: @@ -224,14 +241,15 @@ def _require_https_base_uri(base_uri: str) -> None: return if parts.scheme == "http": raise ValueError( - f"FDv2SkillStore refuses base_uri {base_uri!r}: a plain http:// URI " + f"FDv2SkillStore refuses {option} {base_uri!r}: a plain http:// URI " "would send the server-side SDK key in cleartext. Use https:// " - "(the default is https://sdk.launchdarkly.com). Plain http:// is " + "(the defaults are https://sdk.launchdarkly.com for polling and " + "https://stream.launchdarkly.com for streaming). Plain http:// is " "allowed only for a loopback host (localhost, 127.0.0.1, ::1) " "serving a local test double." ) raise ValueError( - f"FDv2SkillStore refuses base_uri {base_uri!r}: expected an https:// URI " + f"FDv2SkillStore refuses {option} {base_uri!r}: expected an https:// URI " "with a host, such as https://sdk.launchdarkly.com." ) @@ -1038,10 +1056,15 @@ def __init__( base_uri: str, *, read_timeout: float, + stream_uri: str | None = None, opener: Any = None, ) -> None: self._sdk_key = sdk_key self._base_uri = base_uri.rstrip("/") + # Streaming and polling are served from different hosts by LaunchDarkly; + # a caller that names only one host means both, which is a relay or a + # private instance. ``FDv2SkillStore`` resolves the two-default case. + self._stream_uri = (stream_uri or base_uri).rstrip("/") self._read_timeout = read_timeout # Injectable, so an alternative transport can be supplied. The default # never follows a redirect; see ``_RefuseRedirects``. @@ -1066,25 +1089,28 @@ def interrupt(self) -> None: if response is not None: _interrupt_read(response) - def _url(self, path: str, basis: str | None) -> str: + def _url(self, origin: str, path: str, basis: str | None) -> str: """ The request URL: the path, plus ``basis`` once a payload has committed. + *origin* is the host for this path — polling and streaming have one + each. + 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})}" + return f"{origin}{path}" + return f"{origin}{path}?{urllib.parse.urlencode({'basis': basis})}" def _request( - self, path: str, basis: str | None, headers: dict[str, str] + self, origin: str, 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" + self._url(origin, path, basis), headers=all_headers, method="GET" ) def poll(self, basis: str | None, etag: str | None) -> _PollResult: @@ -1092,7 +1118,7 @@ def poll(self, basis: str | None, etag: str | None) -> _PollResult: headers = {"Accept": "application/json"} if etag: headers["If-None-Match"] = etag - request = self._request(POLL_PATH, basis, headers) + request = self._request(self._base_uri, POLL_PATH, basis, headers) try: with self._opener.open(request, timeout=self._read_timeout) as response: with self._lock: @@ -1126,6 +1152,7 @@ def poll(self, basis: str | None, etag: str | None) -> _PollResult: def stream(self, basis: str | None) -> _StreamConnection: """Opens ``GET /sdk/stream``.""" request = self._request( + self._stream_uri, STREAM_PATH, basis, {"Accept": "text/event-stream", "Cache-Control": "no-cache"}, @@ -1328,11 +1355,17 @@ class FDv2SkillStore: **Server-side only.** A mobile key or a client-side environment ID is refused in the constructor. - **The SDK key goes only where it was pointed.** *base_uri* must be - ``https://`` — plain ``http://`` is refused except to a loopback host, for - local test doubles — and redirects are never followed, so a 3xx from a proxy - or a private instance is a fatal failure rather than a request carrying the - key to whatever host ``Location`` named. + **The SDK key goes only where it was pointed.** *base_uri* and *stream_uri* + must each be ``https://`` — plain ``http://`` is refused except to a + loopback host, for local test doubles — and redirects are never followed, so + a 3xx from a proxy or a private instance is a fatal failure rather than a + request carrying the key to whatever host ``Location`` named. + + **Polling and streaming have separate hosts.** LaunchDarkly serves them from + different origins, so the defaults are a pair (``DEFAULT_BASE_URI`` and + ``DEFAULT_STREAM_URI``). A *base_uri* given on its own applies to both, + which is what a relay or a private instance serving both endpoints from one + host needs. **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 @@ -1362,7 +1395,8 @@ def __init__( self, sdk_key: str, *, - base_uri: str = DEFAULT_BASE_URI, + base_uri: str | None = None, + stream_uri: str | None = None, mode: Mode = "stream", poll_interval: float = 30.0, read_timeout: float | None = None, @@ -1372,8 +1406,14 @@ def __init__( _requester: Any = None, ) -> None: """ - *base_uri* must be ``https://``; ``http://`` is accepted only for - ``localhost``, ``127.0.0.1`` or ``::1``. Raises ``ValueError`` otherwise. + *base_uri* is where ``GET /sdk/poll`` is sent (``DEFAULT_BASE_URI``) and + *stream_uri* where ``GET /sdk/stream`` is sent (``DEFAULT_STREAM_URI``), + because LaunchDarkly serves the two from different hosts. A *base_uri* + given **without** a *stream_uri* is used for both, which is what a relay + or a private instance serving both endpoints from one host needs; naming + both overrides them independently. Each must be ``https://``; + ``http://`` is accepted only for ``localhost``, ``127.0.0.1`` or + ``::1``. Raises ``ValueError`` otherwise. *mode* is ``"stream"`` by default. Prefer it: a ``delete-object`` reaches a live stream in seconds. ``"poll"`` exists for environments that cannot @@ -1395,7 +1435,14 @@ def __init__( payload resets the count. """ _require_server_side_credential(sdk_key) - _require_https_base_uri(base_uri) + # A lone ``base_uri`` means "both endpoints are here"; the two-host + # default applies only when neither was named. + if stream_uri is None: + stream_uri = DEFAULT_STREAM_URI if base_uri is None else base_uri + if base_uri is None: + base_uri = DEFAULT_BASE_URI + _require_https_uri(base_uri) + _require_https_uri(stream_uri, "stream_uri") if mode not in ("stream", "poll"): raise ValueError(f'mode must be "stream" or "poll", got {mode!r}') if poll_interval <= 0: @@ -1427,6 +1474,7 @@ def __init__( sdk_key.strip(), base_uri, read_timeout=read_timeout, + stream_uri=stream_uri, ) self._stop = threading.Event() diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index d7affd6..3adb5e0 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -42,8 +42,10 @@ ) from launchdarkly_ai_server.skills_core import SKILL_OBJECT_KIND from launchdarkly_ai_server.skills_fdv2 import ( + DEFAULT_BASE_URI, DEFAULT_POLL_TIMEOUT, DEFAULT_STREAM_READ_TIMEOUT, + DEFAULT_STREAM_URI, FDV2_KEY_DELIMITER, FDV2_OBJECT_KIND, MAX_RESPONSE_BYTES, @@ -2543,6 +2545,80 @@ def test_https_is_accepted(self) -> None: assert FDv2SkillStore(SDK_KEY, base_uri="https://sdk.example.com/") is not None assert FDv2SkillStore(SDK_KEY) is not None + def test_a_plain_http_stream_uri_is_refused_by_name(self) -> None: + """The streaming host is checked too, and the message names it.""" + with pytest.raises(ValueError, match="cleartext") as excinfo: + FDv2SkillStore( + SDK_KEY, + base_uri="https://sdk.example.com", + stream_uri="http://stream.example.com", + ) + assert "stream_uri" in str(excinfo.value) + + +class TestStreamHostDefaults: + """ + LaunchDarkly serves ``/sdk/stream`` from a different host than ``/sdk/poll``, + and ``mode="stream"`` is the default — so a single-host default would have + the *default* configuration talk to the wrong host on first contact with a + real environment. Both base server-side SDKs ship the hosts as a pair + (``ldclient.Config``'s ``stream_uri``, ``js-server-sdk-common``'s + ``streamUri``), which is what these defaults follow. + """ + + @staticmethod + def _origins(store: FDv2SkillStore) -> tuple[str, str]: + requester = store._requester + return requester._base_uri, requester._stream_uri + + def test_the_two_defaults_are_different_hosts(self) -> None: + assert DEFAULT_BASE_URI == "https://sdk.launchdarkly.com" + assert DEFAULT_STREAM_URI == "https://stream.launchdarkly.com" + assert self._origins(FDv2SkillStore(SDK_KEY)) == ( + DEFAULT_BASE_URI, + DEFAULT_STREAM_URI, + ) + + def test_a_base_uri_alone_serves_both_endpoints(self) -> None: + # A relay or a private instance serving both from one host needs one + # option, not two. + assert self._origins( + FDv2SkillStore(SDK_KEY, base_uri="https://relay.example.com") + ) == ("https://relay.example.com", "https://relay.example.com") + + def test_naming_both_overrides_them_independently(self) -> None: + assert self._origins( + FDv2SkillStore( + SDK_KEY, + base_uri="https://sdk.example.com/", + stream_uri="https://stream.example.com/", + ) + ) == ("https://sdk.example.com", "https://stream.example.com") + + def test_a_stream_uri_alone_leaves_polling_on_its_default(self) -> None: + assert self._origins( + FDv2SkillStore(SDK_KEY, stream_uri="https://stream.example.com") + ) == (DEFAULT_BASE_URI, "https://stream.example.com") + + def test_the_stream_request_goes_to_the_stream_host( + self, endpoint: Any, second_endpoint: Any + ) -> None: + """The split is on the wire, not only in the attributes. + + Polling at one host and streaming at another is the whole point, so + assert it where it is observable: the streaming request arrives at the + streaming endpoint and nothing arrives at the polling one. + """ + requester = _Requester( + SDK_KEY, + endpoint.base_uri, + read_timeout=5.0, + stream_uri=second_endpoint.base_uri, + ) + requester.stream(None).close() + assert [r["path"] for r in second_endpoint.requests] == ["/sdk/stream"] + assert endpoint.requests == [] + # --------------------------------------------------------------------------- # The eager re-reconcile, end to end over the transport From c69d880563540c47e0ef8c42074daef2888a4f86 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 09:57:40 -0400 Subject: [PATCH 02/10] fix(client): floor Retry-After, and count only revocations that landed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two independent transport defects. **`Retry-After: 0` had no floor.** `_retry_after_seconds` returns `max(0.0, seconds)`, and the retry site capped the value but never floored it, so a server — or an intermediate proxy — answering `0` had the delivery loop reconnect as fast as it could schedule, spending the whole of `max_consecutive_failures` in milliseconds and hammering the endpoint on the way. It is bounded, so nothing runs away, but the retry budget is gone and `Retry-After` is honoured in the least useful direction. Floored at `initial_backoff`, the same floor our own backoff starts from, matching `skills-fdv2.ts`. A short `Retry-After` is therefore raised to `initial_backoff`, which is what made `test_a_retry_after_header_is_honoured` discriminate before: it asked for 0.25s against a 5s initial backoff. Rewritten to ask for *longer* than our own backoff instead, so honouring the header is still the only way the assertion can pass, and `test_a_retry_after_header_is_parsed_off_the_wire` likewise — it was passing only because `poll_store`'s 0.05s `max_backoff` capped the floored value, so it no longer told the two apart. **`objects_revoked` counted tombstones for keys the store never held.** The increment fired whenever a `delete-object` parsed, whether or not it took anything away. The counter is operator-facing and gets read exactly when somebody is working out whether a revocation landed (TESTING.md §3.25, "Counters are assertable facts"), so an inflated figure misleads at the worst moment. Gated on `_SkillObjectSet.delete` having returned something. `changes` still carries every tombstone, so listeners are unaffected. Note this is a divergence from TypeScript rather than a port of it: `skills-fdv2.ts`'s `deleteObject` increments unconditionally too, and its `keysFullyRevoked` applies to the full-transfer revoke-by-omission diff, which this side does not implement. The same fix is owed there. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 18 +++- packages/client/tests/test_skills_fdv2.py | 82 +++++++++++++++++-- 2 files changed, 89 insertions(+), 11 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index 8036f27..0981e6f 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -704,8 +704,14 @@ def _delete_object(self, data: Any) -> _TransferOutcome: tombstone = _tombstone_from_delete(data) if tombstone is None: return _TransferOutcome() - target.delete(tombstone) - self.diagnostics.objects_revoked += 1 + removed = target.delete(tombstone) + if removed: + # Only a tombstone that took something away is a revocation. A + # delete for a key the store never held revoked nothing, and + # counting it inflates the one counter an operator reads to work + # out whether a revocation actually landed. ``changes`` below + # carries the tombstone either way, so a listener still sees it. + self.diagnostics.objects_revoked += 1 # A revocation identifies the skill payload just as a put does. self._skills_in_payload += 1 # A tombstone carries identity and no content; see @@ -1753,6 +1759,14 @@ def _run(self) -> None: delay = _backoff_delay( failures, base=self._initial_backoff, maximum=self._max_backoff ) + else: + # A server asking for no delay still gets one: honouring + # ``Retry-After: 0`` literally would reconnect as fast as + # the loop allows and burn the whole retry bound in + # milliseconds, hammering the endpoint on the way. The + # floor is ``initial_backoff``, the same floor our own + # backoff starts from. + delay = max(delay, self._initial_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. diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 3adb5e0..8fa1acb 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -639,6 +639,33 @@ def test_a_delete_object_revokes_the_skill(self) -> None: assert held.get("pdf-extraction", None) is None assert reader.diagnostics.objects_revoked == 1 + def test_a_delete_for_a_key_never_held_is_not_counted_as_a_revocation( + self, + ) -> None: + """``objects_revoked`` counts what went away, not tombstones seen. + + The counter is operator-facing, and it is read precisely when somebody + is working out whether a revocation landed. A delete for a key the store + never held revoked nothing, so counting it inflates the one number that + answers that question. The tombstone still reaches listeners through + ``changes``, which is where "every revocation the server stated" lives. + """ + held = _SkillObjectSet() + reader = _ProtocolReader(held) + drive(reader, full_payload(("put-object", put_skill(key="kept")))) + outcomes = drive( + reader, + events( + ("server-intent", server_intent("xfer-changes")), + ("delete-object", delete_skill(key="never-delivered")), + ("payload-transferred", transferred("basis-2")), + ), + ) + assert reader.diagnostics.objects_revoked == 0 + assert held.get("kept", None) is not None + # Reported, just not counted. + assert outcomes[-1].changes == [{"key": "never-delivered", "version": 3}] + def test_a_delete_notifies_with_a_tombstone_carrying_no_content(self) -> None: held = _SkillObjectSet() reader = _ProtocolReader(held) @@ -1814,31 +1841,68 @@ def test_stream_retries_are_bounded(self) -> None: def test_a_retry_after_header_is_honoured(self) -> None: requester = _ScriptedRequester( - _RecoverableTransportError("slow down", retry_after=0.25), + _RecoverableTransportError("slow down", retry_after=0.5), ) store = FDv2SkillStore( SDK_KEY, mode="poll", poll_interval=10.0, - initial_backoff=5.0, + initial_backoff=0.01, + max_backoff=5.0, _requester=requester, ) try: started = time.monotonic() store.start() - assert wait_until(lambda: len(requester.calls) >= 2, timeout=3) + assert wait_until(lambda: len(requester.calls) >= 2, timeout=5) elapsed = time.monotonic() - started - # The server asked for 0.25s; our own backoff would have been 5s. - assert 0.2 <= elapsed < 3.0 + # The server asked for 0.5s and our own backoff would have been + # 0.01s, so waiting is the only way the header could have been read. + # Asked *longer* rather than shorter on purpose: a shorter request + # is floored at ``initial_backoff``, so it cannot discriminate. + assert elapsed >= 0.4 + finally: + store.close() + + def test_a_retry_after_of_zero_still_waits_the_initial_backoff(self) -> None: + """``Retry-After: 0`` is floored, not taken literally. + + A server — or an intermediate proxy — answering ``0`` would otherwise + have the loop reconnect as fast as it can schedule, spending the whole + bounded retry budget in milliseconds and hammering the endpoint on the + way. The floor is ``initial_backoff``, the same floor our own backoff + starts from. + """ + requester = _ScriptedRequester( + _RecoverableTransportError("slow down", retry_after=0.0), + ) + store = FDv2SkillStore( + SDK_KEY, + mode="poll", + poll_interval=10.0, + initial_backoff=0.5, + max_backoff=5.0, + _requester=requester, + ) + try: + started = time.monotonic() + store.start() + assert wait_until(lambda: len(requester.calls) >= 2, timeout=5) + assert time.monotonic() - started >= 0.4 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(status=429, retry_after="0.5") 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 + started = time.monotonic() + with poll_store( + endpoint, initial_backoff=0.01, max_backoff=5.0, poll_interval=10.0 + ) as store: + assert store.wait_for_skills(timeout=5) is True + # 0.5s is only obtainable from the header: our own backoff here is 0.01s + # and the cap is 5s, so neither could have produced this wait. + assert time.monotonic() - started >= 0.4 @pytest.mark.parametrize("raw", ["inf", "Infinity", "-inf", "nan", "1e309"]) def test_a_non_finite_retry_after_is_ignored(self, raw: str) -> None: From 86fd04cca3f4e1b6985bc8ffb0d2a7db98d87447 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 09:58:27 -0400 Subject: [PATCH 03/10] fix(client): make close() final, and reclassify HTTP 400 and 404 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three §3.25 deviations, all in the transport's lifecycle and reconnection rules. Grouped because each one's existing test pinned the old answer and had to move in the same change. **`close()` is final** (§3.25 lifecycle). A subsequent `start()` now raises instead of resuming delivery. Finality is what gives `close()` a postcondition a caller can rely on — delivery has stopped — including when the join times out on a thread parked somewhere no interrupt reaches. A store that could be restarted from there leaves the caller unable to tell whether delivery stopped, and a restart that silently never delivered again is the failure this forecloses. To resume, build a new store. Giving up at the failure bound is *not* closing, so a store that gave up can still be restarted and the re-arm path is unchanged. `test_a_restarted_store_waits_again` therefore drives the give-up path now, and `test_a_close_that_timed_out_leaves_the_store_restartable` becomes `..._is_still_final` — it was asserting exactly the behaviour finality removes. **HTTP 404 is fatal** (§3.25 reconnection). A 404 on `/sdk/poll` or `/sdk/stream` means the endpoint does not exist for this credential or instance — a mistyped base URI, typically — and no reconnect produces one. It was recoverable, so it burned the whole retry budget first. **HTTP 400 is retried exactly once, then fatal** (§3.25 reconnection). A 400 is what a stale `basis` selector looks like: the request carried state the server no longer recognises, and a connection built from nothing fixes it. It was fatal, so that repair never ran. Implemented as `_StaleRequestStateError`, a `_RecoverableTransportError` subclass whose handling drops `basis` and `etag` and asks for a full transfer — and gives up when there was neither to drop, since a request carrying no client state was refused on its own terms. That bound is what keeps this from being "400 is recoverable", and it needs no counter. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 105 ++++++++++-- packages/client/tests/test_skills_fdv2.py | 159 ++++++++++++++++-- 2 files changed, 237 insertions(+), 27 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index 0981e6f..24d5739 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -903,6 +903,25 @@ def __init__(self, message: str, retry_after: float | None = None) -> None: self.retry_after = retry_after +class _StaleRequestStateError(_RecoverableTransportError): + """ + An HTTP 400 for a request carrying client state — the ``basis`` selector, or + an ``If-None-Match`` etag. + + That state is the one part of the request that can go stale, so it is + dropped and a full transfer asked for **once** before the status is treated + as fatal. The bound is what keeps a 400 from being plain "recoverable": a + request carrying no state at all was itself refused, and no reconnect fixes + that. + """ + + +_REQUEST_ADVICE = ( + "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." +) + _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 " @@ -954,12 +973,24 @@ def _classify_status(status: int, headers: Any) -> Exception: "base URI, and any proxy in between, for the address being redirected " "to." ) - if status in (400, 405, 406, 414, 501): + if status == 404: + # The endpoint does not exist for this credential or instance — a + # mistyped base URI, typically. No reconnect produces one. + return _FatalTransportError( + "LaunchDarkly returned HTTP 404 for the FDv2 endpoint. Check the " + "base URI, and that this instance serves /sdk/poll and /sdk/stream." + ) + if status == 400: + # The one rejection this adapter can act on: the selector it sent may be + # one the server no longer accepts. Recoverable so the state can be + # dropped and a full transfer requested; fatal once that has been tried. + return _StaleRequestStateError( + f"LaunchDarkly returned HTTP 400. {_REQUEST_ADVICE}" + ) + if status in (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." + f"{_REQUEST_ADVICE}" ) return _RecoverableTransportError( f"LaunchDarkly returned HTTP {status}", _retry_after_seconds(headers) @@ -1373,6 +1404,10 @@ class FDv2SkillStore: which is what a relay or a private instance serving both endpoints from one host needs. + **``close`` is final.** A closed store still answers from what it received, + but delivery cannot be resumed: ``start`` afterwards raises. Construct a new + store instead. + **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 @@ -1483,6 +1518,16 @@ def __init__( stream_uri=stream_uri, ) + self._closed = False + """ + ``close`` has been called. Final: ``start`` raises afterwards. + + What gives ``close`` a postcondition a caller can rely on — "delivery + has stopped" — including when the join timed out. A store that could be + restarted after a timed-out close leaves the caller unable to tell + whether delivery stopped, and a restart that silently never delivers + again is the failure this forecloses. To resume, construct a new store. + """ self._stop = threading.Event() self._first_payload = threading.Event() """A payload has committed. The fact ``wait_for_skills`` reports.""" @@ -1518,8 +1563,21 @@ 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. + + Raises ``RuntimeError`` on a **closed** store: ``close`` is final, so + there is no resuming it. A store that gave up at its failure bound is + not closed and can be started again — the retry budget and the terminal + reason both belong to the run that spent them. """ with self._lock: + if self._closed: + raise RuntimeError( + "This FDv2SkillStore has been closed, and close() is final: " + "delivery cannot be resumed, so a restarted store would " + "report itself started and never deliver. Construct a new " + "FDv2SkillStore to resume delivery. Held content is still " + "readable from the closed store." + ) # Read before the rearm clears it: a thread inside ``_give_up`` is # still alive and no longer delivering, so ``is_alive`` on its own # would have this call adopt a run that is about to return and @@ -1531,13 +1589,9 @@ def start(self) -> FDv2SkillStore: ) self._rearm_waiters() if delivering: - # A ``close`` whose join timed out leaves the previous thread - # running with the stop flag still set. Clearing it lets that - # thread carry on delivering, rather than leaving a store that - # reports itself started and never delivers again. - self._stop.clear() return self - self._stop.clear() + # Only ``close`` sets the stop flag, and a closed store never gets + # here, so there is nothing to clear. self._thread = threading.Thread( target=self._run, name="ld-ai-skills-fdv2", daemon=True ) @@ -1546,9 +1600,9 @@ def start(self) -> FDv2SkillStore: def _rearm_waiters(self) -> None: """ - Re-arms ``wait_for_skills`` for a store being started again after a - ``close``. A payload already held stays an answer; an ended delivery - does not, or the next waiter would be released before it began. + Re-arms ``wait_for_skills`` for a store being started again after it + gave up. A payload already held stays an answer; an ended delivery does + not, or the next waiter would be released before it began. A terminal ``failed`` reason is dropped for the same reason: it says why delivery stopped for good, and delivery is about to run again. Leaving @@ -1566,12 +1620,21 @@ def _rearm_waiters(self) -> None: def close(self, timeout: float = 5.0) -> None: """ - Stops delivery. Idempotent, and safe to call from any thread. + Stops delivery. Idempotent, safe to call from any thread, and **final**: + a subsequent ``start`` raises rather than resuming. Construct a new + store to resume delivery. + + Finality is what gives this call a postcondition — delivery has + stopped — even when the join below times out on a thread parked + somewhere no interrupt reaches. A store that could be restarted from + there would leave the caller unable to tell whether delivery stopped. 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. """ + with self._lock: + self._closed = True self._stop.set() # A waiter parked in ``wait_for_skills`` is owed an answer now rather # than at the end of its timeout; delivery is over either way. @@ -1739,6 +1802,20 @@ def _run(self) -> None: # would spend a retry from the bounded budget and leave a # misleading ``last_error`` on a healthy store. return + if isinstance(exc, _StaleRequestStateError): + # The selector and the etag are the only client state in the + # request, so a rejection of a request carrying neither is + # the request itself being refused, and reconnecting cannot + # fix it. Carrying one, the state may be stale: drop it, ask + # for a full transfer, and let the next 400 be the fatal one. + with self._lock: + exhausted = self._basis is None and self._etag is None + if not exhausted: + self._basis = None + self._etag = None + if exhausted: + self._give_up(str(exc)) + return with self._lock: # Whatever the dropped connection had transferred so far is # not a payload; the next connection starts one afresh. diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 8fa1acb..577bf87 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -50,6 +50,7 @@ FDV2_OBJECT_KIND, MAX_RESPONSE_BYTES, _backoff_delay, + _classify_status, _FatalTransportError, _is_skill_event, _iter_sse, @@ -58,6 +59,7 @@ _Requester, _retry_after_seconds, _SkillObjectSet, + _StaleRequestStateError, _store_object_from_put, _StreamConnection, _tombstone_from_delete, @@ -1642,6 +1644,83 @@ def test_a_restart_returns_the_retry_budget(self, endpoint: Any) -> None: assert store.wait_for_skills(timeout=5) is True assert store.failed is None + def test_a_404_stops_delivery_immediately(self, endpoint: Any) -> None: + """A 404 means the endpoint does not exist for this credential. + + A mistyped base URI, typically, or an instance that does not serve the + FDv2 endpoints. No reconnect produces one, so it is fatal rather than + retried — and it is the exception to the shape the recoverable-by-default + rule would suggest. + """ + endpoint.queue_poll(status=404) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert wait_until(lambda: store.failed is not None) + assert store.wait_for_skills(timeout=1) is False + assert "404" in store.failed + assert "/sdk/poll" in store.failed + # Fatal means one request, not a retry that happened to find the payload. + assert len(endpoint.requests) == 1 + assert store.diagnostics.connection_failures == 0 + + def test_a_400_reconnects_once_from_scratch_and_is_then_fatal( + self, endpoint: Any + ) -> None: + """A 400 is what a stale ``basis`` selector looks like. + + The selector and the etag are the only client state the request carries, + so a fresh connection built from nothing is the one repair available. + It gets exactly one: the retry bound is what keeps this from being + "400 is recoverable". + """ + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=400) + endpoint.queue_poll(status=400) + 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 wait_until(lambda: store.failed is not None) + assert "400" in store.failed + # The first payload, then the retried request, then the fatal one. The + # fourth queued payload is never asked for. + assert len(endpoint.requests) == 3 + # The premise: the rejected request did carry client state to drop. + assert "basis" in endpoint.requests[1]["query"] + # The retry was from scratch: no selector and no etag on the way back. + retried = endpoint.requests[2] + assert retried["query"] == {} + assert retried["if_none_match"] is None + # Last known good survives both. + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + + def test_a_400_carrying_no_client_state_is_fatal_at_once( + self, endpoint: Any + ) -> None: + """There is nothing to drop on a first connection, so nothing to repair. + + A request that carried neither a selector nor an etag and was still + refused was refused on its own terms. + """ + endpoint.queue_poll(status=400) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert wait_until(lambda: store.failed is not None) + assert "400" in store.failed + assert len(endpoint.requests) == 1 + + def test_the_two_exceptional_statuses_are_classified_apart(self) -> None: + # The classification is the contract; the end-to-end tests above are + # what prove the loop honours it. + assert isinstance(_classify_status(404, None), _FatalTransportError) + assert isinstance(_classify_status(400, None), _StaleRequestStateError) + # A stale-state error is still a recoverable one, so the retry path + # reaches it at all. + assert isinstance(_classify_status(400, None), _RecoverableTransportError) + for status in (405, 406, 414, 501): + assert isinstance(_classify_status(status, None), _FatalTransportError) + assert isinstance(_classify_status(503, None), _RecoverableTransportError) + assert not isinstance(_classify_status(503, None), _StaleRequestStateError) + def test_a_401_stops_delivery(self, endpoint: Any) -> None: endpoint.queue_poll(status=401) with poll_store(endpoint) as store: @@ -2823,6 +2902,42 @@ def test_close_is_idempotent(self, endpoint: Any) -> None: store.close() store.close() + def test_a_closed_store_does_not_restart(self, endpoint: Any) -> None: + """``close`` is final, and a restart raises rather than resuming. + + Finality is what gives ``close`` a postcondition a caller can rely on — + delivery has stopped — including when the join timed out. A store that + could be restarted from there leaves the caller unable to tell whether + delivery stopped, and a restart that silently never delivered again is + the failure this forecloses. To resume, construct a new store. + """ + store = poll_store(endpoint) + store.start() + store.close() + with pytest.raises(RuntimeError, match="close\\(\\) is final") as excinfo: + store.start() + # The remedy is in the message, not only in the docs. + assert "Construct a new FDv2SkillStore" in str(excinfo.value) + + def test_a_store_closed_before_it_started_also_refuses_to_start( + self, endpoint: Any + ) -> None: + store = poll_store(endpoint) + store.close() + with pytest.raises(RuntimeError, match="close\\(\\) is final"): + store.start() + + def test_reentering_a_closed_store_as_a_context_manager_raises( + self, endpoint: Any + ) -> None: + # ``__enter__`` is ``start``, so finality reaches the ``with`` form too. + store = poll_store(endpoint) + with store: + pass + with pytest.raises(RuntimeError, match="close\\(\\) is final"): + with store: + pass + 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() @@ -3115,19 +3230,36 @@ def test_a_payload_already_held_still_answers_true_after_close(self) -> None: store.close() assert store.wait_for_skills(timeout=5) is True - def test_a_restarted_store_waits_again(self) -> None: - # The released flag is sticky by design, so a store closed before any - # payload and then started again has to re-arm: otherwise the next - # waiter is let go before delivery has had a chance to begin. - store = stream_store(_requester=_SilentStreamRequester()) + def test_a_store_restarted_after_giving_up_waits_again(self) -> None: + # The released flag is sticky by design, so a store that gave up before + # any payload and is then started again has to re-arm: otherwise the + # next waiter is let go before delivery has had a chance to begin. + # Restarting after a *close* is not available — see + # ``TestCloseIsFinal`` — so the give-up path is what exercises this. + class _FailsThenGoesQuiet(_FakeRequester): + """One failure, enough to give up; silent on every run after.""" + + def __init__(self) -> None: + self.attempts = 0 + + def stream(self, basis: str | None) -> Any: + self.attempts += 1 + if self.attempts == 1: + raise _RecoverableTransportError("x") + return _BlockingConnection() + + store = stream_store( + max_consecutive_failures=0, _requester=_FailsThenGoesQuiet() + ) store.start() - store.close() + assert wait_until(lambda: store.failed is not None) assert store.wait_for_skills(timeout=0.1) is False + store.start() try: started = time.monotonic() assert store.wait_for_skills(timeout=0.5) is False - # Waited, rather than being released by the previous close. + # Waited, rather than being released by the previous give-up. assert time.monotonic() - started >= 0.4 finally: store.close() @@ -3206,11 +3338,12 @@ def test_a_poll_we_interrupted_is_not_a_delivery_failure( assert store.diagnostics.last_error is None assert store.failed is None - def test_a_close_that_timed_out_leaves_the_store_restartable(self) -> None: + def test_a_close_that_timed_out_is_still_final(self) -> None: # A request blocked inside its connect is beyond any interrupt, so - # ``close`` can still return with the thread alive. ``start`` must not - # then find that thread and return with the stop flag set: the store - # would report itself started and never deliver again. + # ``close`` can still return with the thread alive. This is the case + # finality exists for: the caller cannot tell whether delivery stopped, + # and a ``start`` that adopted the dying thread would leave a store + # reporting itself started and never delivering. Raising says so. requester = _SlowPollRequester() store = FDv2SkillStore( SDK_KEY, mode="poll", poll_interval=0.01, _requester=requester @@ -3220,8 +3353,8 @@ def test_a_close_that_timed_out_leaves_the_store_restartable(self) -> None: assert requester.entered.wait(timeout=5) store.close(timeout=0.2) assert store._thread is not None and store._thread.is_alive() - store.start() - assert store._stop.is_set() is False + with pytest.raises(RuntimeError, match="close\\(\\) is final"): + store.start() finally: requester.release.set() store.close(timeout=2) From 2f59b9a64619fc43c7856ea6babe9b30ada8594e Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 09:59:31 -0400 Subject: [PATCH 04/10] fix(client): refuse a skill-store listener on a kind that never fires MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `add_listener` recorded a registration for any kind and then never notified it, on both shipped stores — `InMemorySkillStore`, whose `put` only accepts skill objects, and `FDv2SkillStore`, which only ever delivers `SKILL_OBJECT_KIND`. TESTING.md §3.21 now requires both to raise. The argument is §3.26's own, the one that already makes `watch_skills` refuse a store with no `add_listener` at all rather than degrading to a one-shot reconcile: a listener that silently never fires "looks exactly like a watcher whose skills never changed". Registering on a kind the store will never deliver is the same failure with the same signature, and a store that accepted the registration has promised something it cannot keep. `ValueError` rather than `TypeError`: `kind` is a `str`, which is the accepted type, carrying a value the store cannot serve — the same distinction A.12 records for `write_skills` versus `get_skills`. `remove_listener` for an unregistered kind stays a no-op, so a caller tearing down need not track what it attached. `test_put_does_not_notify_other_kind_listeners` pinned the old behaviour and is replaced by the raise assertion. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills.py | 17 +++++++++++--- .../src/launchdarkly_ai_server/skills_fdv2.py | 14 ++++++++++++ packages/client/tests/test_skills.py | 22 +++++++++++-------- packages/client/tests/test_skills_fdv2.py | 13 +++++++++++ 4 files changed, 54 insertions(+), 12 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills.py b/packages/client/src/launchdarkly_ai_server/skills.py index c29bafc..d83dc2b 100644 --- a/packages/client/src/launchdarkly_ai_server/skills.py +++ b/packages/client/src/launchdarkly_ai_server/skills.py @@ -153,10 +153,21 @@ def add_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: """ Registers *fn* to be called with each raw object ``put`` under *kind*. - Only ``kind == SKILL_OBJECT_KIND`` is ever notified, because ``put`` only - accepts skill objects; a listener registered under any other kind is - recorded and never fires. + Only ``kind == SKILL_OBJECT_KIND`` is ever notified, because ``put`` + only accepts skill objects — so a registration for any other kind + **raises** rather than being recorded and silently never firing. This is + the reason ``watch_skills`` refuses a store with no ``add_listener`` at + all: a listener that never fires looks exactly like one whose objects + never changed, and a store that accepted the registration has promised + something it cannot keep. ``FDv2SkillStore.add_listener`` refuses the + same way. """ + if kind != SKILL_OBJECT_KIND: + raise ValueError( + f"InMemorySkillStore notifies only {SKILL_OBJECT_KIND!r} " + f"changes, so a listener on {kind!r} would never fire. Register " + f"it on {SKILL_OBJECT_KIND!r}." + ) self._listeners.setdefault(kind, []).append(fn) def remove_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index 24d5739..f62db28 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -1743,7 +1743,21 @@ def add_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: *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. + + Only ``SKILL_OBJECT_KIND`` is ever delivered, so a registration for any + other kind **raises** rather than being recorded and silently never + firing. ``InMemorySkillStore.add_listener`` refuses the same way, for + the reason ``watch_skills`` refuses a store with no ``add_listener`` at + all: a listener that never fires looks exactly like one whose objects + never changed, and a store that accepted the registration has promised + something it cannot keep. """ + if kind != SKILL_OBJECT_KIND: + raise ValueError( + f"FDv2SkillStore notifies only {SKILL_OBJECT_KIND!r} changes, so " + f"a listener on {kind!r} would never fire. Register it on " + f"{SKILL_OBJECT_KIND!r}." + ) with self._lock: self._listeners.setdefault(kind, []).append(fn) diff --git a/packages/client/tests/test_skills.py b/packages/client/tests/test_skills.py index 1b8ebd5..9deefd9 100644 --- a/packages/client/tests/test_skills.py +++ b/packages/client/tests/test_skills.py @@ -555,16 +555,20 @@ def test_put_notifies_skill_kind_listeners(self, make_raw_skill: Any) -> None: assert seen == [raw] - def test_put_does_not_notify_other_kind_listeners( - self, make_raw_skill: Any - ) -> None: + def test_add_listener_for_a_non_skill_kind_raises(self) -> None: + """A listener that can never fire is refused, not recorded. + + ``put`` only accepts skill objects, so nothing else is ever delivered: + a recorded listener on another kind would silently never fire, and that + is indistinguishable from one whose objects never changed. It is the + same failure §3.26 refuses when the store has no ``add_listener`` at + all, so it gets the same loud answer. + """ s = InMemorySkillStore() - seen: list[dict[str, Any]] = [] - s.add_listener("flag", seen.append) - - s.put(make_raw_skill(key="a")) - - assert seen == [] + with pytest.raises(ValueError, match="would never fire") as excinfo: + s.add_listener("flag", print) + assert "skill" in str(excinfo.value) + assert s._listeners == {} def test_remove_listener_stops_put_notifying_it(self, make_raw_skill: Any) -> None: s = InMemorySkillStore() diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 577bf87..878256d 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -2868,6 +2868,19 @@ class TestListenerRegistration: def _skill_listeners(store: Any) -> list[Any]: return list(store._listeners.get(SKILL_OBJECT_KIND, [])) + def test_fdv2_add_listener_for_a_non_skill_kind_raises(self, endpoint: Any) -> None: + """The transport refuses the same registration the in-memory store does. + + Only skill objects are ever delivered here, so a listener on any other + kind would never fire — and a store that accepted it has promised + something it cannot keep. + """ + with poll_store(endpoint) as store: + with pytest.raises(ValueError, match="would never fire") as excinfo: + store.add_listener("flag", print) + assert SKILL_OBJECT_KIND in str(excinfo.value) + assert store._listeners == {} + def test_fdv2_remove_listener_of_an_unregistered_callable_is_a_no_op( self, endpoint: Any ) -> None: From daeedec32370d904af1452ae33c8fd6ec7bf6dda Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 09:59:52 -0400 Subject: [PATCH 05/10] fix(client): reject a non-finite watch_skills debounce, and test the window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **`debounce=float("nan")` defeated the guard** (TESTING.md §3.26). `nan < 0` is `False`, so it passed `if debounce < 0` and then collapsed the coalescing window to nothing: `threading.Event().wait(nan)` returns immediately, so every delivered object reconciled on its own with no coalescing at all — the opposite of what the option is for. Now `math.isfinite` is checked alongside the sign, matching how `write_skills` already guards its `timeout`. `inf` is refused for the same reason it is refused there: a window nothing ever leaves. Also fills the §3.26 test gaps, which had no coverage either way: negative and non-finite debounce, `on_reconcile` firing for subsequent reports and not the initial one, a reconcile that raises not killing the watcher, `prune` / `on_unavailable` pass-through, and an invalid root raising out of `watch_skills` rather than into a worker thread's log. **The coalescing test was one of the vacuous shapes §3.26 warns about** — twelve puts inside a single synchronous listener pass, asserted with an upper bound (`reconciles <= 2`). A burst that arrives before the worker has woken collapses to one reconcile whether or not the debounce exists, so that shape passes against an implementation with the debouncing deleted. Replaced with six puts spread across one window, an exact figure, and an assertion that the counter moved at all; verified by mutation — removing the debounce sleep turns it red. The exact figure is two, not one, and that is a real deviation from §3.26's "exactly one reconcile per debounce window": this watcher clears its wake flag *before* the window rather than after, so a change arriving inside the window schedules a further pass instead of merging into the reconcile that is about to run. TypeScript restarts its timer on each notification and so does merge them. The behaviour is deliberate and documented in `_run` (a redundant reconcile converges; a missed one does not) and it avoids the starvation a restart-the-timer debounce admits, so it is left alone here and raised separately rather than changed under a test-coverage commit. Co-Authored-By: Claude Opus 5 --- .../launchdarkly_ai_server/skills_watch.py | 17 +- packages/client/tests/test_skills_watch.py | 177 +++++++++++++++++- 2 files changed, 185 insertions(+), 9 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_watch.py b/packages/client/src/launchdarkly_ai_server/skills_watch.py index 72cad73..5ee8649 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_watch.py +++ b/packages/client/src/launchdarkly_ai_server/skills_watch.py @@ -24,6 +24,7 @@ import asyncio import logging +import math import os import threading from collections.abc import Callable, Sequence @@ -267,6 +268,10 @@ async def watch_skills( never changed. The optional ``remove_listener`` lets ``SkillWatcher.close`` detach from the store; a store without it still works, but each closed watcher then stays registered for the store's lifetime. + + *debounce* must be a non-negative finite number of seconds. ``NaN`` raises + alongside a negative value: it would pass a bare ``< 0`` guard and then + collapse the coalescing window to nothing. """ store = get_store() if store is None: @@ -282,8 +287,16 @@ async def watch_skills( "observed. Use write_skills for a one-shot reconcile, or configure a " "store with a delivery transport (FDv2SkillStore)." ) - if debounce < 0: - raise ValueError(f"debounce must not be negative, got {debounce!r}") + # ``NaN`` is the case a bare ``< 0`` guard misses: ``nan < 0`` is ``False``, + # so it passes validation and then collapses the window to nothing, because + # ``Event.wait(nan)`` returns immediately. Every delivered object would + # reconcile on its own with no coalescing at all — the opposite of what the + # option is for. ``write_skills`` already guards its ``timeout`` this way. + if not math.isfinite(debounce) or debounce < 0: + raise ValueError( + f"debounce must be a non-negative finite number of seconds, got " + f"{debounce!r}" + ) # The watcher attaches its listener before the initial reconcile, not after. # The reconcile snapshots the store as its first step and then spends the diff --git a/packages/client/tests/test_skills_watch.py b/packages/client/tests/test_skills_watch.py index 4c30f88..5c60d06 100644 --- a/packages/client/tests/test_skills_watch.py +++ b/packages/client/tests/test_skills_watch.py @@ -60,22 +60,185 @@ async def test_the_in_memory_store_can_also_drive_a_watch( finally: watcher.close() - async def test_a_burst_of_changes_coalesces_into_few_reconciles( + async def test_a_burst_of_changes_coalesces_into_a_debounce_window( self, tmp_path: Any, make_raw_skill: Any ) -> None: + """Six notifications spread across one window produce two reconciles. + + Three things are needed for this to discriminate rather than pass + vacuously, and this test does all three: + + * The puts happen **after** ``watch_skills`` has attached its listener. + A payload committed before that reaches nobody, the counter stays at + zero, and an upper bound then passes against an implementation with + the debouncing deleted. + * The counter is asserted to have **moved**, and against an exact + figure. A run in which nothing was ever notified is not a test of + coalescing. + * The notifications are **spread over time**. A burst arriving inside + one synchronous pass of the listener collapses to a single reconcile + whether or not the debounce exists, because the worker had not woken + yet — so a payload of twelve objects put back to back proves nothing. + Each put here lands in its own scheduler pass. + + **Two, not one.** This watcher clears its wake flag before the window + rather than after, so a change arriving *inside* the window schedules a + further pass instead of being merged into the reconcile that is about to + run: six puts across one window are one reconcile for the window plus + one for the spill. Six separate reconciles is what the debounce + prevents, and is what this figure discriminates against. + """ store = InMemorySkillStore() await init_client(options={"skillStore": store}, client=object()) - _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.1) + _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.6) try: - for i in range(12): + assert watcher.reconciles == 0 + for i in range(6): store.put(make_raw_skill(key=f"skill-{i}")) - time.sleep(0.5) - # Twelve objects put back to back fire twelve listener calls; without - # coalescing that is twelve reconciles of one root. - assert watcher.reconciles <= 2 + time.sleep(0.1) + assert wait_until(lambda: watcher.reconciles >= 2, timeout=10) + # Settle well past two windows, so a third would have landed by now. + time.sleep(1.5) + assert watcher.reconciles == 2 finally: watcher.close() + async def test_changes_spread_beyond_the_window_do_not_coalesce( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + """The control for the test above: the window is what merges them. + + Without it the two tests could both be satisfied by a watcher that + reconciles once and then never again. + """ + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.05) + try: + store.put(make_raw_skill(key="one")) + assert wait_until(lambda: watcher.reconciles >= 1, timeout=10) + store.put(make_raw_skill(key="two")) + assert wait_until(lambda: watcher.reconciles >= 2, timeout=10) + finally: + watcher.close() + + @pytest.mark.parametrize( + "debounce", [-1.0, -0.001, float("nan"), float("inf"), float("-inf")] + ) + async def test_a_negative_or_non_finite_debounce_raises( + self, tmp_path: Any, debounce: float + ) -> None: + """``NaN`` is the case a bare ``< 0`` guard misses. + + ``nan < 0`` is ``False``, so it would pass validation and then collapse + the window to nothing — ``Event.wait(nan)`` returns immediately, so + every delivered object reconciles on its own with no coalescing at all. + ``write_skills`` already guards its ``timeout`` this way. + """ + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + with pytest.raises(ValueError, match="debounce"): + await watch_skills("*", tmp_path / "s", debounce=debounce) + # Refused before anything was attached, so there is no watcher to close. + assert store._listeners.get(SKILL_OBJECT_KIND, []) == [] + + async def test_on_reconcile_receives_each_subsequent_report( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + """Not the initial report: that one is returned to the caller directly.""" + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="body")) + await init_client(options={"skillStore": store}, client=object()) + seen: list[Any] = [] + report, watcher = await watch_skills( + "*", tmp_path / "s", debounce=0.05, on_reconcile=seen.append + ) + try: + assert seen == [] + store.put(make_raw_skill(key="a", version=2, content="new body")) + assert wait_until(lambda: len(seen) >= 1, timeout=10) + assert seen[0] is not report + assert seen[0].ok is True + assert [a.key for a in seen[0].actions] == ["a"] + finally: + watcher.close() + + async def test_a_reconcile_that_raises_does_not_kill_the_watcher( + self, tmp_path: Any, make_raw_skill: Any, caplog: Any + ) -> None: + """A watcher that died on one bad run would silently stop pruning.""" + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + calls: list[int] = [] + + def explodes_once(_report: Any) -> None: + calls.append(1) + if len(calls) == 1: + raise RuntimeError("boom") + + _report, watcher = await watch_skills( + "*", tmp_path / "s", debounce=0.05, on_reconcile=explodes_once + ) + try: + with caplog.at_level("ERROR"): + store.put(make_raw_skill(key="a", version=1, content="one")) + assert wait_until(lambda: len(calls) >= 1, timeout=10) + store.put(make_raw_skill(key="a", version=2, content="two")) + assert wait_until(lambda: len(calls) >= 2, timeout=10) + written = tmp_path / "s" / "a" / "SKILL.md" + assert wait_until(lambda: written.read_text() == "two", timeout=10) + finally: + watcher.close() + + async def test_every_write_skills_option_is_passed_straight_through( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + """``prune``, ``timeout`` and ``on_unavailable`` are not reinterpreted. + + Driven through one of ``write_skills``' own cases: a store that cannot + answer, with ``on_unavailable="raise"``, raises out of the initial + reconcile exactly as it would out of ``write_skills``. + """ + + class Unavailable(InMemorySkillStore): + def all_objects(self, _kind: str) -> dict[str, Any]: + raise RuntimeError("store is down") + + store = Unavailable() + await init_client(options={"skillStore": store}, client=object()) + with pytest.raises(RuntimeError, match="store is down"): + await watch_skills( + "*", tmp_path / "s", on_unavailable="raise", debounce=0.05 + ) + assert store._listeners.get(SKILL_OBJECT_KIND, []) == [] + + async def test_prune_is_passed_through_and_can_be_turned_off( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="body")) + await init_client(options={"skillStore": store}, client=object()) + report, watcher = await watch_skills( + "*", tmp_path / "s", prune=False, debounce=0.05 + ) + watcher.close() + written = tmp_path / "s" / "a" / "SKILL.md" + assert written.read_text() == "body" + assert report.ok is True + assert all(a.action != "removed" for a in report.actions) + + async def test_an_invalid_root_raises_out_of_watch_skills( + self, tmp_path: Any + ) -> None: + """The initial reconcile runs on the caller's thread, so a bad root is + the caller's exception rather than a line in a worker thread's log.""" + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + not_a_directory = tmp_path / "file" + not_a_directory.write_text("") + with pytest.raises(ValueError, match="not a directory"): + await watch_skills("*", not_a_directory, debounce=0.05) + async def test_a_store_with_no_listener_support_is_refused_loudly( self, tmp_path: Any ) -> None: From 4a204f8ea933ba906c9e7b2eea7c508d41f14b5c Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 10:00:36 -0400 Subject: [PATCH 06/10] fix(client): fail the config parse on an explicit skills: null MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `parse_ai_config` gated on `if skills is not None`, so `skills: null` was accepted and read as absent. TESTING.md §3.5 now calls this case out specifically, and the reason is the reason it is worth a commit of its own: read as absent, it makes `skill_refs` return `[]`, and a `prune=True` reconcile then **deletes** previously-materialized skill files on the strength of a field the SDK could not parse. §3.21 establishes exactly this principle on the store path — reading a malformed thing as absent "would let a tampered object look like a deleted one … and would let prune delete the last known-good copy on disk" — and it applies identically here. Failing the whole parse is the louder and safer outcome, and it is what TypeScript already does. Gated on key presence instead, so `_parse_skills` sees the `None` and rejects it as a non-array, with `skills` in the message. There was no Python test either way; both halves are now pinned, including that an *absent* `skills` key is still valid, since rejecting `null` must not cost backward compatibility for configs that simply have no `skills`. Co-Authored-By: Claude Opus 5 --- .../types_validation.py | 12 ++++++--- packages/client/tests/test_schema.py | 27 +++++++++++++++++++ 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/types_validation.py b/packages/client/src/launchdarkly_ai_server/types_validation.py index 37cd56b..77ad601 100644 --- a/packages/client/src/launchdarkly_ai_server/types_validation.py +++ b/packages/client/src/launchdarkly_ai_server/types_validation.py @@ -156,9 +156,15 @@ def parse_ai_config(raw: Any) -> ParseResult: error={"message": "outputFormat must be an object (JSON Schema)"}, ) - skills = raw.get("skills") - if skills is not None: - err = _parse_skills(skills) + # ``in`` rather than ``is not None``: an explicit ``skills: null`` must fail + # the parse, not read as absent. Read as absent it makes ``skill_refs`` + # return ``[]``, and a ``prune=True`` reconcile then *deletes* previously + # materialized skill files on the strength of a field the SDK could not + # parse — the hazard the store path already refuses, where reading a + # malformed object as absent would let prune delete the last known-good + # copy on disk. Failing the whole parse is the louder and safer outcome. + if "skills" in raw: + err = _parse_skills(raw["skills"]) if err: return ParseFailure(success=False, error={"message": err}) diff --git a/packages/client/tests/test_schema.py b/packages/client/tests/test_schema.py index 3be146d..359a890 100644 --- a/packages/client/tests/test_schema.py +++ b/packages/client/tests/test_schema.py @@ -124,6 +124,33 @@ def test_key_at_length_bound_accepted(self) -> None: def test_non_array_skills_fails(self, bad_skills: Any) -> None: assert parse_ai_config(self._base(skills=bad_skills)).success is False + def test_an_explicit_null_skills_fails_rather_than_reading_as_absent( + self, + ) -> None: + """``skills: null`` is the non-array that reads as "no skills". + + It must not be treated that way. Read as absent it makes ``skill_refs`` + return ``[]``, and a ``prune=True`` reconcile then *deletes* previously + materialized skill files on the strength of a field the SDK could not + parse — the same hazard the store path refuses, where reading a + malformed object as absent would let prune delete the last known-good + copy on disk. Failing the whole parse is the louder and safer outcome, + and it is why this case is pinned apart from the other non-arrays. + """ + result = parse_ai_config(self._base(skills=None)) + assert result.success is False + assert "skills" in result.error["message"] + + def test_an_absent_skills_key_is_still_valid(self) -> None: + """The other half of the bullet above: *absent* is not *null*. + + Rejecting ``null`` must not cost backward compatibility for the + configs that simply have no ``skills`` field. + """ + raw = self._base() + assert "skills" not in raw + assert parse_ai_config(raw).success is True + @pytest.mark.parametrize("entry", ["pdf-extraction", 1, None, ["a", 1]]) def test_non_object_entry_fails(self, entry: Any) -> None: assert parse_ai_config(self._base(skills=[entry])).success is False From 030e8c33458ab43e89dd60301d5111fd2042b354 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 10:00:50 -0400 Subject: [PATCH 07/10] fix(client): check content size before encoding it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `verified_bytes` encoded before checking the size cap, so content that is both over the cap and carries a lone surrogate reported `not_utf8`. TESTING.md §3.21 fixes the order as shape, then **size**, then **encoding**, then hash, and that input is the one boundary where the order is observable: without it, the code is whichever check the implementation happens to reach first, and §3.21's "one code per failure class" rule cannot be tested against it. Size first is also the cheap check — running an encoding pass over a 10 MiB body before rejecting it for being 10 MiB is a DoS foothold rather than a nicety. Python cannot reorder these literally, because the size is a count of encoded bytes and `str.encode("utf-8")` raises on an unpaired surrogate rather than substituting. So the unencodable case now encodes a second time with `errors="replace"`, purely to measure, and reports `over_size_cap` or `not_utf8` in that order. Those replacement bytes are never hashed and never returned — the `not_utf8` branch returns before the hash comparison — which is what makes `errors="replace"` safe here where `errors="surrogatepass"` is not safe anywhere: fabricated bytes that reached the comparison could satisfy it. TypeScript arrives at the same order for free, since `TextEncoder` substitutes U+FFFD silently and its guard is an explicit round-trip check (A.12). Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_core.py | 45 ++++++++++++++----- packages/client/tests/test_skills.py | 25 +++++++++++ 2 files changed, 58 insertions(+), 12 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_core.py b/packages/client/src/launchdarkly_ai_server/skills_core.py index 8da0ba8..3cc1751 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_core.py +++ b/packages/client/src/launchdarkly_ai_server/skills_core.py @@ -421,7 +421,15 @@ def verified_bytes( key: str, content: str | bytes, expected_hash: str, version: int ) -> VerifiedContent | VerificationFailure: """ - The whole content half of integrity verification: encode, size, hash. + The whole content half of integrity verification: size, encoding, hash. + + The order is fixed, so content failing two classes reports one determined + code: **size** before **encoding** before **hash**. Size first is what makes + over-cap content carrying a lone surrogate report ``over_size_cap`` rather + than ``not_utf8`` — without a fixed order that input's code is whichever + check the implementation happens to reach first, and §3.21's "one code per + failure class" rule cannot be tested against it. The shape checks that + precede all three are in ``verify_raw_skill``. Accepts either shape content arrives in. A wire-shaped ``str`` is UTF-8 encoded here, once — the only place that encode happens. ``bytes`` is an @@ -439,6 +447,7 @@ def verified_bytes( first one's verdict forward: that puts a "trust the value computed upstream" branch inside the one function whose job is not to. """ + encodable = True if isinstance(content, bytes): encoded = content else: @@ -447,17 +456,18 @@ def verified_bytes( except UnicodeEncodeError: # json.loads turns a "\ud800" escape into an unpaired surrogate, # which has no UTF-8 encoding — so there are no bytes the server - # could have hashed. Never reach for errors="surrogatepass": it - # would fabricate bytes that could satisfy the hash comparison. - reason = "content is not encodable as UTF-8" - record_integrity_failure( - key, - reason, - reason_code="not_utf8", - version=version, - expected_hash=expected_hash, - ) - return VerificationFailure(reason) + # could have hashed. + # + # These replacement bytes exist only to measure the content against + # the size cap, so that the *reported* failure follows the fixed + # order above. They are never hashed and never returned: the + # ``not_utf8`` branch below returns before the hash comparison, and + # nothing else reads ``encoded`` on this path. That is the whole + # reason ``errors="replace"`` is safe here and + # ``errors="surrogatepass"`` would not be anywhere — fabricated + # bytes that reached the comparison could satisfy it. + encodable = False + encoded = content.encode("utf-8", errors="replace") if len(encoded) > MAX_SKILL_CONTENT_BYTES: reason = ( @@ -473,6 +483,17 @@ def verified_bytes( ) return VerificationFailure(reason) + if not encodable: + reason = "content is not encodable as UTF-8" + record_integrity_failure( + key, + reason, + reason_code="not_utf8", + version=version, + expected_hash=expected_hash, + ) + return VerificationFailure(reason) + # sha256, lowercase hex, over the verbatim bytes — no canonicalization and # no content parsing of any kind anywhere in the integrity path. observed_hash = hashlib.sha256(encoded).hexdigest() diff --git a/packages/client/tests/test_skills.py b/packages/client/tests/test_skills.py index 9deefd9..15a885a 100644 --- a/packages/client/tests/test_skills.py +++ b/packages/client/tests/test_skills.py @@ -1731,6 +1731,31 @@ async def test_one_record_per_reason_code( # a well-formed key and digest never enter a redaction branch. assert "Do the thing." not in json.dumps(record) + async def test_over_cap_content_with_a_lone_surrogate_reports_the_size( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """The one input on which the check order is observable. + + Content that is both over the cap and unencodable fails two classes, so + without a fixed order its code is whichever check the implementation + reaches first — and the "one code per failure class" rule above cannot + be tested against it. The order is shape, then **size**, then + **encoding**, then hash: size is the cheap check, and running an + encoding pass over a 10 MiB body before rejecting it for being 10 MiB + is a DoS foothold rather than a nicety. + """ + from launchdarkly_ai_server import skills_core + + body = _OVERSIZE + json.loads(r'"\ud800"') + await self._withhold({"a": _raw_object(content=body)}) + + records = _integrity_records(caplog) + assert len(records) == 1 + assert records[0]["reason_code"] == "over_size_cap" + # The cap is named, interpolated from the constant rather than restated, + # so raising it cannot leave a stale figure in the message. + assert str(skills_core.MAX_SKILL_CONTENT_BYTES) in records[0]["reason"] + def test_the_case_table_exhausts_the_vocabulary(self) -> None: """The vocabulary is closed, and every token in it is reachable. From 4092de71f56f1c14a55117991ac0a9733f47c425 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 10:02:17 -0400 Subject: [PATCH 08/10] docs(client): correct the stale cross-language parity claim, and the README MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `agents.md`'s POSIX-only justification claimed the TypeScript SDK "could not match them at all — Node exposes no `*at()` family on *any* platform, so its racy floor is universal rather than Windows-only." The first half is still true; the second is not. TypeScript commit `0a15b10` added a `SUPPORTS_PROC_FD` probe and `/proc/self/fd//` addressing, which the Linux kernel resolves from the inode a descriptor holds rather than from the name it was opened under — closing the swap window there exactly as `*at()` does here. TypeScript's `lstat` floor now applies on macOS and Windows only, the same shape as this side's. The passage is load-bearing — it is why the Windows reparse-point checks are not implemented — so the rewrite says explicitly that the decision is unchanged and why: it never rested on TypeScript being equally exposed, it rests on there being no Windows CI runner to verify the checks against, which is still true of both repositories. README, for the behaviour changes in this branch: the two transport hosts and the `stream_uri` option, `close()` being final, `watch_skills`' `debounce` unit and bound and its `on_reconcile` contract, `add_listener` refusing a kind it will never deliver, and a non-array `skills` field failing the config parse rather than reading as absent. Co-Authored-By: Claude Opus 5 --- packages/client/README.md | 27 +++++++++++++++++++-------- packages/client/agents.md | 24 +++++++++++++++++++----- 2 files changed, 38 insertions(+), 13 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index ca834fe..75311a2 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -538,10 +538,21 @@ objects through the `SkillStore` interface and cannot tell which store produced customer-confidential. A mobile key (`mob-…`) or a client-side environment ID raises from the constructor. -**The SDK key goes only where you pointed it.** `base_uri` must be `https://` (plain `http://` -is refused, except to a loopback host for a local test double), and redirects are never -followed, so a 3xx from a proxy or a misconfigured private instance stops delivery rather than -forwarding the key to whatever host the `Location` header names. +**The SDK key goes only where you pointed it.** `base_uri` and `stream_uri` must each be +`https://` (plain `http://` is refused, except to a loopback host for a local test double), +and redirects are never followed, so a 3xx from a proxy or a misconfigured private instance +stops delivery rather than forwarding the key to whatever host the `Location` header names. + +**Polling and streaming have separate hosts.** LaunchDarkly serves `/sdk/poll` from +`https://sdk.launchdarkly.com` and `/sdk/stream` from `https://stream.launchdarkly.com`, so +the defaults are a pair. Pass `base_uri` on its own and it applies to both — what a relay or a +private instance serving both endpoints from one host needs — or pass `stream_uri` as well to +override them independently. + +**`close()` is final.** A closed store still answers from the content it received, but +delivery cannot be resumed: `start()` afterwards raises. That is what gives `close()` a +postcondition you can rely on — delivery has stopped — even when its join times out. +Construct a new store to resume. **Reads are memory-bounded.** No poll body or streamed event is held past `MAX_RESPONSE_BYTES` (64 MiB, far above any real payload); one that crosses it is dropped without being applied, the @@ -582,16 +593,16 @@ Windows. | Export | Description | |---|---| -| `skill_refs(config)` | Project a config's `skills` array into `list[SkillReference]`. Pure — no client, store, or network needed. Returns `[]` when absent. | +| `skill_refs(config)` | Project a config's `skills` array into `list[SkillReference]`. Pure — no client, store, or network needed. Returns `[]` when the field is absent. A `skills` field that is present but not an array — including an explicit `null` — fails the config parse instead, so a field the SDK could not read never reaches a pruning reconcile as "no skills". | | `get_skill(key, *, version=None)` | One verified skill, or `None`. `version=None` means newest available; a specific `version` matches exactly. Raises only when no store is configured. | | `get_skill_result(key, *, version=None)` | The same retrieval, reporting **why**: a frozen `SkillOutcome` with `.skill`, `.reason` (`ok` / `absent` / `integrity_failure` / `store_unavailable` / `wrong_version`), and `.detail`. Use it to fail closed on tampering — see *Failing closed on tampering* above. Raises only when no store is configured. | | `get_skills(refs)` | Batch form. Accepts `SkillReference` values and bare key strings (string = latest). Results follow input order; missing or unverifiable entries are omitted. | | `all_skills()` | Every verified skill the store holds, one per key at its newest version. | | `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 `is_initialized()`, `add_listener(kind, fn)` / `remove_listener(kind, fn)`. A store without `is_initialized()` is treated as initialized. | +| `SkillStore` | The structural interface content arrives through: `get_object(kind, key, version=None)`, `all_objects(kind)`, optional `is_initialized()`, `add_listener(kind, fn)` / `remove_listener(kind, fn)`. A store without `is_initialized()` is treated as initialized. Both shipped stores deliver only the skill kind, so `add_listener` on any other kind raises rather than being recorded and silently never firing. | | `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)`, `is_initialized()`, `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. | +| `FDv2SkillStore(sdk_key, *, base_uri=…, stream_uri=…, mode="stream", …)` | The delivery transport: a store fed by LaunchDarkly over the SDK-facing FDv2 channel. `start()`, `wait_for_skills(timeout)`, `is_initialized()`, `close()`, `diagnostics`, `failed`; also a context manager. `base_uri` and `stream_uri` are separate hosts, defaulting to LaunchDarkly's polling and streaming origins; `base_uri` alone covers both. `close()` is **final** — `start()` afterwards raises. **Server-side only** — a mobile key or client-side environment ID raises. See *Receiving skills from LaunchDarkly* above. | +| `watch_skills(skills, root, *, debounce=0.5, on_reconcile=None, …)` | `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. `debounce` is in **seconds** and must be non-negative and finite; `on_reconcile` is called with each *subsequent* report, the initial one being returned directly. One watcher per root. | | `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, diff --git a/packages/client/agents.md b/packages/client/agents.md index 5533afb..d995575 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -673,11 +673,25 @@ it as soon as the write returns. **The platform bound is POSIX-only, and that is a decision — do not quietly "fix" it.** Windows reparse-point checks (`GetFileAttributesW`, `FILE_FLAG_OPEN_REPARSE_POINT`) are not implemented because Windows is not a supported or tested platform for this release: there is -no Windows CI runner in either repository, so the checks would ship unverified, and the -TypeScript SDK could not match them at all — Node exposes no `*at()` family on *any* -platform, so its racy floor is universal rather than Windows-only. Implementing them in -Python alone would break cross-language parity and trade a documented bound for an unverified -one. Two follow-on facts: on Windows write permission on the managed root is the only +no Windows CI runner in either repository, so the checks would ship unverified, and there is +no second implementation to check them against — the TypeScript SDK has no Windows story +either. Implementing them in Python alone would trade a documented bound for an unverified +one. + +The parity argument used to be stronger than that, and the correction matters because the +old wording is now wrong. It read: Node exposes no `*at()` family on *any* platform, so its +racy floor is universal rather than Windows-only. The first half is still true and the +second is not. `*at()` is not the only way to address a child relative to a pinned inode: +TypeScript commit `0a15b10` added a `SUPPORTS_PROC_FD` probe and `/proc/self/fd//` +addressing, which the Linux kernel resolves from the inode the descriptor holds rather than +from the name it was opened under. That **closes** the swap window on Linux exactly as +`*at()` does here, so TypeScript's `lstat` floor now applies on macOS and Windows only — +the same shape as this side's, not a universal one. + +None of which reopens the decision above. It never rested on TypeScript being equally +exposed; it rests on there being no Windows CI runner to verify the checks against, which is +still the case in both repositories. Two follow-on facts: on Windows write permission on the +managed root is the only boundary, which is why the privilege-separated deployment is documented as the mitigation rather than as advice; and this bound retroactively lowers the priority of the reserved-device-name work above — keep that code, but do not read it as evidence that Windows is hardened. If From 204e287a0c4a3fd50b075c5413ec545279d94018 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 13:06:52 -0400 Subject: [PATCH 09/10] Simplify comment --- .../client/src/launchdarkly_ai_server/skills_fdv2.py | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index f62db28..aeb6241 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -93,13 +93,8 @@ """ Where ``GET /sdk/stream`` is served. -LaunchDarkly serves streaming from a **different host** than polling, which is -why this is a second default rather than a path under ``DEFAULT_BASE_URI``. Both -base server-side SDKs ship the pair: ``ldclient.config.Config`` defaults -``stream_uri`` to ``https://stream.launchdarkly.com`` alongside its own polling -host, and ``@launchdarkly/js-server-sdk-common`` does the same. ``mode="stream"`` -is this store's default, so a single-host default would have the default -configuration connect to the wrong host on first contact with a real environment. +LaunchDarkly serves streaming from a different host than polling, which is +why this is a second default rather than a path under ``DEFAULT_BASE_URI``. A *base_uri* given on its own applies to both endpoints, because a relay or a private instance serving both from one host should need only one option; see From c2d482a0a09217f06266ddc37f341e264985b7ec Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 13:36:30 -0400 Subject: [PATCH 10/10] fix(client): keep the stale-selector repair out of the retry budget An HTTP 400 for a request carrying a `basis` selector or an etag means that state may be one the server no longer accepts, so the transport drops it and asks for a full transfer once before treating the status as fatal. That one request was not guaranteed to go out: it was prepared and then measured against the consecutive-failure bound like any other retry, so a 400 arriving on a budget an outage had already spent gave up while holding the only request known to repair it, and delivery stopped for the process lifetime over state the store had just dropped. The repair is now exempt from the bound. It cannot unbound the loop: the repaired request carries no state, so a second 400 is fatal on its own, and any other failure after it meets a budget still over the bound. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 14 +++++- packages/client/tests/test_skills_fdv2.py | 43 +++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index aeb6241..d04cb89 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -1468,7 +1468,9 @@ def __init__( *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. + payload resets the count. The one request built from nothing after a + stale selector is refused is exempt, so an outage that has already + spent the budget cannot swallow the one repair available. """ _require_server_side_credential(sdk_key) # A lone ``base_uri`` means "both endpoints are here"; the two-host @@ -1811,6 +1813,7 @@ def _run(self) -> None: # would spend a retry from the bounded budget and leave a # misleading ``last_error`` on a healthy store. return + repairing_state = False if isinstance(exc, _StaleRequestStateError): # The selector and the etag are the only client state in the # request, so a rejection of a request carrying neither is @@ -1822,6 +1825,7 @@ def _run(self) -> None: if not exhausted: self._basis = None self._etag = None + repairing_state = True if exhausted: self._give_up(str(exc)) return @@ -1834,7 +1838,13 @@ def _run(self) -> None: answered = self._attempt_answered self._reader.diagnostics.connection_failures = failures self._reader.diagnostics.last_error = str(exc) - if failures > self._max_consecutive_failures: + if failures > self._max_consecutive_failures and not repairing_state: + # The one-shot request built from nothing is exempt from the + # bound, so an outage that has already spent the budget + # cannot swallow the repair a stale selector is asking for. + # It cannot unbound the loop either: the repaired request + # carries no state, so a second 400 is fatal on its own and + # any other failure meets a budget still over the bound. self._give_up( f"gave up after {failures} consecutive failures; " f"last error: {exc}" diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 878256d..5ec48bf 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -1693,6 +1693,49 @@ def test_a_400_reconnects_once_from_scratch_and_is_then_fatal( # Last known good survives both. assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + def test_a_400_still_reconnects_from_scratch_on_a_spent_budget( + self, endpoint: Any + ) -> None: + """The one repair available does not compete with the retry bound. + + A 400 arriving on a budget an outage has already spent would otherwise + give up while holding the one request known to fix it, and delivery + would stop for the process lifetime over state the store was about to + drop. Exempting that request cannot unbound the loop: it carries no + state, so a second 400 is fatal on its own. + """ + store = poll_store(endpoint, max_consecutive_failures=1) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=500) + endpoint.queue_poll(status=400) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with store: + assert store.wait_for_skills(timeout=5) is True + # The premise: the budget is spent by the time the 400 arrives. + assert wait_until(lambda: len(endpoint.requests) == 4) + assert store.failed is None + # The repair went out from scratch rather than never going out at all. + repair = endpoint.requests[3] + assert repair["query"] == {} + assert repair["if_none_match"] is None + + def test_a_non_400_after_the_repair_meets_the_spent_budget( + self, endpoint: Any + ) -> None: + """The exemption is for the repair, not for the run that follows it.""" + store = poll_store(endpoint, max_consecutive_failures=1) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=500) + endpoint.queue_poll(status=400) + endpoint.queue_poll(status=500) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with store: + assert store.wait_for_skills(timeout=5) is True + assert wait_until(lambda: store.failed is not None) + assert "gave up after 3 consecutive failures" in store.failed + # The fifth queued payload is never asked for. + assert len(endpoint.requests) == 4 + def test_a_400_carrying_no_client_state_is_fatal_at_once( self, endpoint: Any ) -> None: