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=( diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index d04cb89..cb8d60c 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/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: 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 diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index 5ec48bf..b11ef4e 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) @@ -940,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() @@ -1169,13 +1277,76 @@ 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_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_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_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"') @@ -2851,6 +3022,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: