From a6c2af0d10dd8d62bb0fea055e0d4a4e96d36882 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 11:18:59 -0400 Subject: [PATCH 1/5] fix(client): publish the revocations an xfer-full states by omission MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_ProtocolReader` built a fresh empty `_pending` for `xfer-full`, so a skill the full transfer omitted was correctly dropped at `_payload_transferred` — but `_changes` only ever accumulated from `_put_object` and `_delete_object`, so nothing was published for the omitted skill. A full transfer carrying other puts still wakes a watcher through those puts, so the observable gap was the transfer carrying no puts at all: the environment's last skill revoked. The store emptied while `changes` stayed empty and `objects_revoked` stayed zero, `watch_skills` never reconciled, and the skill's `SKILL.md` survived on disk until some unrelated change — in exactly the case pruning exists for. Port TypeScript's `revocationsBetween` / `keysFullyRevoked`: diff the committed set against the incoming one before the swap, at `(key, version)` granularity so a key whose version moved yields both a put for the arrival and a tombstone for the departure, and count only keys that left the payload entirely toward `objects_revoked` — the same rule `_delete_object` follows in counting only a tombstone that took something away. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 69 +++++++++++++- packages/client/tests/test_skills_fdv2.py | 93 +++++++++++++++++++ 2 files changed, 161 insertions(+), 1 deletion(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index f62db28..7169d49 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -572,6 +572,57 @@ class _TransferOutcome: """ +def _identity_of(raw: dict[str, Any]) -> tuple[str, Any]: + """ + One object's ``(key, version)`` identity, as something comparable. + + An object with no usable version compares alike to any other of its key, + which is what holding it under its key alone already means. + """ + version = raw.get("version") + return (raw["key"], version if is_valid_skill_version(version) else None) + + +def _revocations_between( + current: _SkillObjectSet, pending: _SkillObjectSet +) -> list[dict[str, Any]]: + """ + Tombstones for every object *pending* no longer holds. + + A full transfer states the whole payload, so its revocations arrive as an + absence rather than as an event; this recovers them. At ``(key, version)`` + granularity to match ``delete-object``, so a key whose version moved yields + both a put for the arrival and a tombstone for the departure — what a + listener that reads versions needs, and harmless to one that only needs + "something changed". + """ + surviving = {_identity_of(raw) for raw in pending.all_raw()} + return [ + {"key": key, "version": version} + for key, version in (_identity_of(raw) for raw in current.all_raw()) + if (key, version) not in surviving + ] + + +def _keys_fully_revoked(revoked: list[dict[str, Any]], pending: _SkillObjectSet) -> int: + """ + How many of *revoked* are true revocations rather than version moves. + + Counted per key, not per tombstone: a key *pending* still holds under some + other version has moved, and only a key that left the payload entirely is + gone. That is what ``objects_revoked`` counts, the same rule + ``_delete_object`` applies when it counts only a tombstone that took + something away. ``changes`` carries every tombstone regardless. + """ + return len( + { + tombstone["key"] + for tombstone in revoked + if pending.get(tombstone["key"], None) is None + } + ) + + class _ProtocolReader: """ Applies FDv2 events to an object set. Pure — no sockets, no threads, no @@ -730,6 +781,20 @@ def _payload_transferred(self, data: Any) -> _TransferOutcome: self.diagnostics.payloads_ignored += 1 self._changes = [] elif self._pending is not None: + if self._intent == _INTENT_TRANSFER_FULL: + # A full transfer revokes by omission: whatever it did not carry + # is gone, and no ``delete-object`` ever says so. Diffed before + # the swap, so those departures reach listeners as tombstones + # like any other revocation — without which the one case pruning + # exists for, an environment's last skill being revoked, would + # empty the store and wake nobody. + revoked = _revocations_between(self._committed, self._pending) + self._changes.extend(revoked) + # Every departure is reported; only a key that left counts as + # revoked. + self.diagnostics.objects_revoked += _keys_fully_revoked( + revoked, self._pending + ) self._committed.replace_with(self._pending) _warn_if_nothing_can_verify(self._committed) if self._skills_in_payload and payload_id is not None: @@ -1738,7 +1803,9 @@ def add_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: 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. + reads content must check for ``content`` rather than assume it. Both + ways of stating a revocation arrive that way: a ``delete-object``, and a + full transfer that simply stopped carrying the object. *fn* runs on the delivery thread. Keep it cheap and non-blocking. An exception it raises is logged and swallowed, because a broken listener diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 878256d..93210a5 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -611,6 +611,72 @@ def test_a_full_transfer_replaces_rather_than_merges(self) -> None: assert held.get("first", None) is None assert held.get("second", None) is not None + def test_a_full_transfer_that_omits_a_skill_publishes_a_tombstone(self) -> None: + """ + A full transfer states the whole payload, so it revokes by omission and + no ``delete-object`` ever says so. Without the diff the store empties + while ``changes`` stays empty, and the case pruning exists for — the + environment's last skill revoked — wakes no listener at all. + """ + held = _SkillObjectSet() + reader = _ProtocolReader(held) + drive(reader, full_payload(("put-object", put_skill(object_version=3)))) + outcomes = drive(reader, full_payload(state="basis-2")) + assert len(held) == 0 + assert outcomes[-1].changes == [{"key": "pdf-extraction", "version": 3}] + assert reader.diagnostics.objects_revoked == 1 + + def test_an_omitted_version_less_object_is_reported_as_departed(self) -> None: + """ + An object too malformed to carry a version is held under its key alone, + and leaves the same way: as a tombstone with no version, which is what a + ``delete-object`` naming no version spells too. + """ + held = _SkillObjectSet() + reader = _ProtocolReader(held) + drive(reader, full_payload(("put-object", put_skill(object_version=None)))) + outcomes = drive(reader, full_payload(state="basis-2")) + assert outcomes[-1].changes == [{"key": "pdf-extraction", "version": None}] + assert reader.diagnostics.objects_revoked == 1 + + def test_a_version_move_reports_both_ends_and_counts_no_revocation(self) -> None: + """ + The diff runs at ``(key, version)``, so a key whose version moved yields + a put for the arrival and a tombstone for the departure — what a + listener that reads versions needs. ``objects_revoked`` counts per key, + though, and this key never left the payload: counting it would tell an + operator a revocation landed when a publish did. + """ + held = _SkillObjectSet() + reader = _ProtocolReader(held) + drive(reader, full_payload(("put-object", put_skill(object_version=3)))) + outcomes = drive( + reader, + full_payload(("put-object", put_skill(object_version=4)), state="basis-2"), + ) + arrived, departed = outcomes[-1].changes + assert (arrived["key"], arrived["version"]) == ("pdf-extraction", 4) + assert departed == {"key": "pdf-extraction", "version": 3} + assert reader.diagnostics.objects_revoked == 0 + + def test_a_change_transfer_revokes_nothing_by_omission(self) -> None: + """Only a full transfer states the whole payload. A delta that carries + no tombstone revoked nothing, and diffing one would drop every skill it + simply had no reason to mention.""" + held = _SkillObjectSet() + reader = _ProtocolReader(held) + drive(reader, full_payload(("put-object", put_skill()))) + outcomes = drive( + reader, + events( + ("server-intent", server_intent("xfer-changes")), + ("payload-transferred", transferred("basis-2")), + ), + ) + assert held.get("pdf-extraction", None) is not None + assert outcomes[-1].changes == [] + assert reader.diagnostics.objects_revoked == 0 + def test_a_change_transfer_applies_deltas_over_what_is_held(self) -> None: held = _SkillObjectSet() reader = _ProtocolReader(held) @@ -2808,6 +2874,33 @@ async def test_a_revocation_prunes_without_a_restart( finally: watcher.close() + async def test_a_full_transfer_that_omits_every_skill_prunes( + self, endpoint: Any, tmp_path: Any + ) -> None: + """ + The environment's last skill revoked. The full transfer that follows + carries nothing at all, so the only thing that can wake the watcher is + the revocation the transfer states by omission. + """ + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(full_payload(state="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) + assert store.diagnostics.objects_revoked == 1 + finally: + watcher.close() + async def test_a_new_version_is_rewritten_without_a_restart( self, endpoint: Any, tmp_path: Any ) -> None: From 0bdbdd35513d4a80e5f6c0aba38cd50e2335edbf Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 15:58:24 -0400 Subject: [PATCH 2/5] test(client): pin the key-mismatch detection surface (failing) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tests for TESTING.md §3.21/§3.24 as just reconciled: a store answering under a different key records the `ld.skills.integrity_failure` log record with `reason_code: key_mismatch` and records no product signal. `key_mismatch` cannot join `REASON_CODE_CASES` — that table is uniformly driven through `all_skills`, and this code is decided at the retrieval boundary after `verify_raw_skill` has passed, so it is unreachable from a listing. It gets its own test and is unioned into the vocabulary assertion, with the reason recorded there so the next reader does not try to move it into the table. Adds the redaction guard, called directly since no store can drive it, and the listing-path case: an object listed under a disagreeing map key is filed under its own key with neither surface firing. That one passes already; it is a regression guard against a "fix" that would break every multi-version store. 3 failing: no record is emitted yet, the vocabulary is short a token, and `record_key_mismatch` does not exist. The no-signal half already passes — the implementation is additive. Co-Authored-By: Claude Opus 5 --- packages/client/tests/test_skills.py | 147 ++++++++++++++++++++++++++- 1 file changed, 143 insertions(+), 4 deletions(-) diff --git a/packages/client/tests/test_skills.py b/packages/client/tests/test_skills.py index 15a885a..e7ff293 100644 --- a/packages/client/tests/test_skills.py +++ b/packages/client/tests/test_skills.py @@ -1348,6 +1348,47 @@ async def test_all_skills_returns_one_entry_per_key_at_the_newest_version( skills = await all_skills() assert sorted((s.key, s.version) for s in skills) == [("a", 2), ("b", 5)] + async def test_a_listed_object_is_filed_under_its_own_key( + self, + caplog: pytest.LogCaptureFixture, + make_raw_skill: Any, + recording_emitter: Any, + ) -> None: + """The counterpart to the pinned path's ``key_mismatch``. + + The asymmetry is deliberate rather than a gap. A listing carries no + requested key, so there is nothing for the object's key to disagree + *with*: identity comes off the object, and the store's map key is used + for one thing only — attributing a failure when the object's own key is + unusable. + + The seam never promised a map key spells a skill key, either: a store + holding several versions of one key has reason to spell it + ``key:version``, which is exactly what ``FDv2SkillStore`` does. Pinned + because the asymmetry with the pinned path is surprising enough to + invite a "fix" that would break every multi-version store. + """ + caplog.set_level("ERROR", logger="launchdarkly_ai_server.skills_core") + + class _MisfiledStore: + def get_object( + self, kind: str, key: str, version: int | None = None + ) -> Any: + return None + + def all_objects(self, kind: str) -> dict[str, Any]: + return {"filed-under-this": make_raw_skill(key="its-own-key")} + + skills_module._set_store(_MisfiledStore()) + skills_module._set_emitter_for_testing(recording_emitter) + + skills = await all_skills() + + assert [s.key for s in skills] == ["its-own-key"] + # Neither surface fires: nothing failed. + assert recording_emitter.records == [] + assert _integrity_records(caplog) == [] + async def test_a_store_answering_with_the_wrong_version_is_withheld( self, make_raw_skill: Any ) -> None: @@ -1759,16 +1800,114 @@ async def test_over_cap_content_with_a_lone_surrogate_reports_the_size( def test_the_case_table_exhausts_the_vocabulary(self) -> None: """The vocabulary is closed, and every token in it is reachable. - Both directions matter. A ninth token added to the source without a call - site fails here, and so does a ninth call site that invented a token the + Both directions matter. A tenth token added to the source without a call + site fails here, and so does a tenth call site that invented a token the table does not cover — which is what keeps the Python and TypeScript vocabularies from drifting apart one edit at a time. + + ``key_mismatch`` is added in rather than living in the table because it + is the one token that is *not* a verification failure: it is decided at + the retrieval boundary, after ``verify_raw_skill`` has already passed, so + it is unreachable through ``all_skills`` and cannot join a table that is + uniformly driven through it. Its own coverage is + ``test_key_mismatch_records_the_log_but_not_the_signal``. """ from launchdarkly_ai_server import skills_core - covered = {case.values[1] for case in REASON_CODE_CASES} + covered = {case.values[1] for case in REASON_CODE_CASES} | {"key_mismatch"} assert covered == skills_core.INTEGRITY_REASON_CODES - assert len(covered) == 8 + assert len(covered) == 9 + + async def test_key_mismatch_records_the_log_but_not_the_signal( + self, + caplog: pytest.LogCaptureFixture, + make_raw_skill: Any, + recording_emitter: Any, + ) -> None: + """The one-directional exception to "the two surfaces agree". + + The log record fires because a substituting store is a genuine tampering + indicator and the record is the customer-owned detection path — the only + one that works with telemetry off. Reusing the event identity is + deliberate: a customer's existing SIEM rule catches this case without + being rewritten, and ``reason_code`` is what distinguishes it. + + The product signal stays out of it because the overwhelmingly common + cause of a key mismatch is not an attacker but a broken store adapter — + a stale cache entry, a colliding key, a wrong index lookup — and + LaunchDarkly's own counter must not fill up with customers' adapter + bugs. That is the same false positive the pinned-non-dict path refuses + for the same reason. + """ + + class _AliasingStore: + def get_object( + self, kind: str, key: str, version: int | None = None + ) -> Any: + return make_raw_skill(key="served-key") + + def all_objects(self, kind: str) -> dict[str, Any]: + return {} + + skills_module._set_store(_AliasingStore()) + skills_module._set_emitter_for_testing(recording_emitter) + + assert await get_skill("asked-for") is None + + # The signal surface saw nothing at all, not merely no integrity signal. + assert recording_emitter.records == [] + + records = _integrity_records(caplog) + assert len(records) == 1 + record = records[0] + assert record["reason_code"] == "key_mismatch" + assert record["event"] == INTEGRITY_EVENT + assert record["action"] == "withheld" + assert record["language"] == "python" + assert record["reason"] + + # Both keys are named, and ``skill_key`` keeps the meaning it has on + # every other record — the key the *caller asked for* — so a rule + # grouping by it still works. The key the store actually answered under + # is what makes a broken adapter diagnosable, so it is a parseable field + # rather than prose buried in ``reason``. + assert record["skill_key"] == "asked-for" + assert record["served_key"] == "served-key" + + # Verification passed, so there is no hash disagreement to report and + # the two hash fields stay absent rather than being emitted as null. + assert "expected_hash" not in record + assert "observed_hash" not in record + assert None not in record.values() + + # Sorted, like every other record, so the line stays byte-comparable + # across SDKs. ``served_key`` has to land in its alphabetical place. + assert list(record) == sorted(record) + + # The structured mirror is required alongside the text. + errors = [r for r in caplog.records if r.levelname == "ERROR"] + assert len(errors) == 1 + assert errors[0].__dict__["ld_skills"] == record + + async def test_a_hostile_served_key_is_redacted( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Unreachable today, asserted anyway. + + Verification accepts the served key before this path runs, so the value + is well-formed by construction — but that is a property of the current + call order rather than of the recorder, and the guard is what keeps a + future reordering from publishing a body here. Called directly, since no + store can currently drive it. + """ + from launchdarkly_ai_server import skills_core + + skills_core.record_key_mismatch("asked-for", LOGGED_BODY) + + records = _integrity_records(caplog) + assert len(records) == 1 + assert records[0]["served_key"] == "" + assert LOGGED_BODY not in json.dumps(records[0]) async def test_the_event_name_is_in_the_message_text( self, caplog: pytest.LogCaptureFixture From 3a9df62bb5fc0c18439780e56aa79bad851724b4 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Thu, 17 Sep 2026 16:05:49 -0400 Subject: [PATCH 3/5] fix(client): record the key mismatch on the log surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the §3.21/§3.24 decision the previous commit's tests pin: a store answering under a different key now writes the `ld.skills.integrity_failure` log record with `reason_code: key_mismatch`, and still records no product signal. Adds `record_key_mismatch` beside `record_integrity_failure`, so the single-emission-site rule still holds by reading one module, and widens `IntegrityReasonCode` to nine tokens. The check stays where it was, at the retrieval boundary: `verify_raw_skill` is unary and this comparison is relational, so moving it inside would mean an optional expected-key parameter that silently disables the check when a caller omits it. The record carries both keys. `skill_key` keeps the meaning it has everywhere else — the key requested — and the key the store answered under goes in `served_key`, the one record-only field beyond the four, since it is what makes a broken adapter diagnosable and prose in `reason` is not parseable. No hash fields and no `version`: verification passed, so neither the hashes nor the version is what disqualified the answer, and reporting a served version beside a requested key would mix two frames in one record. Docs corrected in the same change: the README's `reason_code` table and record field table gain rows, `skill_key` is described as the *requested* key, and `AGENTS.md` records why `key_mismatch` cannot join `REASON_CODE_CASES` — that table is driven uniformly through `all_skills`, which cannot reach a code decided after verification passes. Verified the emitted JSON is byte-identical to the TypeScript SDK's for the same input, modulo `language`. 1875 tests passing, mypy and ruff clean. Co-Authored-By: Claude Opus 5 --- packages/client/README.md | 17 +++- packages/client/agents.md | 19 ++++- .../src/launchdarkly_ai_server/skills_core.py | 85 ++++++++++++++++++- 3 files changed, 112 insertions(+), 9 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index 75311a2..3c53c26 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -356,8 +356,9 @@ LaunchDarkly's AI SDKs for the same input. |---|---| | `event` | Always `ld.skills.integrity_failure`. | | `action` | Always `withheld` — the content was not returned to your code. | -| `skill_key` | The skill key, or `` when the delivered key was itself malformed. | -| `version` | The delivered version. Omitted when it was not a valid version. | +| `skill_key` | The skill key **requested**, or `` when the key was itself malformed. | +| `served_key` | Only on `key_mismatch`: the key the store actually answered under. Same redaction as `skill_key`. Omitted on every other failure mode. | +| `version` | The delivered version. Omitted when it was not a valid version, and on `key_mismatch`. | | `expected_hash` | The delivered `contentHash`, or `` when it was not one. Omitted when none was delivered. | | `observed_hash` | The sha256 the SDK computed. Omitted when the failure happened before anything was hashed. | | `reason_code` | A stable token naming the failure mode — see below. | @@ -378,12 +379,22 @@ could carry it, never appears in the record; neither does any filesystem path. | `not_utf8` | The content string had no UTF-8 encoding, so there are no bytes that could have been hashed. | | `over_size_cap` | The content exceeded the SDK's local size cap. | | `hash_mismatch` | The computed sha256 did not match the delivered `contentHash`. | +| `key_mismatch` | The store answered under a different key than the one requested. Carries an extra `served_key` field naming the key it answered under, and — uniquely — records **no** `AgentControl Skill Integrity Failure` signal. | -**`hash_mismatch` is the one worth paging on.** The other seven describe a malformed or +**`hash_mismatch` is the one worth paging on.** The other eight describe a malformed or truncated payload; a mismatch means content was delivered whose bytes are not the bytes LaunchDarkly hashed, which is a possible **active-tampering** signal. Alert on it, and treat `expected_hash` / `observed_hash` as the evidence pair. +**`key_mismatch` is the one code that reaches this record without the product signal.** It +is decided after verification has passed, and its usual cause is a bug in a custom +`SkillStore` adapter — a stale cache entry, a colliding key, a wrong index lookup — rather +than tampering, so it does not inflate LaunchDarkly's own integrity counter. It still +reaches this record, because a store substituting one skill for another is worth seeing, +and a rule on `ld.skills.integrity_failure` catches it without modification. Treat it like +`hash_mismatch` if `FDv2SkillStore` is your only store; behind a custom adapter, suspect +the adapter first. + #### Failing closed on tampering The log record above is the operator's surface. `get_skill_result` is the application's: diff --git a/packages/client/agents.md b/packages/client/agents.md index d995575..c80645d 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -548,9 +548,11 @@ Each of those choices is load-bearing; do not undo one as a simplification. - **`reason_code` is in the record only.** The signal's property set is the allowlist above and does not grow; the local record is where the detection vocabulary lives. -`reason_code` is a **closed vocabulary of exactly eight tokens** — `IntegrityReasonCode`, a -`Literal`, so a typo at a call site is a type error — one per `record_integrity_failure` -call site, and the same eight in every language implementation: +`reason_code` is a **closed vocabulary of exactly nine tokens** — `IntegrityReasonCode`, a +`Literal`, so a typo at a call site is a type error — and the same nine in every language +implementation. Eight are one per `record_integrity_failure` call site; the ninth, +`key_mismatch`, comes from `record_key_mismatch` and is the only one that fires the log +record **without** the product signal: | `reason_code` | Call site | |---|---| @@ -562,8 +564,17 @@ call site, and the same eight in every language implementation: | `not_utf8` | `verified_bytes` — `UnicodeEncodeError` on encode (wire-`str` path only; a `Skill` already holds bytes) | | `over_size_cap` | `verified_bytes` — over `MAX_SKILL_CONTENT_BYTES` | | `hash_mismatch` | `verified_bytes` — observed sha256 != `contentHash` | +| `key_mismatch` | `resolve_from_store` — the served object's own `key` is not the key requested. **Log record only, no signal**, and carries a `served_key` field no other record has | -Adding a ninth failure mode means widening `IntegrityReasonCode`, adding a case to +`key_mismatch` cannot join `REASON_CODE_CASES`: that table is driven uniformly through +`all_skills`, and this code is decided at the retrieval boundary after `verify_raw_skill` +has passed, so a listing cannot reach it. It is unioned into the exhaustiveness assertion +instead, and covered by `test_key_mismatch_records_the_log_but_not_the_signal`. The +record-without-signal split is deliberate — a mismatch is usually a broken store adapter +rather than an attacker, and LaunchDarkly's counter must not fill with customers' adapter +bugs — and tests pin both directions. Do not "fix" it by emitting the signal. + +Adding a tenth failure mode means widening `IntegrityReasonCode`, adding a case to `REASON_CODE_CASES` in `test_skills.py` (whose exhaustiveness assertion fails otherwise), documenting it in the README table, **and** doing the same in the other language SDKs. A token added on one side only is a drift bug: a customer's detection rule stops matching diff --git a/packages/client/src/launchdarkly_ai_server/skills_core.py b/packages/client/src/launchdarkly_ai_server/skills_core.py index 3cc1751..a6eb523 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_core.py +++ b/packages/client/src/launchdarkly_ai_server/skills_core.py @@ -94,12 +94,21 @@ "not_utf8", "over_size_cap", "hash_mismatch", + "key_mismatch", ] """ -The closed ``reason_code`` vocabulary — one token per -``record_integrity_failure`` call site. Stable: a detection rule written against +The closed ``reason_code`` vocabulary. Stable: a detection rule written against these tokens keeps working, so adding one is a deliberate edit here rather than a new string invented at the call site that needed it. + +Eight of the nine are one token per ``record_integrity_failure`` call site, +decided inside ``verify_raw_skill`` over a single object, and they fire **both** +detection surfaces. ``key_mismatch`` is the exception on both counts: it comes +from ``record_key_mismatch`` at the retrieval boundary, after verification has +already passed, and it fires the log record only. It shares this vocabulary +anyway because a customer's detection rule cares that integrity failed, not +about which layer noticed — see ``record_key_mismatch`` for why the signal +stays out. """ INTEGRITY_REASON_CODES: frozenset[str] = frozenset(get_args(IntegrityReasonCode)) @@ -352,6 +361,70 @@ def record_integrity_failure( emit(_SIGNAL_INTEGRITY_FAILURE, properties) +def record_key_mismatch(requested: Any, served: Any) -> None: + """ + Records a store answering under a key other than the one requested. + + **Log record only — no product signal.** This is the one integrity failure + that fires one surface rather than both, and the asymmetry is the decision + rather than an oversight. + + The record fires because a substituting store is a genuine tampering + indicator, and the record is the customer-owned detection path — the only + one that works when telemetry is opt-out or the instance has no telemetry + destination at all. It reuses ``INTEGRITY_FAILURE_EVENT`` deliberately: that + string is a documented compatibility surface a customer's SIEM matches on, + so reusing it means an existing rule catches this case without being + rewritten, with ``reason_code`` distinguishing it. + + The signal stays out because the overwhelmingly common cause of a key + mismatch is not an attacker but a **broken store adapter** — a stale cache + entry, a colliding key, a wrong index lookup. Counting those as integrity + failures in LaunchDarkly's own product counter is the same false positive + ``resolve_from_store`` already refuses when a pinned ``get_object`` answers + with a non-dict: it reads that as ``absent`` rather than inventing a + tampering signal from a merely broken adapter. + + Lives here, beside ``record_integrity_failure``, so the single-emission-site + rule still holds by reading one module. + + Both keys are shape-checked and redacted on the same rule as every other key + that reaches a surface. *served* cannot actually be hostile on the path that + calls this — ``verify_raw_skill`` accepted it first — but that is a property + of the current call order rather than of this function, and the check is + what stops a future reordering from publishing a body here. + """ + record: dict[str, Any] = { + "event": INTEGRITY_FAILURE_EVENT, + "action": _ACTION_WITHHELD, + "reason_code": "key_mismatch", + # Named apart from the eight so a reader of the line can tell the + # retrieval boundary from a verification failure without the spec. + "reason": ( + "the skill store answered under a different key than the one requested" + ), + "language": _LANGUAGE, + # ``skill_key`` keeps the meaning it has on every other record — the key + # the *caller asked for* — so a rule that groups by it keeps working. + "skill_key": requested if is_valid_skill_key(requested) else "", + # The key the store answered under: the one datum that makes a broken + # adapter diagnosable, so it is a parseable field rather than prose + # buried in ``reason``. Record-only, never on the signal's allowlist. + "served_key": served if is_valid_skill_key(served) else "", + } + # No ``expected_hash``, ``observed_hash`` or ``version``: verification + # passed, so there is no hash disagreement to report and the served object's + # version is not what disqualified the answer — reporting it beside a + # ``skill_key`` that means the requested key would mix the two frames. + # Absent fields stay absent rather than being emitted as null. + logger.error( + "%s %s", + INTEGRITY_FAILURE_EVENT, + json.dumps(record, sort_keys=True, separators=(",", ":")), + extra={"ld_skills": record}, + ) + + def record_materialized( skill_key: str, content_bytes: int, content_hash: str, reconcile_action: str ) -> None: @@ -776,6 +849,14 @@ def resolve_from_store( # is invited to tolerate. It is not ``wrong_version`` either — that # token names a version mismatch specifically, and there is deliberately # no ``wrong_key`` to parallel it. + # + # Records the log surface but not the product signal. ``verify_raw_skill`` + # has already passed, so this is not a verification failure and does not + # go through ``record_integrity_failure``; see ``record_key_mismatch`` + # for why the two surfaces part company here. The asymmetry is pinned by + # a test in both directions, because an implementation that emitted the + # signal too would look correct from every other angle. + record_key_mismatch(key, skill.key) return Resolution( reason="integrity_failure", error=( From a4fc1eb1855c26b47737c2dfae5c0f194d2b8caa Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Fri, 18 Sep 2026 16:14:41 -0400 Subject: [PATCH 4/5] fix(client): keep the resume point on the payload skills arrive on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A transfer this layer declines as foreign kept its contents out of the store but still handed its selector up as the new basis, so the next poll or stream resumed from someone else's payload while every diagnostic read healthy. The check was also gated on a pending set, which a `none` intent never builds — so a `none` for another payload was not even recognised as foreign, and its selector was adopted outright. Ask the question unconditionally and drop the basis of a payload whose contents were thrown away, matching js-ai-sdk#67. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 14 ++++- packages/client/tests/test_skills_fdv2.py | 63 +++++++++++++++++++ 2 files changed, 75 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 7169d49..9c9f226 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -776,7 +776,12 @@ def _payload_transferred(self, data: Any) -> _TransferOutcome: state = data.get("state") if isinstance(data, dict) else None version = data.get("version") if isinstance(data, dict) else None payload_id = self._intent_payload_id or _payload_id_from_selector(state) - if self._pending is not None and self._is_foreign_payload(payload_id): + # Asked regardless of whether a pending set exists: a ``none`` intent + # builds none, and the transfer that completes it still names a payload + # whose selector must not become the resume point if it is not the + # payload skills arrive on. + foreign = self._is_foreign_payload(payload_id) + if foreign: self._warn_foreign_payload(payload_id) self.diagnostics.payloads_ignored += 1 self._changes = [] @@ -817,7 +822,12 @@ def _payload_transferred(self, data: Any) -> _TransferOutcome: return _TransferOutcome( committed=True, changes=changes, - basis=state if isinstance(state, str) and state else None, + # A declined payload must not move the resume point. Adopting the + # selector of a transfer whose contents this layer just threw away + # would ask the next poll or stream to resume from someone else's + # payload, and skill updates could stop arriving while every + # diagnostic still read healthy. + basis=state if not foreign and isinstance(state, str) and state else None, ) def _abandon_in_flight(self) -> None: diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 93210a5..8b9ac95 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -1006,6 +1006,48 @@ def test_a_declined_transfer_warns_once_however_often_it_repeats( assert len(_payload_warnings(caplog, "was not applied")) == 1 assert reader.diagnostics.payloads_ignored == 2 + def test_a_declined_transfer_does_not_move_the_resume_point(self) -> None: + """ + Ignoring a foreign payload's contents while adopting its resume point + would ask the next poll or stream to resume from someone else's + payload: skill updates could stop arriving while every diagnostic read + healthy. + """ + reader = _ProtocolReader(_SkillObjectSet()) + ours = drive( + reader, skill_payload(("put-object", put_skill()), state="skills-basis") + ) + assert ours[-1].basis == "skills-basis" + outcomes = drive( + reader, + skill_payload( + ("put-object", put_flag()), payload_id="env-flags", state="flag-basis" + ), + ) + assert outcomes[-1].basis is None + assert reader.diagnostics.payloads_ignored == 1 + + def test_a_none_intent_for_another_payload_moves_nothing(self) -> None: + """ + A ``none`` intent builds no pending set, and the foreign check must not + be gated on one: the transfer that follows still names a payload, and + adopting its selector would resume the next connection from someone + else's payload with every diagnostic reading healthy. + """ + held = _SkillObjectSet() + reader = _ProtocolReader(held) + drive(reader, full_payload(("put-object", put_skill()), state="basis-skills")) + outcomes = drive( + reader, + events( + ("server-intent", server_intent("none", "env-flags")), + ("payload-transferred", transferred("basis-flags")), + ), + ) + assert outcomes[-1].basis is None + assert reader.diagnostics.payloads_ignored == 1 + assert held.get("pdf-extraction", None) is not None + def test_a_full_transfer_of_the_skill_payload_still_empties_it(self) -> None: """Every skill deleted is a real state, and the guard must not mask it.""" held = _SkillObjectSet() @@ -1235,6 +1277,27 @@ def test_the_basis_advances_across_successive_payloads(self, endpoint: Any) -> N bases = [r["query"].get("basis") for r in endpoint.requests[:3]] assert bases == [None, "basis-1", "basis-2"] + def test_the_basis_stays_on_the_skill_payload_when_another_transfers( + self, endpoint: Any + ) -> None: + """The wire half of the declined-payload case: the store must resume from + the payload skills arrive on, not from the one it just threw away.""" + endpoint.queue_poll( + full_payload(("put-object", put_skill()), state="skills-basis") + ) + endpoint.queue_poll( + events( + ("server-intent", server_intent("xfer-full", "env-flags")), + ("put-object", put_flag()), + ("payload-transferred", transferred("flag-basis")), + ) + ) + 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, "skills-basis", "skills-basis"] + 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) From a30750a866ca4a2c45cd12c7a52754ce86a95458 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Fri, 18 Sep 2026 16:35:03 -0400 Subject: [PATCH 5/5] fix(client): pair the poll etag with the basis it was issued against MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ``_poll_once`` held one ``_etag`` with no record of which request produced it, and adopted it before the body was applied. Two consequences: An ETag validates one representation of one resource, and the ``basis`` selector is part of the request that names it. Once a payload moved the basis on, the etag from the response before it was still offered, so a server validating it would be answering the question the store had stopped asking. Held as a pair now, and offered only while the pair holds — the base SDK keys its etag cache by request URL for the same reason. The cost is one unconditional request after each commit, which was never going to be a 304. A body that broke off partway — an ``error`` or ``goodbye`` after an announced transfer — left the payload it described unapplied, but its etag was already stored. The next 304 then reported that store as current and reset the failure count, where the 200 it replaced would have been retried and eventually given up on. Adopted after the body is applied in full. Ports js-ai-sdk#67. Co-Authored-By: Claude Opus 5 --- .../src/launchdarkly_ai_server/skills_fdv2.py | 30 +++++++++-- packages/client/tests/test_skills_fdv2.py | 50 +++++++++++++++++-- 2 files changed, 72 insertions(+), 8 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index 9c9f226..6e7b31c 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -1585,6 +1585,12 @@ def __init__( self._basis: str | None = None self._etag: str | None = None + # The basis ``_etag`` was issued against. An ETag validates one + # representation of one resource, and the basis is part of the request + # that names it; holding the pair is what lets ``_poll_once`` tell an + # etag that still answers the question it is about to ask from one that + # answers a question it has stopped asking. + self._etag_basis: str | None = None self._requester = _requester or _Requester( sdk_key.strip(), @@ -1904,6 +1910,7 @@ def _run(self) -> None: if not exhausted: self._basis = None self._etag = None + self._etag_basis = None if exhausted: self._give_up(str(exc)) return @@ -2017,18 +2024,33 @@ def _apply(self, name: str, data: Any) -> None: def _poll_once(self) -> None: with self._lock: - basis, etag = self._basis, self._etag + basis = self._basis + # Offered only while the pair still holds. The basis is part of the + # request, so an etag issued before the basis moved validates a + # payload this store has stopped asking for, and a server answering + # it ``304`` would be answering the previous question. One + # unconditional request after each commit is the whole cost: a + # payload that changed was never going to be a 304 anyway. + etag = self._etag if self._etag_basis == basis else None 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. + # cached basis is not blocked on a transfer the server will not + # send. It is a current answer because the etag that asked for it + # was issued for a body this store applied in full. self._publish_first_payload() return for name, data in result.events: self._apply(name, data) + with self._lock: + # Adopted only once the whole body has been applied. A body that + # broke off partway — an ``error`` or ``goodbye`` after an announced + # transfer — left the payload it described unapplied, and keeping + # its etag would let the next 304 report a store that is missing + # that payload as current and healthy. + self._etag = result.etag + self._etag_basis = basis def _stream_once(self) -> None: with self._lock: diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 8b9ac95..3b21654 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -1298,13 +1298,55 @@ def test_the_basis_stays_on_the_skill_payload_when_another_transfers( bases = [r["query"].get("basis") for r in endpoint.requests[:3]] assert bases == [None, "skills-basis", "skills-basis"] - 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"') + def test_an_etag_is_returned_for_the_basis_it_was_issued_against( + self, endpoint: Any + ) -> None: + """ + An ETag validates one representation of one resource, and the basis is + part of the request that names it. ``W/"v1"`` answers the request that + carried no basis at all, so it is not offered once the payload it came + with moved the basis on; ``W/"v2"`` answers a request from ``basis-1``, + which is still the question being asked, so it is. + """ + endpoint.queue_poll( + full_payload(("put-object", put_skill()), state="basis-1"), etag='W/"v1"' + ) + endpoint.queue_poll( + events(("server-intent", server_intent("none"))), etag='W/"v2"' + ) 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"' + 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-1"] + offered = [r["if_none_match"] for r in endpoint.requests[:3]] + assert offered == [None, None, 'W/"v2"'] + + def test_the_etag_of_a_body_never_applied_is_not_offered( + self, endpoint: Any + ) -> None: + """ + The body announced a transfer and then broke off, so the payload it + described was never committed. Offering its etag would invite a 304 that + reports a store still missing that payload as current and healthy — and + unlike the 200 it replaces, a 304 carries nothing to notice that on. + """ + endpoint.queue_poll( + events( + ("server-intent", server_intent("xfer-full")), + ("put-object", put_skill()), + ("error", {"reason": "cut off mid-payload"}), + ), + etag='W/"v1"', + ) + endpoint.queue_poll( + full_payload(("put-object", put_skill()), state="basis-1"), etag='W/"v2"' + ) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=5) is True + assert endpoint.requests[1]["if_none_match"] is None + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None def test_a_304_keeps_the_held_content(self, endpoint: Any) -> None: endpoint.queue_poll(full_payload(("put-object", put_skill())), etag='W/"v1"')