diff --git a/packages/client/README.md b/packages/client/README.md index 924aefdf..6d7e5022 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -533,6 +533,10 @@ async def main(): asyncio.run(main()) ``` +This reads the config once, which is right at boot. To keep a config's skills on disk as the +config changes, pass `config_skills(key, context)` to `watch_skills` instead; see +[Following an AI Config's changes](#following-an-ai-configs-changes). + **Know what `"*"` asks for.** Passing `"*"` instead of a reference list writes the **whole project library**, which puts every skill's `description` into the agent's context, including skills no AI Config references and skills belonging to other teams. `write_skills(skill_refs(...), root)`, @@ -806,11 +810,71 @@ an outage the store keeps serving its last content and retries for as long as it `write_skills`' default `on_unavailable="keep"` leaves managed files alone, so an outage does not read as "everything was revoked". -**With an explicit skill list, revocation does not reach the disk.** Given a list such as -`skill_refs(config)`, a skill deleted in LaunchDarkly is reported as an `error` action under -its key, and its `SKILL.md` is kept rather than pruned. The watcher also listens only to the -skill store, not to flag changes, so unpinning a skill from a variation is not seen either. Use -`"*"` when revocation must reach the disk. +**What reaches the disk depends on what you pass as `skills`.** A fixed list, such as +`skill_refs(config)`, is never re-read, so the watcher cannot see a config change: + +| Change in LaunchDarkly | `"*"` | Fixed list | `config_skills(key, context)` | +|---|---|---|---| +| A skill is deleted (possible only once no variation pins it) | Pruned | Stays on disk, reported `error` on every run | Already pruned, when the config stopped referencing it | +| A config moves `pdf-extraction` from v2 to v3 | Unaffected: `"*"` already follows the newest version | v2 stays on disk, reported `error`; v3 is never written | v3 replaces v2 | +| A config stops referencing a skill | Unaffected: `"*"` asks for the whole library | Stays on disk | Pruned | + +So a revoked skill leaves the disk under `"*"` and `config_skills`, never under a fixed list. +Keep a fixed list for a set that genuinely does not change for the life of the process. + +#### Following an AI Config's changes + +`config_skills(key, context)` returns a `SkillSource`: a requested set that is read on every +reconcile instead of being fixed when you call `watch_skills`. The watcher also re-reconciles +when that AI Config changes in the base LaunchDarkly client, so a version bump or an unpinned +skill reaches the disk within `debounce`. + +```python +from launchdarkly_ai_server import init_client +from launchdarkly_ai_server.experimental.skills import config_skills, watch_skills + +await init_client() # config_skills needs an initialized client; it never creates one +report, watcher = await watch_skills( + config_skills("doc-agent", {"kind": "user", "key": "user-123"}), + ".claude/skills", +) +``` + +- **A failed evaluation never reads as "no skills".** Each read evaluates the config with + `variation(key, context, None)`. Only a disabled variation, or a config with no `skills` + field, answers `[]`, and those are answers from LaunchDarkly, so their skills are pruned. No + initialized client, a `variation` that raises, the default (a missing flag or a client that + is not ready), or a malformed variation or `skills` field is a failed read: the reconcile + reports it as a run-level `error` (or raises, with `on_unavailable="raise"`) and changes + nothing on disk. There is no retry timer: the next store or config change is the retry, + and a client that finishes initializing reports a change for every flag. +- **Each read is a real evaluation.** It records an evaluation event in the base SDK, as + `inspect_config` does: once per reconcile, which the debounce bounds. It sends no AI + telemetry. +- **Following changes needs the base client's change signal.** The watcher registers one + `flag_tracker` listener and raises `RuntimeError` when no AI client is initialized, when the + client has no `flag_tracker` (a client brought to `init_client` that implements only + `variation`), or when the client is offline. The listener only wakes the watcher; the + config is read on the watcher's own thread, never on the base SDK's delivery thread. +- **A dropped skill is pruned on the next reconcile whose retrieval is available.** A skill + the store answers `absent` for keeps its file under its own key, reported `error`, and does + not stop the dropped skill being pruned in the same run. A store that raises, has not + received its initial data, or runs out of `timeout` holds the prune until a later run. +- **Daemon mode follows skill-store changes only.** With `use_ldd=True`, or + `ldclient.datasystem.daemon(...)`, the base client has no data source, so its `flag_tracker` + never fires and nothing reports it. Call `watcher.notify()` yourself if you learn of config + changes some other way. +- **Several configs, or a custom resolution, take a hand-built source.** + `SkillSource(refs, config_keys=(...))` takes a zero-argument function, sync or async, + returning `SkillReference` values and key strings, plus the config keys whose changes should + re-read it. Raise from `refs` when you cannot tell; returning `[]` prunes every managed + skill. An answer that is not a sequence of references and keys, or holds one invalid entry, + is a failed read too. With `config_keys=()` the watcher subscribes to nothing; call + `watcher.notify()` from your own change listener. `config_keys` must be a sequence: + `config_keys="doc-agent"` raises `TypeError`. +- **`timeout` covers the read, with one limit.** An async `refs` still pending at the deadline + is cancelled. A synchronous one cannot be interrupted, so a slow one uses up the time left + for the rest of the reconcile. **Without the watcher, the revocation bound is process lifetime.** If you call `write_skills` once at boot and never run `watch_skills`, a skill revoked after boot stays on disk, and in the @@ -841,16 +905,18 @@ that skips verification. | Export | Description | |---|---| +| `config_skills(key, context)` | A `SkillSource` that evaluates AI Config `key` for `context` on every read. Raises from the read, rather than answering `[]`, for anything but a disabled variation or a parsed config. Never initializes the client. See *Following an AI Config's changes* above. | +| `SkillSource(refs, config_keys=())` | A requested set read when it is needed: `refs` is a zero-argument function, sync or async, returning `SkillReference` values and key strings; `config_keys` names the AI Configs whose changes make a watcher read it again. Frozen; `config_keys` is held as a tuple. A non-callable `refs`, a string `config_keys`, or an entry that is not a non-empty string raises `TypeError`. | | `skill_refs(config)` | Project a config's `skills` array into `list[SkillReference]`. Pure — no client, store, or network needed. Returns `[]` when the field is absent. Raises `ValueError` when the config is not a dict (including the `None` a failed `inspect_config` returns) or the field is present but malformed (including `null`), so neither reaches a pruning reconcile as "no skills". | | `get_skill(key, *, version=None)` | One verified skill, or `None`. `version=None` means newest available; a specific `version` matches exactly. Raises only when no store is configured. | | `get_skill_result(key, *, version=None)` | The same retrieval, reporting **why**: a frozen `SkillOutcome` with `.skill`, `.reason` (`ok` / `absent` / `integrity_failure` / `store_unavailable` / `wrong_version`), and `.detail`. See *Failing closed on tampering* above. Raises only when no store is configured. | | `get_skills(refs)` | Batch form. Accepts `SkillReference` values and bare key strings (string = latest). Results follow input order; missing or unverifiable entries are omitted. | | `all_skills()` | Every verified skill the store holds, one per key at its newest version. | -| `write_skills(skills, root, *, prune=True, timeout=10.0, on_unavailable="keep")` | Materialize skills under `root`, returning a `ReconcileReport`. `prune` removes formerly-managed skills no longer requested. `on_unavailable="raise"` raises instead of reporting when content cannot be retrieved. Raises `ValueError` for an unusable root, a negative or non-finite `timeout`, or an unrecognised `on_unavailable`. `timeout=0` is valid and makes every skill report an error. **Performs synchronous filesystem I/O — see the note below.** | +| `write_skills(skills, root, *, prune=True, timeout=10.0, on_unavailable="keep")` | Materialize skills under `root`, returning a `ReconcileReport`. `skills` is a sequence of `Skill` / `SkillReference` / key strings, `"*"`, or a `SkillSource`, which is read once; a source that raises or answers anything malformed is an unavailable retrieval. `prune` removes formerly-managed skills no longer requested. `on_unavailable="raise"` raises instead of reporting when content cannot be retrieved. Raises `ValueError` for an unusable root, a negative or non-finite `timeout`, or an unrecognised `on_unavailable`. `timeout=0` is valid and makes every skill report an error. **Performs synchronous filesystem I/O — see the note below.** | | `SkillStore` | The structural interface content arrives through: `get_object(kind, key, version=None)`, `all_objects(kind)`, optional `is_initialized()`, `add_listener(kind, fn)` / `remove_listener(kind, fn)`. A store without `is_initialized()` is treated as initialized. Both shipped stores deliver only the skill kind, so `add_listener` on any other kind raises. | | `InMemorySkillStore(objects=None)` | A dict-backed store with `put(raw)`, for local development and testing. Holds several versions of a key. | | `FDv2SkillStore(sdk_key, *, base_uri=…, stream_uri=…, mode="stream", …)` | The delivery transport: a store fed by LaunchDarkly over the SDK-facing FDv2 channel. `start()`, `wait_for_skills(timeout)`, `is_initialized()`, `close()`, `diagnostics`, `failed`; also a context manager. `close()` is **final** — `start()` afterwards raises. `poll_interval`, `read_timeout`, `initial_backoff` and `max_backoff` must be positive and finite, and `initial_backoff` may not exceed `max_backoff`. **Server-side only.** See *Receiving skills from LaunchDarkly* above. | -| `watch_skills(skills, root, *, debounce=0.5, on_reconcile=None, …)` | `write_skills` plus a re-reconcile on every delivery change, so revocation takes effect within `debounce` rather than at the next restart. Returns `(initial report, SkillWatcher)`; close the watcher when done. `debounce` is in **seconds**, non-negative and finite. `on_reconcile` receives each *subsequent* report. One watcher per root. | +| `watch_skills(skills, root, *, debounce=0.5, on_reconcile=None, …)` | `write_skills` plus a re-reconcile on every delivery change, and, for a `SkillSource`, on every change to one of its `config_keys`, so revocation takes effect within `debounce` rather than at the next restart. `watcher.notify()` schedules a reconcile through the same debounce. Returns `(initial report, SkillWatcher)`; close the watcher when done. `debounce` is in **seconds**, non-negative and finite. `on_reconcile` receives each *subsequent* report. One watcher per root. | | `StoreDiagnostics` | What the transport has seen: `payloads_transferred`, `skill_objects_received`, `objects_ignored`, `objects_revoked`, `payloads_ignored`, `hashless_objects`, `connection_failures`, `last_error`. | Configure the store with `set_skill_store(store)`. It applies on every call, before or after @@ -874,7 +940,8 @@ and `MANIFEST_VERSION`. So are the closed-set types, for annotating your own hel `ok` / `store_unavailable` / `wrong_version`). **`write_skills` blocks.** It is `async` for parity with the other accessors, but awaits -nothing: all file I/O runs inline and holds the event loop for the duration. Wrap it in +nothing except an async `SkillSource`: all file I/O runs inline and holds the event loop for +the duration. Wrap it in `asyncio.to_thread` if that matters; `timeout` is checked between steps, not mid-step. Do not run two reconciles against the same root concurrently — they race on the manifest. diff --git a/packages/client/agents.md b/packages/client/agents.md index b93877a7..eb64680e 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -32,7 +32,8 @@ No other `launchdarkly-ai-*` package may define or duplicate these. They import | `src/launchdarkly_ai_server/skills.py` | Agent Skills, retrieval half — `skill_refs`, `get_skill`/`get_skills`/`all_skills`, `InMemorySkillStore`, and the store/telemetry injection points `_set_store` / `_set_emitter_for_testing` | | `src/launchdarkly_ai_server/skills_core.py` | Shared skills internals — the `SkillStore` seam, module state, the telemetry seam and its three recorders, integrity verification, and store resolution. Imported by both `skills.py` and the materialization layer; imports neither | | `src/launchdarkly_ai_server/skills_fdv2.py` | Agent Skills, delivery transport — the FDv2 protocol, the wire-key/`version` translation, the held object set, and `FDv2SkillStore`. Sits **below** the store interface; imports `skills_core` only, and nothing imports it | -| `src/launchdarkly_ai_server/skills_watch.py` | Agent Skills, eager re-reconcile — `watch_skills` / `SkillWatcher`, wiring the store's change listener to `write_skills`. Sits **above** `skills_fs` and modifies none of it | +| `src/launchdarkly_ai_server/skills_watch.py` | Agent Skills, eager re-reconcile — `watch_skills` / `SkillWatcher`, wiring the store's change listener, and for a `SkillSource` the base client's `flag_tracker` listener, to `write_skills`. Sits **above** `skills_fs` and modifies none of it | +| `src/launchdarkly_ai_server/skills_source.py` | Agent Skills, config-scoped requested set — `config_skills(key, context)`, which evaluates one AI Config per read and classifies every failure as a raised read, never `[]`. With `skills_watch`, the only skills module that imports `lifecycle`, and it takes only `get_client` | | `src/launchdarkly_ai_server/skills_fs.py` | Agent Skills, materialization half — `write_skills`, request resolution, the manifest format and on-disk filenames, per-skill reconcile, and pruning | | `src/launchdarkly_ai_server/safe_fs.py` | Descriptor-pinned filesystem primitives — `atomic_write`, `unlink_file`, `pinned_directory`, `open_directory_nofollow`, `open_or_create_directory`, `SymlinkRefused`, `DirectoryMissing`, and the `*at()` capability probe. Owns the descriptor-vs-path platform split; knows nothing about skills | | `src/launchdarkly_ai_server/utils.py` | `parse_template`, `parse_json_with_possible_fences`, `create_handler`, `parse_usage`, `make_track_data`, `to_ld_context` | @@ -83,7 +84,7 @@ from launchdarkly_ai_server.experimental.evaluations import init_evaluations from launchdarkly_ai_server.experimental.skills import ( set_skill_store, SkillStore, InMemorySkillStore, FDv2SkillStore, StoreDiagnostics, skill_refs, get_skill, get_skill_result, get_skills, all_skills, write_skills, - watch_skills, SkillWatcher, + watch_skills, SkillWatcher, SkillSource, config_skills, Skill, SkillReference, SkillOutcome, ReconcileAction, ReconcileReport, SKILL_FILENAME, MANIFEST_FILENAME, MANIFEST_VERSION, ReconcileActionKind, OnUnavailable, SkillOutcomeReason, # the three closed-set unions @@ -219,7 +220,16 @@ Three layers, in increasing order of blast radius: an actionable `RuntimeError`. A delivery transport can be added behind the seam without touching the public API. 3. **Materialization** — `write_skills(skills, root)` writes `//SKILL.md` and - reconciles against a manifest at `/.launchdarkly-skills.json`. + reconciles against a manifest at `/.launchdarkly-skills.json`. `skills` may be a + `SkillSource` (`types.py`), whose `refs` is read once per reconcile; `skills_fs` + recognizes it by type and never imports `skills_source`. A source that raises or answers + anything but a sequence of valid references and keys is a run-level unavailable + retrieval, never an empty set. + +`config_skills(key, context)` is the documented form for config-scoped materialization +under `watch_skills`. Do not write examples that feed `skill_refs(inspect_config(...))` to +`watch_skills`: that is a fixed list, so the watcher never follows the config. At boot, +with `write_skills`, either form is safe. ### The store seam, and why version is part of the lookup @@ -955,17 +965,35 @@ true. Tampered content must never trigger deletion. ### 6. Expecting revocation to reach a boot-only `write_skills` deployment -Even with `watch_skills`, only the `"*"` form removes a revoked skill from disk. With an -explicit list such as `skill_refs(config)`, the list is fixed: a skill the store answers -`absent` for is reported as an `error` and its files are kept, and the watcher listens only to -the skill store, so unpinning a skill or moving it to a new version is not seen until the refs -are read again. +Even with `watch_skills`, a fixed list never follows its config. With an explicit list such +as `skill_refs(config)`, a skill the store answers `absent` for is reported as an `error` and +its files are kept, and unpinning a skill or moving it to a new version is not seen until the +refs are read again. `"*"` prunes a skill deleted from the library; `config_skills(key, +context)` is re-read on every reconcile and on every change to its config, so a version bump +replaces the file and an unpinned skill is pruned. Without `watch_skills`, the revocation bound is process lifetime: a skill revoked after boot stays on disk until the process reconciles again, so a restart (or an explicit re-run of `write_skills`) is the incident-response action — and content an agent has already read into a conversation is out of reach at this layer either way. +### 7. Reading a `SkillSource` inside the flag listener, or answering `[]` for "I could not tell" + +The base SDK calls `flag_tracker` listeners on the thread applying flag data, and under FDv2 +while holding the store lock every `variation()` takes. `SkillWatcher._on_flag_change` must +only set the wake flag; the source is read on the watcher's worker. A source (hand-built or +`config_skills`) that returns `[]` when evaluation fails tells the reconcile that every +skill was revoked, so it prunes them all. Raise instead. For the same reason the +`config_skills` `variation` default is `None`: any default that reads as a disabled variation +would turn "client not ready" into "no skills". + +A client without a working change signal (no client, no `flag_tracker`, `is_offline()`) makes +`watch_skills` raise for a source with `config_keys`. Do not turn that into a silent no-op: a +watcher that never hears about config changes looks exactly like one whose config never +changed. Daemon mode passes the probe and never fires; that gap is documented, not detected. +Tests use a hand-built fake client, never a `MagicMock`, which passes the probe and never +fires. + --- ## Adding a New Export diff --git a/packages/client/api-report/launchdarkly_ai_server.experimental.skills.txt b/packages/client/api-report/launchdarkly_ai_server.experimental.skills.txt index 11f3fee8..4732aab2 100644 --- a/packages/client/api-report/launchdarkly_ai_server.experimental.skills.txt +++ b/packages/client/api-report/launchdarkly_ai_server.experimental.skills.txt @@ -1,5 +1,6 @@ # Public API of launchdarkly_ai_server.experimental.skills. Generated by `make api-report`; do not edit. all_skills() -> list[Skill] [async] +config_skills(key: str, context: LDContext) -> SkillSource FDv2SkillStore [class] __init__(sdk_key: str, *, base_uri: str | None = None, stream_uri: str | None = None, mode: Mode = 'stream', poll_interval: float = 30.0, read_timeout: float | None = None, initial_backoff: float = 1.0, max_backoff: float = 30.0, _requester: Any = None) __enter__() -> FDv2SkillStore @@ -61,11 +62,15 @@ SkillReference [class, @dataclass(frozen=True)] __init__(key: str, version: int) key: str version: int +SkillSource [class, @dataclass(frozen=True, init=False)] + __init__(refs: Callable[[], Sequence[SkillReference | str] | Awaitable[Sequence[SkillReference | str]]], config_keys: Sequence[str] = ()) + config_keys: tuple[str, ...] + refs: Callable[[], Sequence[SkillReference | str] | Awaitable[Sequence[SkillReference | str]]] SkillStore(Protocol) [class] all_objects(kind: str) -> dict[str, dict[str, Any]] get_object(kind: str, key: str, version: int | None = None) -> dict[str, Any] | None SkillWatcher [class] - __init__(request: Sequence[Skill | SkillReference | str] | str, root: str | os.PathLike[str], store: Any, *, prune: bool, timeout: float, on_unavailable: OnUnavailable, debounce: float, on_reconcile: Callable[[ReconcileReport], Any] | None) + __init__(request: Sequence[Skill | SkillReference | str] | str | SkillSource, root: str | os.PathLike[str], store: Any, *, prune: bool, timeout: float, on_unavailable: OnUnavailable, debounce: float, on_reconcile: Callable[[ReconcileReport], Any] | None) __enter__() -> SkillWatcher __exit__(*_exc: Any) -> None close(timeout: float = 15.0) -> None @@ -81,5 +86,5 @@ StoreDiagnostics [class, @dataclass] payloads_ignored: int = 0 payloads_transferred: int = 0 skill_objects_received: int = 0 -watch_skills(skills: Sequence[Skill | SkillReference | str] | str, root: str | os.PathLike[str], *, prune: bool = True, timeout: float = 10.0, on_unavailable: OnUnavailable = 'keep', debounce: float = DEFAULT_DEBOUNCE_SECONDS, on_reconcile: Callable[[ReconcileReport], Any] | None = None) -> tuple[ReconcileReport, SkillWatcher] [async] -write_skills(skills: Sequence[Skill | SkillReference | str] | str, root: str | os.PathLike[str], *, prune: bool = True, timeout: float = 10.0, on_unavailable: OnUnavailable = 'keep') -> ReconcileReport [async] +watch_skills(skills: Sequence[Skill | SkillReference | str] | str | SkillSource, root: str | os.PathLike[str], *, prune: bool = True, timeout: float = 10.0, on_unavailable: OnUnavailable = 'keep', debounce: float = DEFAULT_DEBOUNCE_SECONDS, on_reconcile: Callable[[ReconcileReport], Any] | None = None) -> tuple[ReconcileReport, SkillWatcher] [async] +write_skills(skills: Sequence[Skill | SkillReference | str] | str | SkillSource, root: str | os.PathLike[str], *, prune: bool = True, timeout: float = 10.0, on_unavailable: OnUnavailable = 'keep') -> ReconcileReport [async] diff --git a/packages/client/src/launchdarkly_ai_server/experimental/skills.py b/packages/client/src/launchdarkly_ai_server/experimental/skills.py index 18a2c9c6..88dac528 100644 --- a/packages/client/src/launchdarkly_ai_server/experimental/skills.py +++ b/packages/client/src/launchdarkly_ai_server/experimental/skills.py @@ -3,8 +3,9 @@ Configure a store with ``set_skill_store``, then read verified skill content with ``get_skill``, ``get_skills`` or ``all_skills``, or write it to disk with -``write_skills``. These names may change in a minor release, so review the -changelog when you upgrade. +``write_skills``. To keep the skills an AI Config pins on disk as the config +changes, use ``watch_skills(config_skills(key, context), root)``. These names +may change in a minor release, so review the changelog when you upgrade. """ from ..skills import ( @@ -25,6 +26,7 @@ OnUnavailable, write_skills, ) +from ..skills_source import config_skills from ..skills_watch import SkillWatcher, watch_skills from ..types import ( ReconcileAction, @@ -34,6 +36,7 @@ SkillOutcome, SkillOutcomeReason, SkillReference, + SkillSource, ) __all__ = [ # noqa: RUF022 @@ -46,6 +49,9 @@ "StoreDiagnostics", "watch_skills", "SkillWatcher", + # a requested set read on every reconcile, following AI Config changes + "SkillSource", + "config_skills", # accessors "skill_refs", "get_skill", diff --git a/packages/client/src/launchdarkly_ai_server/skills.py b/packages/client/src/launchdarkly_ai_server/skills.py index c44b9ff7..5ddc1489 100644 --- a/packages/client/src/launchdarkly_ai_server/skills.py +++ b/packages/client/src/launchdarkly_ai_server/skills.py @@ -220,6 +220,11 @@ def skill_refs(config: AiConfigRep | None) -> list[SkillReference]: objects with a valid key and an integer version >= 1. This includes ``skills: null``. """ + return _skill_refs(config) + + +def _skill_refs(config: AiConfigRep | None) -> list[SkillReference]: + """``skill_refs`` for SDK code, which must not count as a caller's use.""" if not isinstance(config, dict): raise ValueError( f"skill_refs was given {type(config).__name__}, not an AI Config. " diff --git a/packages/client/src/launchdarkly_ai_server/skills_fs.py b/packages/client/src/launchdarkly_ai_server/skills_fs.py index 2d7bf29d..43c16c3a 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fs.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fs.py @@ -21,7 +21,9 @@ from __future__ import annotations +import asyncio import hashlib +import inspect import json import logging import math @@ -66,6 +68,7 @@ ReconcileReport, Skill, SkillReference, + SkillSource, ) from .types_validation import ( is_valid_skill_key, @@ -130,7 +133,7 @@ class _PendingWrite: async def write_skills( - skills: Sequence[Skill | SkillReference | str] | str, + skills: Sequence[Skill | SkillReference | str] | str | SkillSource, root: str | os.PathLike[str], *, prune: bool = True, @@ -140,9 +143,16 @@ async def write_skills( """ Materializes skills under a managed root at ``//SKILL.md``. - *skills* is a sequence of ``Skill`` / ``SkillReference`` / key strings, or - ``"*"`` for every skill the store holds. ``Skill`` values are written as-is; - references and keys resolve through the configured store. + *skills* is a sequence of ``Skill`` / ``SkillReference`` / key strings, + ``"*"`` for every skill the store holds, or a ``SkillSource`` such as + ``config_skills(key, context)``, whose ``refs`` is read once and treated as + the sequence it returns. ``Skill`` values are written as-is; references and + keys resolve through the configured store. + + A source that raises, or returns anything but a sequence of + ``SkillReference`` values and valid key strings, is an unavailable + retrieval: nothing is written or pruned, and ``on_unavailable`` decides + between a run-level error action and ``RuntimeError``. It is not retried. The reconcile is driven by a manifest, ``/.launchdarkly-skills.json``: it only overwrites or deletes files the manifest records as written by the @@ -150,9 +160,12 @@ async def write_skills( - ``prune``: remove previously managed skills that are no longer requested. This is how revocation takes effect. - - ``timeout``: seconds, non-negative and finite. Bounds retrieval, writes and - pruning; checked between steps, not mid-operation. The manifest rewrite - always runs. + - ``timeout``: seconds, non-negative and finite. Bounds reading a source, + retrieval, writes and pruning; checked between steps, not mid-operation. + An async source still pending at the deadline is cancelled; a + synchronous one cannot be interrupted, and a slow one uses up the time + left for everything after it. ``0`` never reads a source. The manifest + rewrite always runs. - ``on_unavailable``: ``"keep"`` reports content that could not be retrieved and leaves existing files alone; ``"raise"`` raises ``RuntimeError``. @@ -181,6 +194,13 @@ async def write_skills( deadline = time.monotonic() + timeout root_path = _resolve_root(root) + # Read before the root is pinned, so no descriptor is held across an await. + requested: Sequence[Skill | SkillReference | str] | str | _RetrievalBlocked = ( + await _read_source(skills, deadline) + if isinstance(skills, SkillSource) + else skills + ) + # Pinned once and held for the whole reconcile. None where the *at() family # is absent; the path-based checks are then the only defense. try: @@ -206,7 +226,12 @@ async def write_skills( # Run-level failure: there is no single skill key to hang it off. actions.append(_run_error(manifest_error)) - requests, incomplete = _resolve_requests(skills, deadline, on_unavailable) + if isinstance(requested, _RetrievalBlocked): + requests, incomplete = _unavailable_run(requested.reason, on_unavailable) + else: + requests, incomplete = _resolve_requests( + requested, deadline, on_unavailable + ) written, write_timed_out = _write_all( root_path, root_fd, requests, entries, deadline @@ -369,6 +394,73 @@ def _available_store(deadline: float, subject: str) -> SkillStore | _RetrievalBl return store +async def _read_source( + source: SkillSource, deadline: float +) -> list[SkillReference | str] | _RetrievalBlocked: + """ + Reads *source* once: its answer as a list, or why there is none. + + Every failure is a retrieval that failed before any key was known, never an + empty answer, which would prune every managed skill. + """ + remaining = deadline - time.monotonic() + if remaining <= 0: + return _RetrievalBlocked( + _unavailable( + "the timeout was exhausted before the skill source could be read" + ) + ) + try: + answer = source.refs() + if inspect.isawaitable(answer): + answer = await asyncio.wait_for(answer, remaining) + except TimeoutError: + return _RetrievalBlocked( + _unavailable( + f"the skill source did not answer within the {remaining:.3g}s left " + "of the timeout" + ) + ) + except Exception as exc: + return _RetrievalBlocked( + _unavailable(f"the skill source raised {type(exc).__name__}: {exc}") + ) + + rejection = _source_answer_rejection(answer) + if rejection is not None: + return _RetrievalBlocked(_unavailable(f"the skill source returned {rejection}")) + return list(answer) + + +def _source_answer_rejection(answer: Any) -> str | None: + """ + Why a source's answer is unusable, or ``None``. + + An invalid entry rejects the whole answer rather than being dropped: + dropping it would shorten the requested set and prune that skill. + """ + if isinstance(answer, (str, bytes)) or not isinstance(answer, Sequence): + return ( + f"{type(answer).__name__}, not a sequence of SkillReference values " + "and key strings" + ) + for index, item in enumerate(answer): + if isinstance(item, SkillReference): + if not is_valid_skill_key(item.key): + return f"an invalid skill key {item.key!r} at index {index}" + if not is_valid_skill_version(item.version): + return f"an invalid version {item.version!r} for '{item.key}'" + elif isinstance(item, str): + if not is_valid_skill_key(item): + return f"an invalid skill key {item!r} at index {index}" + else: + return ( + f"{type(item).__name__} at index {index}; a source answers " + "SkillReference values and key strings only" + ) + return None + + def _resolve_requests( skills: Sequence[Skill | SkillReference | str] | str, deadline: float, diff --git a/packages/client/src/launchdarkly_ai_server/skills_source.py b/packages/client/src/launchdarkly_ai_server/skills_source.py new file mode 100644 index 00000000..145b1741 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/skills_source.py @@ -0,0 +1,111 @@ +""" +Agent Skills — the skills an AI Config pins, read when they are needed. + +``config_skills(key, context)`` builds a ``SkillSource`` that evaluates one AI +Config for one context on every read. Passed to ``watch_skills``, it follows +the config: a version bump replaces the file on disk, and a skill the config +stops referencing is pruned. A list from ``skill_refs`` is fixed when it is +built, so a watcher fed one never sees the config change. + +Every way an evaluation can fail raises rather than answering ``[]``, because +``[]`` is the answer that prunes every managed skill. +""" + +from __future__ import annotations + +import inspect +from typing import Any + +from .lifecycle import get_client +from .sdk_usage import report_usage +from .skills import _skill_refs +from .types import LDContext, SkillReference, SkillSource +from .types_validation import parse_ai_config +from .utils import to_ld_context + + +def config_skills(key: str, context: LDContext) -> SkillSource: + """ + A ``SkillSource`` for the skills AI Config *key* pins for *context*. + + The documented form for keeping a config's skills on disk with + ``watch_skills``:: + + report, watcher = await watch_skills( + config_skills("support-agent", context), "/etc/agent/skills" + ) + + Each read evaluates the config afresh with the initialized client's + ``variation``, which records an evaluation event, as ``inspect_config`` + does: once per reconcile under a watcher, bounded by its debounce. A read + sends no AI telemetry. + + A read answers ``[]`` only when LaunchDarkly says there are no skills: the + variation is disabled, or the config has no ``skills`` field. It raises + ``RuntimeError`` naming *key* and the cause when no client is initialized + (it never initializes one), when ``variation`` raises or answers its + default, which is how a missing flag and a client that is not ready yet + both arrive, and when the variation or its ``skills`` field is malformed. + ``write_skills`` treats that as an unavailable retrieval and changes + nothing on disk. + + For several configs, or a client brought to ``init_client`` without a + ``flag_tracker``, build a ``SkillSource`` by hand. + """ + report_usage("experimental.configSkills") + + async def refs() -> list[SkillReference]: + return await _read_config_skills(key, context) + + return SkillSource(refs, config_keys=(key,)) + + +def _failed(key: str, cause: str) -> RuntimeError: + return RuntimeError( + f"config_skills could not read the skills of AI Config '{key}': {cause}" + ) + + +async def _read_config_skills(key: str, context: LDContext) -> list[SkillReference]: + try: + client = get_client() + except RuntimeError as exc: + raise _failed( + key, "no AI client is initialized; call init_client() first" + ) from exc + + try: + result = client.variation(key, to_ld_context(client, context), None) + value: Any = await result if inspect.isawaitable(result) else result + except Exception as exc: + raise _failed(key, f"variation raised {type(exc).__name__}: {exc}") from exc + + # The default must stay distinguishable from every real answer: anything + # that reads as disabled here would prune every skill. + if value is None: + raise _failed( + key, + "the evaluation returned its default, so the config was not found, " + "the client is not ready, or evaluation failed", + ) + if not isinstance(value, dict): + raise _failed( + key, f"the evaluation returned {type(value).__name__}, not an object" + ) + meta = value.get("_ldMeta") + if not isinstance(meta, dict) or not isinstance(meta.get("enabled"), bool): + raise _failed( + key, "the variation has no _ldMeta object carrying a boolean enabled" + ) + + if not meta["enabled"]: + # A disabled variation is an answer from LaunchDarkly: its skills leave. + return [] + + parsed = parse_ai_config({k: v for k, v in value.items() if k != "_ldMeta"}) + if not parsed.success: + raise _failed(key, f"the config failed to parse: {parsed.error.get('message')}") + try: + return _skill_refs(parsed.data) + except ValueError as exc: + raise _failed(key, str(exc)) from exc diff --git a/packages/client/src/launchdarkly_ai_server/skills_watch.py b/packages/client/src/launchdarkly_ai_server/skills_watch.py index 0283a2d3..b86088d5 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_watch.py +++ b/packages/client/src/launchdarkly_ai_server/skills_watch.py @@ -2,11 +2,14 @@ Agent Skills — keep skills on disk in sync as delivery changes. ``write_skills`` is a one-shot reconcile of what the store holds now. -``watch_skills`` re-runs it whenever the store reports a change. With ``"*"``, -a skill revoked in LaunchDarkly is removed from disk within a debounce interval -rather than at the next restart. An explicit list is fixed: a skill the store -answers ``absent`` for keeps its files and reports an error, and a config change -that unpins a skill or moves it to a new version is not seen. +``watch_skills`` re-runs it whenever the store reports a change and, for a +``SkillSource``, whenever one of its AI Configs changes. With ``"*"``, a skill +revoked in LaunchDarkly is removed from disk within a debounce interval rather +than at the next restart. With ``config_skills(key, context)``, a config moving +a skill to a new version replaces the file, and a skill it stops referencing is +pruned. An explicit list is fixed: a skill the store answers ``absent`` for +keeps its files and reports an error, and a config change that unpins a skill +or moves it to a new version is not seen. ``on_unavailable="keep"`` remains the default, so an outage never deletes the application's skill files. @@ -22,9 +25,10 @@ from collections.abc import Callable, Sequence from typing import Any +from .lifecycle import get_client from .skills_core import SKILL_OBJECT_KIND, get_store from .skills_fs import OnUnavailable, write_skills -from .types import ReconcileReport, Skill, SkillReference +from .types import ReconcileReport, Skill, SkillReference, SkillSource logger = logging.getLogger(__name__) @@ -47,7 +51,7 @@ class SkillWatcher: def __init__( self, - request: Sequence[Skill | SkillReference | str] | str, + request: Sequence[Skill | SkillReference | str] | str | SkillSource, root: str | os.PathLike[str], store: Any, *, @@ -81,6 +85,16 @@ def __init__( store.add_listener(SKILL_OBJECT_KIND, self.notify) self._registered = True + # Set by ``_follow_configs``; detached by ``close``. + self._config_keys: frozenset[str] = frozenset() + self._flag_tracker: Any = None + + def _follow_configs(self, flag_tracker: Any, config_keys: Sequence[str]) -> None: + """Subscribes to the base client's flag changes, filtered to *config_keys*.""" + self._config_keys = frozenset(config_keys) + flag_tracker.add_listener(self._on_flag_change) + self._flag_tracker = flag_tracker + def _start(self) -> None: """Starts the worker. Called once by ``watch_skills`` after the initial reconcile.""" self._thread.start() @@ -89,13 +103,27 @@ def _start(self) -> None: def notify(self, _raw: Any = None) -> None: """ - The store's change listener. Records that something changed; runs nothing. + Schedules a reconcile, through the same debounce as a delivery change. - Called on the delivery thread, so it must not block on filesystem work. - The argument is ignored: any change triggers a full reconcile. + Also the store's change listener. Call it yourself from your own + change listener when the watched ``SkillSource`` has no + ``config_keys``, or when the client cannot report config changes. Runs + nothing itself, so it is safe to call from any thread. The argument is + ignored: any change triggers a full reconcile. """ self._wake.set() + def _on_flag_change(self, change: Any) -> None: + """ + The base client's flag-change listener. + + Only signals. The base SDK calls it on the thread applying flag data, + while holding the store lock every evaluation takes, so reading the + source here would stall flag delivery and every ``variation`` call. + """ + if getattr(change, "key", None) in self._config_keys: + self._wake.set() + # -- the worker -------------------------------------------------------- def _run(self) -> None: @@ -174,9 +202,13 @@ def _detach(self) -> None: if not self._registered: return self._registered = False + flag_tracker, self._flag_tracker = self._flag_tracker, None remove_listener = getattr(self._store, "remove_listener", None) if callable(remove_listener): remove_listener(SKILL_OBJECT_KIND, self.notify) + remove_flag_listener = getattr(flag_tracker, "remove_listener", None) + if callable(remove_flag_listener): + remove_flag_listener(self._on_flag_change) def __enter__(self) -> SkillWatcher: return self @@ -186,7 +218,7 @@ def __exit__(self, *_exc: Any) -> None: async def watch_skills( - skills: Sequence[Skill | SkillReference | str] | str, + skills: Sequence[Skill | SkillReference | str] | str | SkillSource, root: str | os.PathLike[str], *, prune: bool = True, @@ -204,6 +236,28 @@ async def watch_skills( ``remove_listener``, a closed watcher stays registered with the store for the store's lifetime. + What reaches disk depends on the form of *skills*: + + - ``"*"`` follows the whole library: a deleted skill is pruned and a new + version is written. + - A fixed list never re-reads a config. A version it pins that leaves the + store stays on disk, reported ``error``, and the newer one is never + written. + - A ``SkillSource``, such as ``config_skills(key, context)``, is read again + on every reconcile, and the watcher also reconciles when one of its + ``config_keys`` changes in the base client. A version bump replaces the + file, and a skill the config stops referencing is pruned on the next + reconcile whose retrieval is available. A skill the store answers + ``absent`` for beside it keeps its own file and does not hold that prune; + a store that raises, has no initial data, or runs out of time does. + + Following config changes needs the initialized base LaunchDarkly client's + ``flag_tracker``; it is never read for a fixed list or ``"*"``. Under daemon + mode (``use_ldd`` or ``ldclient.datasystem.daemon``) the base client has no + data source, so its ``flag_tracker`` never fires: the watcher follows + skill-store changes only. Call ``watcher.notify()`` yourself if you can + detect config changes another way. + Args: debounce: Seconds to wait after a change before reconciling. Must be a non-negative finite number. @@ -222,6 +276,9 @@ async def watch_skills( Raises: RuntimeError: If no store is configured, or the store has no ``add_listener`` (use ``write_skills`` for a one-shot reconcile). + For a ``SkillSource`` with ``config_keys``, also if no AI client is + initialized, the client has no ``flag_tracker``, or the client is + offline: such a watcher would never hear about a config change. ValueError: If *debounce* is negative, ``NaN``, or infinite. """ store = get_store() @@ -259,6 +316,10 @@ async def watch_skills( on_reconcile=on_reconcile, ) try: + # Subscribed before the initial reconcile, like the store listener, so + # a config change landing during it is followed by one more. + if isinstance(skills, SkillSource) and skills.config_keys: + watcher._follow_configs(_config_flag_tracker(), skills.config_keys) # Run on the caller's task, so a bad root raises to the caller. report = await write_skills( skills, root, prune=prune, timeout=timeout, on_unavailable=on_unavailable @@ -271,3 +332,48 @@ async def watch_skills( # Any change seen during the initial reconcile is already pending. watcher._start() return report, watcher + + +_NO_CONFIG_SIGNAL_REMEDY = ( + "Initialize the AI client with a base LaunchDarkly client " + "(init_client() with an SDK key, or init_client(client=LDClient(...))), or " + "build the SkillSource with config_keys=() and call watcher.notify() from " + "your own change listener." +) + + +def _config_flag_tracker() -> Any: + """ + The base client's ``flag_tracker``, or raises naming the remedy. + + A watcher that never hears about config changes looks exactly like one + whose config never changed, so a missing signal is an error, not a no-op. + """ + try: + client = get_client() + except RuntimeError: + raise RuntimeError( + "watch_skills cannot follow config changes for this SkillSource: no " + "AI client is initialized. " + _NO_CONFIG_SIGNAL_REMEDY + ) from None + flag_tracker = getattr(client, "flag_tracker", None) + if not callable(getattr(flag_tracker, "add_listener", None)): + raise RuntimeError( + "watch_skills cannot follow config changes for this SkillSource: " + "the AI client has no flag_tracker.add_listener, so it cannot report " + "flag changes. " + _NO_CONFIG_SIGNAL_REMEDY + ) + # An offline client still has a flag_tracker, but no data source to fire it. + # One that raises is treated as offline, so the caller still gets the remedy. + is_offline = getattr(client, "is_offline", None) + try: + offline = callable(is_offline) and is_offline() is True + except Exception: + offline = True + if offline: + raise RuntimeError( + "watch_skills cannot follow config changes for this SkillSource: " + "the AI client is offline, so it never reports flag changes. " + + _NO_CONFIG_SIGNAL_REMEDY + ) + return flag_tracker diff --git a/packages/client/src/launchdarkly_ai_server/types.py b/packages/client/src/launchdarkly_ai_server/types.py index 45139325..80272220 100644 --- a/packages/client/src/launchdarkly_ai_server/types.py +++ b/packages/client/src/launchdarkly_ai_server/types.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import AsyncGenerator, Awaitable, Callable +from collections.abc import AsyncGenerator, Awaitable, Callable, Sequence from dataclasses import dataclass, field from typing import Any, Generic, Literal, TypeVar @@ -612,6 +612,71 @@ def errors(self) -> list[ReconcileAction]: return [a for a in self.actions if a.action == "error"] +@dataclass(frozen=True, init=False) +class SkillSource: + """ + A requested set of skills that is read when it is needed, not fixed when + the call is made. Immutable. + + ``write_skills`` reads it once per call; ``watch_skills`` reads it on every + reconcile, and re-reconciles when one of the AI Configs named in + ``config_keys`` changes. ``config_skills(key, context)`` builds the common + case, one AI Config evaluated for one context. + + ``refs`` raising, or answering anything other than a sequence of + ``SkillReference`` values and key strings, is a failed retrieval: nothing on + disk changes. Raise to say "I could not tell"; never return ``[]`` for it, + because ``[]`` prunes every skill the root manages. + """ + + refs: Callable[ + [], Sequence[SkillReference | str] | Awaitable[Sequence[SkillReference | str]] + ] + """Zero-argument callable, sync or async, returning the current references.""" + config_keys: tuple[str, ...] + """The AI Config keys whose changes should make a watcher read ``refs`` again.""" + + def __init__( + self, + refs: Callable[ + [], + Sequence[SkillReference | str] | Awaitable[Sequence[SkillReference | str]], + ], + config_keys: Sequence[str] = (), + ) -> None: + """ + Raises: + TypeError: If *refs* is not callable, or *config_keys* is a string + or holds anything but non-empty strings. A bare string would be + followed as one-character keys, none of which ever changes. + """ + if not callable(refs): + raise TypeError( + f"SkillSource refs must be callable, got {type(refs).__name__}" + ) + if isinstance(config_keys, (str, bytes)): + raise TypeError( + "SkillSource config_keys must be a sequence of config keys, not a " + f"single string; got {config_keys!r}. Pass ({config_keys!r},)." + ) + if not isinstance(config_keys, Sequence): + raise TypeError( + "SkillSource config_keys must be a sequence of config keys, got " + f"{type(config_keys).__name__}" + ) + keys = tuple(config_keys) + for key in keys: + if not isinstance(key, str): + raise TypeError( + "SkillSource config_keys must hold only strings, got " + f"{type(key).__name__}" + ) + if not key: + raise TypeError("SkillSource config_keys must not hold an empty key") + object.__setattr__(self, "refs", refs) + object.__setattr__(self, "config_keys", keys) + + # --------------------------------------------------------------------------- # Model / graph options # --------------------------------------------------------------------------- diff --git a/packages/client/tests/test_sdk_usage.py b/packages/client/tests/test_sdk_usage.py index 5d3c1990..e9947f4c 100644 --- a/packages/client/tests/test_sdk_usage.py +++ b/packages/client/tests/test_sdk_usage.py @@ -22,6 +22,7 @@ shutdown, ) from launchdarkly_ai_server.experimental.evaluations import init_evaluations +from launchdarkly_ai_server.experimental.skills import config_skills CONTEXT = {"kind": "user", "key": "user-1"} HELPER = "client.buildJudgeTasks" @@ -152,6 +153,10 @@ async def _fn(*_args: Any, **_kwargs: Any) -> dict[str, Any]: create_handler(("OpenAI", "messages"), _fn) return None + if helper == "experimental.configSkills": + # Builds the source without reading it: the report is on the call. + config_skills("k", CONTEXT) + return None assert helper == "experimental.initEvaluations", helper try: init_evaluations() @@ -179,6 +184,7 @@ async def client() -> Any: "client.runJudge", "client.buildJudgeTasks", "client.createHandler", + "experimental.configSkills", "experimental.initEvaluations", ], ) diff --git a/packages/client/tests/test_skills.py b/packages/client/tests/test_skills.py index ebfe133a..7bfa6ad3 100644 --- a/packages/client/tests/test_skills.py +++ b/packages/client/tests/test_skills.py @@ -372,6 +372,8 @@ def test_requires_no_client_or_store(self, mock_ld_client: Any) -> None: "StoreDiagnostics", "watch_skills", "SkillWatcher", + "SkillSource", + "config_skills", "skill_refs", "get_skill", "get_skill_result", diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index b35c5d72..14cf2b0c 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -4377,6 +4377,7 @@ def test_nothing_in_the_feature_imports_it(self) -> None: "skills_core.py", "skills_fs.py", "skills_watch.py", + "skills_source.py", "safe_fs.py", "types.py", "types_validation.py", diff --git a/packages/client/tests/test_skills_source.py b/packages/client/tests/test_skills_source.py new file mode 100644 index 00000000..062ae8d8 --- /dev/null +++ b/packages/client/tests/test_skills_source.py @@ -0,0 +1,1321 @@ +""" +Tests for ``SkillSource`` and ``config_skills``, and for ``write_skills`` and +``watch_skills`` given a source. + +A fixed reference list never re-reads its config, so a watcher fed one cannot +see a config move ``pdf-extraction`` from v2 to v3: v2 stays on disk reported +``error`` and v3 is never written. A source is read on every reconcile, and the +watcher re-reconciles when one of its ``config_keys`` changes in the base client. + +The base client is a hand-built fake exposing ``variation``, ``track``, +``flag_tracker`` and ``is_offline``, never a ``MagicMock``: an unspecced mock +passes the watcher's ``flag_tracker`` probe and then never fires, which is the +silent failure the probe exists to catch. One test at the end drives the real +base SDK through ``TestDataV2``, because a fake encodes its author's belief about +the change payload. + +Every test writes only inside pytest's ``tmp_path``. The watcher runs a real +worker thread, so tests wait on its reconcile counter rather than on sleeps, +except to assert that something does *not* happen. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import json +import threading +import time +from collections.abc import Callable, Iterator +from pathlib import Path +from typing import Any + +import pytest + +import launchdarkly_ai_server.lifecycle as lifecycle_module +from launchdarkly_ai_server import skills as skills_module +from launchdarkly_ai_server import skills_fs, skills_source, skills_watch +from launchdarkly_ai_server.experimental.skills import ( + MANIFEST_FILENAME, + InMemorySkillStore, + ReconcileReport, + Skill, + SkillReference, + SkillSource, + config_skills, + set_skill_store, + watch_skills, + write_skills, +) +from launchdarkly_ai_server.sdk_usage import SDK_USAGE_EVENT, reset_sdk_usage +from launchdarkly_ai_server.skills_core import SKILL_OBJECT_KIND + +CONTEXT = {"kind": "user", "key": "user-1"} +CONFIG_KEY = "support-agent" + + +@pytest.fixture(autouse=True) +def _reset() -> Iterator[None]: + lifecycle_module._reset_for_testing() + reset_sdk_usage() + yield + lifecycle_module._reset_for_testing() + reset_sdk_usage() + + +def wait_until(predicate: Callable[[], bool], timeout: float = 10.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return predicate() + + +# --------------------------------------------------------------------------- +# Doubles +# --------------------------------------------------------------------------- + + +class FakeFlagChange: + """The shape of ``ldclient.interfaces.FlagChange``: a ``key`` property.""" + + def __init__(self, key: str) -> None: + self.key = key + + +class FakeFlagTracker: + def __init__(self) -> None: + self.listeners: list[Callable[[Any], None]] = [] + self.in_listener = False + self.firing_thread: int | None = None + + def add_listener(self, fn: Callable[[Any], None]) -> None: + self.listeners.append(fn) + + def remove_listener(self, fn: Callable[[Any], None]) -> None: + self.listeners.remove(fn) + + def fire(self, key: str) -> None: + """Calls every listener synchronously, as the base SDK's delivery thread does.""" + self.firing_thread = threading.get_ident() + self.in_listener = True + try: + for fn in list(self.listeners): + fn(FakeFlagChange(key)) + finally: + self.in_listener = False + + +class FakeLDClient: + """A base client with the surface the skills code touches, and no more.""" + + def __init__(self, value: Any = None, *, offline: bool = False) -> None: + self.value = value + self.raises: Exception | None = None + self.offline = offline + self.flag_tracker = FakeFlagTracker() + self.variation_calls: list[tuple[str, Any, Any]] = [] + self.tracked: list[tuple[str, Any]] = [] + + def variation(self, key: str, context: Any, default: Any) -> Any: + self.variation_calls.append((key, context, default)) + if self.raises is not None: + raise self.raises + return self.value + + def track( + self, event: str, context: Any, data: Any = None, metric: Any = None + ) -> None: + self.tracked.append((event, data)) + + def is_offline(self) -> bool: + return self.offline + + def flush(self) -> None: + return None + + def close(self) -> None: + return None + + def usage_helpers(self) -> list[str]: + return [ + data["helper"] for event, data in self.tracked if event == SDK_USAGE_EVENT + ] + + +class VariationOnlyClient: + """A client brought to ``init_client`` that implements only ``variation``.""" + + def variation(self, key: str, context: Any, default: Any) -> Any: + return None + + def track(self, *_args: Any, **_kwargs: Any) -> None: + return None + + +def ai_config( + *refs: tuple[str, int], enabled: bool = True, **fields: Any +) -> dict[str, Any]: + """A variation value as the base SDK returns it.""" + value: dict[str, Any] = { + "_ldMeta": {"enabled": enabled, "variationKey": "v1", "version": 1}, + "model": {"name": "gpt-test"}, + "provider": {"name": "openai"}, + "instructions": "Be helpful.", + } + if refs: + value["skills"] = [{"key": key, "version": version} for key, version in refs] + value.update(fields) + return value + + +class CountingSource: + """A hand-built source that records every read and where it ran.""" + + def __init__( + self, + answer: Any, + config_keys: tuple[str, ...] = (CONFIG_KEY,), + tracker: FakeFlagTracker | None = None, + ) -> None: + self.answer = answer + self.reads = 0 + self.threads: list[int] = [] + self.in_listener: list[bool] = [] + self._tracker = tracker + self.source = SkillSource(self.refs, config_keys=config_keys) + + def refs(self) -> Any: + self.reads += 1 + self.threads.append(threading.get_ident()) + self.in_listener.append(bool(self._tracker and self._tracker.in_listener)) + if isinstance(self.answer, Exception): + raise self.answer + return self.answer() if callable(self.answer) else self.answer + + +def skill_file(root: Path, key: str) -> Path: + return root / key / "SKILL.md" + + +def manifest_versions(root: Path) -> dict[str, int]: + entries = json.loads((root / MANIFEST_FILENAME).read_text())["entries"] + return {entry["key"]: entry["version"] for entry in entries.values()} + + +def summary(report: ReconcileReport) -> list[tuple[str, str, int | None]]: + return sorted((a.key, a.action, a.version) for a in report.actions) + + +def install(client: Any) -> Any: + lifecycle_module._set_client_for_testing(client) + return client + + +# --------------------------------------------------------------------------- +# SkillSource construction +# --------------------------------------------------------------------------- + + +class TestSkillSource: + def test_is_immutable(self) -> None: + source = SkillSource(list, config_keys=["a"]) + with pytest.raises(dataclasses.FrozenInstanceError): + source.config_keys = ("b",) # type: ignore[misc] + with pytest.raises(dataclasses.FrozenInstanceError): + source.refs = list # type: ignore[assignment] + + def test_config_keys_is_held_as_a_tuple_copy(self) -> None: + keys = ["support-agent", "triage"] + source = SkillSource(list, config_keys=keys) + keys.append("later") + assert source.config_keys == ("support-agent", "triage") + assert isinstance(source.config_keys, tuple) + + def test_config_keys_defaults_to_empty(self) -> None: + assert SkillSource(list).config_keys == () + + def test_a_bare_string_is_a_type_error(self) -> None: + """Iterated, it would be thirteen one-character keys that never change.""" + with pytest.raises(TypeError, match="single string"): + SkillSource(list, config_keys="support-agent") + + @pytest.mark.parametrize("refs", [None, "support-agent", [SkillReference("a", 1)]]) + def test_a_non_callable_refs_is_a_type_error(self, refs: Any) -> None: + with pytest.raises(TypeError, match="callable"): + SkillSource(refs) + + @pytest.mark.parametrize("keys", [[1], [None], ["a", b"b"]]) + def test_a_non_string_key_is_a_type_error(self, keys: Any) -> None: + with pytest.raises(TypeError, match="only strings"): + SkillSource(list, config_keys=keys) + + @pytest.mark.parametrize("keys", [{"a"}, iter(["a"]), 5]) + def test_a_non_sequence_is_a_type_error(self, keys: Any) -> None: + with pytest.raises(TypeError, match="sequence"): + SkillSource(list, config_keys=keys) + + def test_an_empty_key_is_a_type_error(self) -> None: + with pytest.raises(TypeError, match="empty"): + SkillSource(list, config_keys=["a", ""]) + + +# --------------------------------------------------------------------------- +# config_skills — the classification table +# --------------------------------------------------------------------------- + + +class TestConfigSkillsClassification: + """Only a disabled variation and a parsed config answer; everything else is a + failed read, raised rather than returned as ``[]``, which would prune.""" + + async def test_no_initialized_client_is_a_failed_read(self) -> None: + source = config_skills(CONFIG_KEY, CONTEXT) + with pytest.raises(RuntimeError, match=f"'{CONFIG_KEY}'.*no AI client"): + await source.refs() # type: ignore[misc] + + async def test_it_never_initializes_the_client( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Unlike ``inspect_config``, which initializes lazily from ``LD_SDK_KEY``.""" + monkeypatch.setenv("LD_SDK_KEY", "sdk-not-a-real-key") + source = config_skills(CONFIG_KEY, CONTEXT) + with pytest.raises(RuntimeError, match="no AI client"): + await source.refs() # type: ignore[misc] + assert lifecycle_module._client is None + + async def test_a_raising_variation_is_a_failed_read(self) -> None: + client = install(FakeLDClient()) + client.raises = ConnectionError("store lock poisoned") + source = config_skills(CONFIG_KEY, CONTEXT) + with pytest.raises(RuntimeError, match="ConnectionError: store lock poisoned"): + await source.refs() # type: ignore[misc] + + async def test_the_default_is_a_failed_read_and_the_default_is_none(self) -> None: + """Flag not found, client not ready and evaluation error all arrive as + the default, so it must be distinguishable from every real answer.""" + client = install(FakeLDClient(None)) + source = config_skills(CONFIG_KEY, CONTEXT) + with pytest.raises(RuntimeError, match="default"): + await source.refs() # type: ignore[misc] + assert client.variation_calls == [(CONFIG_KEY, CONTEXT, None)] + + @pytest.mark.parametrize("value", ["a string", 3, True, [ai_config()]]) + async def test_a_value_that_is_not_an_object_is_a_failed_read( + self, value: Any + ) -> None: + install(FakeLDClient(value)) + with pytest.raises(RuntimeError, match="not an object"): + await config_skills(CONFIG_KEY, CONTEXT).refs() # type: ignore[misc] + + @pytest.mark.parametrize( + "meta", + [ + "missing", + None, + "enabled", + {}, + {"enabled": "true"}, + {"enabled": 1}, + {"enabled": None}, + ], + ) + async def test_an_object_without_a_boolean_enabled_is_a_failed_read( + self, meta: Any + ) -> None: + value = ai_config(("pdf-extraction", 2)) + if meta == "missing": + del value["_ldMeta"] + else: + value["_ldMeta"] = meta + install(FakeLDClient(value)) + with pytest.raises(RuntimeError, match="_ldMeta"): + await config_skills(CONFIG_KEY, CONTEXT).refs() # type: ignore[misc] + + async def test_an_enabled_config_that_fails_to_parse_is_a_failed_read( + self, + ) -> None: + value = ai_config(("pdf-extraction", 2)) + del value["model"] + install(FakeLDClient(value)) + with pytest.raises(RuntimeError, match="failed to parse"): + await config_skills(CONFIG_KEY, CONTEXT).refs() # type: ignore[misc] + + @pytest.mark.parametrize( + "skills", + [ + None, + "pdf-extraction", + {"key": "pdf-extraction", "version": 2}, + [{"key": "pdf-extraction", "version": 2}, {"key": "Bad Key", "version": 1}], + [{"key": "pdf-extraction", "version": 0}], + [{"key": "pdf-extraction"}], + ], + ) + async def test_a_malformed_skills_field_is_a_failed_read(self, skills: Any) -> None: + """Never read as "no skills".""" + install(FakeLDClient(ai_config(skills=skills))) + with pytest.raises(RuntimeError, match="skills"): + await config_skills(CONFIG_KEY, CONTEXT).refs() # type: ignore[misc] + + async def test_a_disabled_variation_answers_empty(self) -> None: + """An answer from LaunchDarkly: its skills leave disk. Whatever else the + disabled variation carries is not read.""" + install(FakeLDClient(ai_config(("pdf-extraction", 2), enabled=False, skills=7))) + assert await config_skills(CONFIG_KEY, CONTEXT).refs() == [] # type: ignore[misc] + + async def test_a_parsed_config_answers_its_references(self) -> None: + install(FakeLDClient(ai_config(("pdf-extraction", 2), ("triage", 5)))) + assert await config_skills(CONFIG_KEY, CONTEXT).refs() == [ # type: ignore[misc] + SkillReference("pdf-extraction", 2), + SkillReference("triage", 5), + ] + + async def test_a_parsed_config_without_skills_answers_empty(self) -> None: + install(FakeLDClient(ai_config())) + assert await config_skills(CONFIG_KEY, CONTEXT).refs() == [] # type: ignore[misc] + + async def test_an_async_variation_is_awaited(self) -> None: + class AsyncClient(FakeLDClient): + async def variation(self, key: str, context: Any, default: Any) -> Any: # type: ignore[override] + self.variation_calls.append((key, context, default)) + return self.value + + install(AsyncClient(ai_config(("pdf-extraction", 2)))) + assert await config_skills(CONFIG_KEY, CONTEXT).refs() == [ # type: ignore[misc] + SkillReference("pdf-extraction", 2) + ] + + async def test_the_source_follows_exactly_its_one_key(self) -> None: + source = config_skills(CONFIG_KEY, CONTEXT) + assert source.config_keys == (CONFIG_KEY,) + + async def test_each_read_evaluates_afresh(self) -> None: + client = install(FakeLDClient(ai_config(("pdf-extraction", 2)))) + source = config_skills(CONFIG_KEY, CONTEXT) + first = await source.refs() # type: ignore[misc] + client.value = ai_config(("pdf-extraction", 3)) + second = await source.refs() # type: ignore[misc] + assert first == [SkillReference("pdf-extraction", 2)] + assert second == [SkillReference("pdf-extraction", 3)] + assert len(client.variation_calls) == 2 + + +class TestConfigSkillsTelemetry: + async def test_usage_is_reported_on_the_call_not_on_each_read( + self, recording_emitter: Any + ) -> None: + skills_module._set_emitter_for_testing(recording_emitter) + client = install(FakeLDClient(ai_config(("pdf-extraction", 2)))) + source = config_skills(CONFIG_KEY, CONTEXT) + assert client.usage_helpers() == ["experimental.configSkills"] + + for _ in range(3): + await source.refs() # type: ignore[misc] + + # Three evaluations, and nothing tracked but the one usage event: no + # $ld:ai:* telemetry, no experimental.skillRefs, no skills signal. + assert len(client.variation_calls) == 3 + assert client.tracked == [ + (SDK_USAGE_EVENT, client.tracked[0][1]), + ] + assert client.usage_helpers() == ["experimental.configSkills"] + assert recording_emitter.records == [] + + async def test_usage_is_reported_before_the_arguments_are_checked(self) -> None: + client = install(FakeLDClient()) + with pytest.raises(TypeError, match="empty"): + config_skills("", CONTEXT) + assert client.usage_helpers() == ["experimental.configSkills"] + + async def test_write_and_watch_given_a_config_source_report_no_skill_refs( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + store = InMemorySkillStore() + store.put(make_raw_skill(key="pdf-extraction", version=2)) + set_skill_store(store) + client = install(FakeLDClient(ai_config(("pdf-extraction", 2)))) + source = config_skills(CONFIG_KEY, CONTEXT) + + await write_skills(source, tmp_path / "a") + _report, watcher = await watch_skills(source, tmp_path / "b", debounce=0.01) + try: + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles >= 1) + finally: + watcher.close() + + assert len(client.variation_calls) == 3 + assert "experimental.skillRefs" not in client.usage_helpers() + assert client.usage_helpers().count("experimental.configSkills") == 1 + + +# --------------------------------------------------------------------------- +# write_skills given a source +# --------------------------------------------------------------------------- + + +MALFORMED_ANSWERS = [ + pytest.param(None, id="None"), + pytest.param("*", id="star"), + pytest.param("pdf-extraction", id="bare-key"), + pytest.param(b"pdf-extraction", id="bytes"), + pytest.param(3, id="int"), + pytest.param({"pdf-extraction": 2}, id="dict"), + pytest.param({SkillReference("a", 1)}, id="set"), + pytest.param([1], id="int-entry"), + pytest.param([SkillReference("a", 1), None], id="None-entry"), + pytest.param([{"key": "a", "version": 1}], id="dict-entry"), + pytest.param( + [Skill(key="a", version=1, content=b"x", content_hash="0" * 64)], + id="Skill-entry", + ), + pytest.param([SkillReference("Bad Key", 1)], id="invalid-reference-key"), + pytest.param([SkillReference("a", 0)], id="invalid-version"), + pytest.param([SkillReference("a", True)], id="bool-version"), # type: ignore[arg-type] + pytest.param(["Bad Key"], id="invalid-key-string"), +] + + +class TestWriteSkillsFromASource: + @pytest.fixture + def two_on_disk(self, tmp_path: Path, make_raw_skill: Any) -> Any: + """A store holding a v1 and b v2, and a root they were written to.""" + + async def make() -> tuple[InMemorySkillStore, Path]: + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="a body")) + store.put(make_raw_skill(key="b", version=2, content="b body")) + set_skill_store(store) + root = tmp_path / "skills" + report = await write_skills(["a", "b"], root) + assert report.ok + return store, root + + return make + + async def test_a_source_reconciles_exactly_as_its_answer_would( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + store = InMemorySkillStore() + for key, version in [("a", 1), ("b", 2), ("c", 1)]: + store.put(make_raw_skill(key=key, version=version, content=f"{key} body")) + set_skill_store(store) + answer = [SkillReference("a", 1), "b", SkillReference("missing", 1)] + + reports = [] + for root, skills in [ + (tmp_path / "list", answer), + (tmp_path / "source", SkillSource(lambda: answer)), + ]: + await write_skills(["a", "c"], root) + reports.append(await write_skills(skills, root)) + + assert summary(reports[0]) == summary(reports[1]) + # A real reconcile, not two empty ones. + assert ("missing", "error", None) in summary(reports[1]) + assert skill_file(tmp_path / "source", "b").read_text() == "b body" + + async def test_sync_async_and_tuple_answers_are_all_accepted( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="a body")) + set_skill_store(store) + + async def async_refs() -> list[SkillReference]: + await asyncio.sleep(0) + return [SkillReference("a", 1)] + + for name, refs in [ + ("sync", lambda: [SkillReference("a", 1)]), + ("async", async_refs), + ("tuple", lambda: ("a",)), + ]: + report = await write_skills(SkillSource(refs), tmp_path / name) + assert report.ok, name + assert skill_file(tmp_path / name, "a").read_text() == "a body" + + async def test_refs_is_read_exactly_once_even_when_it_raises( + self, tmp_path: Path + ) -> None: + set_skill_store(InMemorySkillStore()) + source = CountingSource(RuntimeError("config service down")) + report = await write_skills(source.source, tmp_path / "skills") + assert source.reads == 1 + assert [(a.key, a.action) for a in report.actions] == [("", "error")] + assert "config service down" in (report.actions[0].error or "") + assert report.ok is False + + @pytest.mark.parametrize("answer", MALFORMED_ANSWERS) + async def test_a_malformed_answer_is_an_unavailable_retrieval( + self, two_on_disk: Any, answer: Any + ) -> None: + _store, root = await two_on_disk() + report = await write_skills(SkillSource(lambda: answer), root) + + assert [(a.key, a.action) for a in report.actions] == [("", "error")] + assert "skill source" in (report.actions[0].error or "") + assert report.ok is False + assert skill_file(root, "a").read_text() == "a body" + assert skill_file(root, "b").read_text() == "b body" + assert manifest_versions(root) == {"a": 1, "b": 2} + + async def test_a_source_answering_none_after_two_writes_leaves_both( + self, two_on_disk: Any + ) -> None: + """The case that matters most: ``None`` must never read as "no skills".""" + _store, root = await two_on_disk() + report = await write_skills(SkillSource(lambda: None), root) + assert all(a.action != "removed" for a in report.actions) + assert manifest_versions(root) == {"a": 1, "b": 2} + + @pytest.mark.parametrize("answer", MALFORMED_ANSWERS) + async def test_raise_mode_names_the_skill_source( + self, two_on_disk: Any, answer: Any + ) -> None: + _store, root = await two_on_disk() + with pytest.raises(RuntimeError, match="skill source"): + await write_skills( + SkillSource(lambda: answer), root, on_unavailable="raise" + ) + assert manifest_versions(root) == {"a": 1, "b": 2} + + async def test_a_raising_source_in_raise_mode_names_the_skill_source( + self, two_on_disk: Any + ) -> None: + _store, root = await two_on_disk() + source = CountingSource(KeyError("nope")) + with pytest.raises(RuntimeError, match="skill source raised KeyError"): + await write_skills(source.source, root, on_unavailable="raise") + assert source.reads == 1 + + async def test_one_invalid_entry_fails_the_read_rather_than_being_dropped( + self, two_on_disk: Any, make_raw_skill: Any + ) -> None: + """Dropping it would shorten the requested set and prune b.""" + store, root = await two_on_disk() + store.put(make_raw_skill(key="c", version=1, content="c body")) + report = await write_skills( + SkillSource(lambda: [SkillReference("c", 1), {"key": "b"}]), root + ) + assert [(a.key, a.action) for a in report.actions] == [("", "error")] + assert not skill_file(root, "c").exists() + assert manifest_versions(root) == {"a": 1, "b": 2} + + async def test_a_zero_timeout_never_reads_the_source( + self, two_on_disk: Any + ) -> None: + _store, root = await two_on_disk() + source = CountingSource(["a"]) + report = await write_skills(source.source, root, timeout=0) + assert source.reads == 0 + assert [(a.key, a.action) for a in report.actions] == [("", "error")] + assert "timeout" in (report.actions[0].error or "") + + async def test_an_async_source_pending_at_the_deadline_is_cancelled( + self, two_on_disk: Any + ) -> None: + _store, root = await two_on_disk() + cancelled = asyncio.Event() + + async def hangs() -> list[str]: + try: + await asyncio.sleep(30) + except asyncio.CancelledError: + cancelled.set() + raise + return ["a"] + + started = time.monotonic() + report = await write_skills(SkillSource(hangs), root, timeout=0.2) + assert time.monotonic() - started < 5 + assert cancelled.is_set() + assert [(a.key, a.action) for a in report.actions] == [("", "error")] + assert "did not answer" in (report.actions[0].error or "") + assert manifest_versions(root) == {"a": 1, "b": 2} + + async def test_a_bad_root_raises_before_the_source_is_read( + self, tmp_path: Path + ) -> None: + set_skill_store(InMemorySkillStore()) + not_a_directory = tmp_path / "file" + not_a_directory.write_text("") + source = CountingSource(["a"]) + with pytest.raises(ValueError, match="not a directory"): + await write_skills(source.source, not_a_directory) + assert source.reads == 0 + + +# --------------------------------------------------------------------------- +# watch_skills following config changes +# --------------------------------------------------------------------------- + + +class TestFixedFormsNeverSubscribe: + @pytest.mark.parametrize("skills", ["*", ["a"]]) + async def test_a_fixed_list_and_star_register_no_config_listener( + self, tmp_path: Path, skills: Any + ) -> None: + set_skill_store(InMemorySkillStore()) + client = install(FakeLDClient()) + _report, watcher = await watch_skills(skills, tmp_path / "s", debounce=0.01) + watcher.close() + assert client.flag_tracker.listeners == [] + + @pytest.mark.parametrize("client", [None, VariationOnlyClient()]) + @pytest.mark.parametrize("skills", ["*", ["a"]]) + async def test_they_need_no_change_listener_api( + self, tmp_path: Path, skills: Any, client: Any + ) -> None: + set_skill_store(InMemorySkillStore()) + if client is not None: + install(client) + _report, watcher = await watch_skills(skills, tmp_path / "s", debounce=0.01) + watcher.close() + + +class TestNoChangeSignal: + """A watcher that never hears about config changes looks exactly like one + whose config never changed, so a missing signal raises.""" + + class NoAddListener: + def variation(self, *_args: Any) -> Any: + return None + + def track(self, *_args: Any, **_kwargs: Any) -> None: + return None + + flag_tracker = object() + + @pytest.mark.parametrize( + ("client", "cause"), + [ + (None, "no AI client is initialized"), + (VariationOnlyClient(), "no flag_tracker"), + (NoAddListener(), "no flag_tracker"), + (FakeLDClient(offline=True), "offline"), + ], + ids=["no-client", "variation-only", "tracker-without-add", "offline"], + ) + async def test_it_raises_naming_both_remedies_and_detaches( + self, tmp_path: Path, client: Any, cause: str + ) -> None: + store = InMemorySkillStore() + set_skill_store(store) + if client is not None: + install(client) + source = CountingSource(["a"]) + + with pytest.raises(RuntimeError) as raised: + await watch_skills(source.source, tmp_path / "s", debounce=0.01) + + message = str(raised.value) + assert cause in message + assert "LDClient" in message + assert "config_keys=()" in message + assert "watcher.notify()" in message + assert store._listeners.get(SKILL_OBJECT_KIND, []) == [] + if isinstance(client, FakeLDClient): + assert client.flag_tracker.listeners == [] + # Refused before the initial reconcile read anything. + assert source.reads == 0 + + async def test_an_is_offline_that_raises_is_treated_as_offline( + self, tmp_path: Path + ) -> None: + class RaisingOffline(FakeLDClient): + def is_offline(self) -> bool: + raise RuntimeError("config not loaded") + + store = InMemorySkillStore() + set_skill_store(store) + client = install(RaisingOffline()) + with pytest.raises(RuntimeError, match=r"offline.*watcher\.notify"): + await watch_skills(CountingSource(["a"]).source, tmp_path / "s") + assert store._listeners.get(SKILL_OBJECT_KIND, []) == [] + assert client.flag_tracker.listeners == [] + + async def test_empty_config_keys_probe_nothing(self, tmp_path: Path) -> None: + set_skill_store(InMemorySkillStore()) + source = CountingSource([], config_keys=()) + _report, watcher = await watch_skills(source.source, tmp_path / "s") + watcher.close() + assert source.reads == 1 + + async def test_notify_schedules_a_reconcile_through_the_debounce( + self, tmp_path: Path + ) -> None: + """The empty-``config_keys`` form depends on it.""" + set_skill_store(InMemorySkillStore()) + source = CountingSource([], config_keys=()) + _report, watcher = await watch_skills( + source.source, tmp_path / "s", debounce=0.3 + ) + try: + watcher.notify() + time.sleep(0.1) + assert watcher.reconciles == 0 # still inside the window + assert wait_until(lambda: watcher.reconciles == 1) + assert source.reads == 2 + finally: + watcher.close() + + +class TestFollowingConfigChanges: + async def _watch( + self, tmp_path: Path, source: SkillSource, debounce: float = 0.05, **kwargs: Any + ) -> tuple[ReconcileReport, Any]: + return await watch_skills(source, tmp_path / "s", debounce=debounce, **kwargs) + + async def test_every_trigger_reads_the_source_once( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + """A store change re-reads the source just as a config change does: the + two arrive on independent connections, so a store change is often when + the source's answer becomes resolvable.""" + store = InMemorySkillStore() + set_skill_store(store) + client = install(FakeLDClient()) + source = CountingSource(["a"]) + _report, watcher = await self._watch(tmp_path, source.source) + try: + assert source.reads == 1 + + store.put(make_raw_skill(key="a", version=1)) + assert wait_until(lambda: watcher.reconciles == 1) + assert source.reads == 2 + + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 2) + assert source.reads == 3 + + time.sleep(0.3) + assert (watcher.reconciles, source.reads) == (2, 3) + finally: + watcher.close() + + async def test_an_unrelated_key_does_not_wake_the_watcher( + self, tmp_path: Path + ) -> None: + set_skill_store(InMemorySkillStore()) + client = install(FakeLDClient()) + source = CountingSource(["a"]) + # A long window keeps the wake flag observable after a followed change. + _report, watcher = await self._watch(tmp_path, source.source, debounce=5) + try: + client.flag_tracker.fire("some-other-flag") + assert not watcher._wake.is_set() + # The positive control, before the worker can clear the flag: the + # listener is the same bound method the tracker holds. + watcher._stop.set() + client.flag_tracker.listeners[0](FakeFlagChange(CONFIG_KEY)) + assert watcher._wake.is_set() + finally: + watcher.close() + assert watcher.reconciles == 0 + + async def test_a_followed_key_wakes_it_and_an_unrelated_one_never_does( + self, tmp_path: Path + ) -> None: + set_skill_store(InMemorySkillStore()) + client = install(FakeLDClient()) + source = CountingSource(["a"]) + _report, watcher = await self._watch(tmp_path, source.source) + try: + for _ in range(3): + client.flag_tracker.fire("some-other-flag") + time.sleep(0.1) + assert watcher.reconciles == 0 + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 1) + finally: + watcher.close() + + async def test_the_listener_only_signals(self, tmp_path: Path) -> None: + """The base SDK calls it on its delivery thread holding the store lock + that every evaluation takes, so the source is read on the worker.""" + set_skill_store(InMemorySkillStore()) + client = install(FakeLDClient()) + source = CountingSource(["a"], tracker=client.flag_tracker) + _report, watcher = await self._watch(tmp_path, source.source, debounce=0.2) + try: + client.flag_tracker.fire(CONFIG_KEY) + assert source.reads == 1 # the listener returned without reading + assert wait_until(lambda: watcher.reconciles == 1) + assert source.reads == 2 + assert source.in_listener == [False, False] + assert source.threads[1] != client.flag_tracker.firing_thread + finally: + watcher.close() + + async def test_it_registers_before_the_initial_reconcile( + self, tmp_path: Path + ) -> None: + """A config change landing during the initial read is followed by one + more reconcile, not lost.""" + set_skill_store(InMemorySkillStore()) + client = install(FakeLDClient()) + + def changes_during_the_first_read() -> list[str]: + if source.reads == 1: + client.flag_tracker.fire(CONFIG_KEY) + return [] + + source = CountingSource(changes_during_the_first_read) + _report, watcher = await self._watch(tmp_path, source.source) + try: + assert wait_until(lambda: watcher.reconciles == 1) + assert source.reads == 2 + finally: + watcher.close() + + async def test_the_debounce_spans_both_trigger_kinds( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + """One window merges store and config notifications alike. + + Spread out so only the debounce can merge them, as in the store-only + coalescing test, and with the same result: the wake flag is cleared + before the window, so a change arriving inside it schedules exactly one + further pass. Every reconcile reads the source exactly once. + """ + store = InMemorySkillStore() + set_skill_store(store) + client = install(FakeLDClient()) + source = CountingSource(["a"]) + _report, watcher = await self._watch(tmp_path, source.source, debounce=0.6) + try: + for i in range(6): + if i % 2: + client.flag_tracker.fire(CONFIG_KEY) + else: + store.put(make_raw_skill(key=f"skill-{i}")) + time.sleep(0.1) + assert wait_until(lambda: watcher.reconciles >= 2) + time.sleep(1.5) + assert watcher.reconciles == 2 + assert source.reads == 1 + 2 + finally: + watcher.close() + + async def test_close_detaches_the_config_listener(self, tmp_path: Path) -> None: + set_skill_store(InMemorySkillStore()) + client = install(FakeLDClient()) + source = CountingSource(["a"]) + _report, watcher = await self._watch(tmp_path, source.source) + assert len(client.flag_tracker.listeners) == 1 + + watcher.close() + watcher.close() + + assert client.flag_tracker.listeners == [] + client.flag_tracker.fire(CONFIG_KEY) + time.sleep(0.3) + assert (watcher.reconciles, source.reads) == (0, 1) + + async def test_a_tracker_without_remove_listener_still_closes( + self, tmp_path: Path + ) -> None: + class AddOnlyTracker: + def __init__(self) -> None: + self.listeners: list[Any] = [] + + def add_listener(self, fn: Any) -> None: + self.listeners.append(fn) + + client = FakeLDClient() + client.flag_tracker = AddOnlyTracker() # type: ignore[assignment] + install(client) + set_skill_store(InMemorySkillStore()) + source = CountingSource(["a"]) + _report, watcher = await self._watch(tmp_path, source.source) + watcher.close() + for fn in client.flag_tracker.listeners: + fn(FakeFlagChange(CONFIG_KEY)) + time.sleep(0.3) + assert source.reads == 1 + + +class TestVersionBump: + """The store holds pdf-extraction v2, the config pins v2, and the initial + reconcile writes it. Either arrival order converges on v3, replaced in + place: no reconcile reports ``removed``.""" + + async def _start( + self, tmp_path: Path, make_raw_skill: Any + ) -> tuple[InMemorySkillStore, FakeLDClient, Path, Any, list[ReconcileReport]]: + store = InMemorySkillStore() + store.put(make_raw_skill(key="pdf-extraction", version=2, content="v2 body")) + set_skill_store(store) + client = install(FakeLDClient(ai_config(("pdf-extraction", 2)))) + root = tmp_path / "s" + reports: list[ReconcileReport] = [] + report, watcher = await watch_skills( + config_skills(CONFIG_KEY, CONTEXT), + root, + debounce=0.05, + on_reconcile=reports.append, + ) + reports.insert(0, report) + assert summary(report) == [("pdf-extraction", "written", 2)] + return store, client, root, watcher, reports + + @staticmethod + def _drop_v2(store: InMemorySkillStore) -> None: + del store._versions["pdf-extraction"][2] + + @staticmethod + def _end_state(root: Path) -> tuple[str, dict[str, int]]: + return skill_file(root, "pdf-extraction").read_text(), manifest_versions(root) + + async def test_config_first(self, tmp_path: Path, make_raw_skill: Any) -> None: + store, client, root, watcher, reports = await self._start( + tmp_path, make_raw_skill + ) + try: + client.value = ai_config(("pdf-extraction", 3)) + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 1) + assert [(a.key, a.action) for a in reports[1].actions] == [ + ("pdf-extraction", "error") + ] + assert self._end_state(root) == ("v2 body", {"pdf-extraction": 2}) + + store.put( + make_raw_skill(key="pdf-extraction", version=3, content="v3 body") + ) + assert wait_until(lambda: watcher.reconciles == 2) + assert summary(reports[2]) == [("pdf-extraction", "updated", 3)] + assert reports[2].ok + config_first = self._end_state(root) + finally: + watcher.close() + + assert config_first == ("v3 body", {"pdf-extraction": 3}) + assert all(a.action != "removed" for r in reports for a in r.actions) + + async def test_store_first(self, tmp_path: Path, make_raw_skill: Any) -> None: + store, client, root, watcher, reports = await self._start( + tmp_path, make_raw_skill + ) + try: + self._drop_v2(store) + store.put( + make_raw_skill(key="pdf-extraction", version=3, content="v3 body") + ) + assert wait_until(lambda: watcher.reconciles == 1) + assert [(a.key, a.action) for a in reports[1].actions] == [ + ("pdf-extraction", "error") + ] + assert self._end_state(root) == ("v2 body", {"pdf-extraction": 2}) + + client.value = ai_config(("pdf-extraction", 3)) + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 2) + assert summary(reports[2]) == [("pdf-extraction", "updated", 3)] + store_first = self._end_state(root) + finally: + watcher.close() + + # Identical to the config-first end state. + assert store_first == ("v3 body", {"pdf-extraction": 3}) + assert all(a.action != "removed" for r in reports for a in r.actions) + + +class PartlyUnreadableStore(InMemorySkillStore): + """A store whose reads raise for the keys in ``unreadable``: an unavailable + retrieval for those references, not an ``absent`` one.""" + + def __init__(self) -> None: + super().__init__() + self.unreadable: set[str] = set() + + def get_object( + self, kind: str, key: str, version: int | None = None + ) -> dict[str, Any] | None: + if key in self.unreadable: + raise ConnectionError(f"store cannot read {key}") + return super().get_object(kind, key, version) + + +class TestDroppingAReference: + """A dropped reference is pruned on the next reconcile whose retrieval is + available. An ``absent`` reference beside it is a per-key outcome and does + not hold the prune; only an unavailable retrieval does.""" + + async def _watch( + self, tmp_path: Path, make_raw_skill: Any + ) -> tuple[PartlyUnreadableStore, FakeLDClient, Path, Any, list[ReconcileReport]]: + store = PartlyUnreadableStore() + store.put(make_raw_skill(key="pdf-extraction", version=2)) + store.put(make_raw_skill(key="triage", version=1, content="triage body")) + set_skill_store(store) + client = install(FakeLDClient(ai_config(("pdf-extraction", 2), ("triage", 1)))) + reports: list[ReconcileReport] = [] + root = tmp_path / "s" + _report, watcher = await watch_skills( + config_skills(CONFIG_KEY, CONTEXT), + root, + debounce=0.05, + on_reconcile=reports.append, + ) + assert skill_file(root, "pdf-extraction").exists() + assert skill_file(root, "triage").exists() + return store, client, root, watcher, reports + + async def test_a_clean_drop_prunes_it( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + _store, client, root, watcher, reports = await self._watch( + tmp_path, make_raw_skill + ) + try: + client.value = ai_config(("triage", 1)) + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 1) + finally: + watcher.close() + assert summary(reports[0]) == [ + ("pdf-extraction", "removed", 2), + ("triage", "skipped_current", 1), + ] + assert not skill_file(root, "pdf-extraction").exists() + + async def test_an_absent_reference_beside_it_does_not_hold_the_prune( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + """The absent skill keeps its file under its own key, and the run is not + incomplete, so the dropped skill still goes in the same run.""" + store, client, root, watcher, reports = await self._watch( + tmp_path, make_raw_skill + ) + try: + del store._versions["triage"] + client.value = ai_config(("triage", 1)) + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 1) + finally: + watcher.close() + assert [(a.key, a.action) for a in reports[0].actions] == [ + ("triage", "error"), + ("pdf-extraction", "removed"), + ] + assert not skill_file(root, "pdf-extraction").exists() + assert skill_file(root, "triage").read_text() == "triage body" + assert manifest_versions(root) == {"triage": 1} + + async def test_an_unavailable_retrieval_holds_the_prune_until_it_is_available( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + store, client, root, watcher, reports = await self._watch( + tmp_path, make_raw_skill + ) + try: + store.unreadable.add("next-skill") + client.value = ai_config(("triage", 1), ("next-skill", 1)) + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 1) + assert all(a.action != "removed" for a in reports[0].actions) + assert ("next-skill", "error", None) in summary(reports[0]) + assert skill_file(root, "pdf-extraction").exists() + assert manifest_versions(root) == {"pdf-extraction": 2, "triage": 1} + + store.unreadable.clear() + store.put(make_raw_skill(key="next-skill", version=1)) + assert wait_until(lambda: watcher.reconciles == 2) + finally: + watcher.close() + assert summary(reports[1]) == [ + ("next-skill", "written", 1), + ("pdf-extraction", "removed", 2), + ("triage", "skipped_current", 1), + ] + assert not skill_file(root, "pdf-extraction").exists() + + +class TestASourceThatCannotAnswer: + async def test_answer_raise_answer_leaves_the_disk_alone_through_the_failure( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="a body")) + store.put(make_raw_skill(key="b", version=1, content="b body")) + set_skill_store(store) + client = install(FakeLDClient()) + source = CountingSource(["a"]) + reports: list[ReconcileReport] = [] + root = tmp_path / "s" + _report, watcher = await watch_skills( + source.source, root, debounce=0.05, on_reconcile=reports.append + ) + try: + source.answer = RuntimeError("config unavailable") + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 1) + # A 'keep' failure reaches on_reconcile, so a caller watching ok sees it. + assert reports[0].ok is False + assert [(a.key, a.action) for a in reports[0].actions] == [("", "error")] + assert skill_file(root, "a").read_text() == "a body" + assert manifest_versions(root) == {"a": 1} + + # The next trigger is the retry. + source.answer = ["b"] + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: watcher.reconciles == 2) + finally: + watcher.close() + assert reports[1].ok + assert summary(reports[1]) == [("a", "removed", 1), ("b", "written", 1)] + + async def test_raise_on_the_initial_reconcile_detaches_every_listener( + self, tmp_path: Path + ) -> None: + store = InMemorySkillStore() + set_skill_store(store) + client = install(FakeLDClient()) + source = CountingSource(None) + with pytest.raises(RuntimeError, match="skill source"): + await watch_skills(source.source, tmp_path / "s", on_unavailable="raise") + assert store._listeners.get(SKILL_OBJECT_KIND, []) == [] + assert client.flag_tracker.listeners == [] + + async def test_keep_on_the_initial_reconcile_returns_the_run_level_error( + self, tmp_path: Path + ) -> None: + set_skill_store(InMemorySkillStore()) + install(FakeLDClient()) + report, watcher = await watch_skills( + CountingSource(None).source, tmp_path / "s" + ) + watcher.close() + assert [(a.key, a.action) for a in report.actions] == [("", "error")] + + async def test_raise_on_a_later_reconcile_is_logged_and_the_watch_survives( + self, tmp_path: Path, make_raw_skill: Any, caplog: pytest.LogCaptureFixture + ) -> None: + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="a body")) + set_skill_store(store) + client = install(FakeLDClient()) + source = CountingSource(["a"]) + reports: list[ReconcileReport] = [] + _report, watcher = await watch_skills( + source.source, + tmp_path / "s", + on_unavailable="raise", + debounce=0.05, + on_reconcile=reports.append, + ) + try: + with caplog.at_level("ERROR"): + source.answer = RuntimeError("down") + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: source.reads == 2) + assert wait_until(lambda: "re-reconcile raised" in caplog.text) + source.answer = ["a"] + client.flag_tracker.fire(CONFIG_KEY) + assert wait_until(lambda: len(reports) == 1) + assert reports[0].ok + finally: + watcher.close() + + +# --------------------------------------------------------------------------- +# Module layout +# --------------------------------------------------------------------------- + + +class TestModuleLayout: + def test_only_the_source_and_the_watcher_import_lifecycle(self) -> None: + import inspect + + for module in (skills_module, skills_fs): + assert "lifecycle" not in inspect.getsource(module), module.__name__ + for module in (skills_source, skills_watch): + assert "from .lifecycle import get_client\n" in inspect.getsource(module) + + def test_write_skills_recognizes_a_source_without_importing_config_skills( + self, + ) -> None: + import inspect + + assert "skills_source" not in inspect.getsource(skills_fs) + + +# --------------------------------------------------------------------------- +# The real base SDK +# --------------------------------------------------------------------------- + + +class TestWithTheRealBaseSdk: + """A fake encodes its author's belief about the change payload, so one + version bump runs end to end through the base SDK's own test data source. + + It has to be ``TestDataV2`` registered as a synchronizer: v1 ``TestData`` + writes to the store without going through the update sink, so it fires no + change events, and an initializer alone delivers no updates. + """ + + async def test_a_version_bump_reaches_disk( + self, tmp_path: Path, make_raw_skill: Any + ) -> None: + import ldclient + import ldclient.datasystem + from ldclient import Config + from ldclient.integrations.test_datav2 import TestDataV2 + + def pin(version: int) -> Any: + return ( + td.flag(CONFIG_KEY) + .variations(ai_config(("pdf-extraction", version))) + .variation_for_all(0) + ) + + td = TestDataV2.data_source() + td.update(pin(2)) + datasystem = ldclient.datasystem.custom() + datasystem.initializers([td.builder]) + datasystem.synchronizers(td.builder) + ld_client = ldclient.LDClient( + Config( + "sdk-test-key", + datasystem_config=datasystem.build(), + send_events=False, + diagnostic_opt_out=True, + ) + ) + try: + assert ld_client.is_initialized() + install(ld_client) + store = InMemorySkillStore() + store.put( + make_raw_skill(key="pdf-extraction", version=2, content="v2 body") + ) + store.put( + make_raw_skill(key="pdf-extraction", version=3, content="v3 body") + ) + set_skill_store(store) + reports: list[ReconcileReport] = [] + root = tmp_path / "s" + + report, watcher = await watch_skills( + config_skills(CONFIG_KEY, CONTEXT), + root, + debounce=0.05, + on_reconcile=reports.append, + ) + try: + assert summary(report) == [("pdf-extraction", "written", 2)] + # Queued here; the listener fires later on the synchronizer thread. + td.update(pin(3)) + assert wait_until(lambda: watcher.reconciles >= 1) + finally: + watcher.close() + finally: + ld_client.close() + + assert summary(reports[-1]) == [("pdf-extraction", "updated", 3)] + assert skill_file(root, "pdf-extraction").read_text() == "v3 body" + assert manifest_versions(root) == {"pdf-extraction": 3}