fix(client): publish the revocations an xfer-full states by omission

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 `<invalid-key>` 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 `<invalid-key>` 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 `<not-a-sha256-digest>` 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:

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

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 "<invalid-key>",
+        # 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 "<invalid-key>",
+    }
+    # 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=(

packages/client/src/launchdarkly_ai_server/skills_fdv2.py

@@ -567,6 +567,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
@@ -720,11 +771,30 @@ 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 = []
         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:
@@ -747,7 +817,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:
@@ -1507,6 +1582,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(),
@@ -1735,7 +1816,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
@@ -1825,6 +1908,7 @@ def _run(self) -> None:
                         if not exhausted:
                             self._basis = None
                             self._etag = None
+                            self._etag_basis = None
                             repairing_state = True
                     if exhausted:
                         self._give_up(str(exc))
@@ -1945,18 +2029,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: