feat(client): Agent Skills — materialize onto disk under a manifest (4/5)

packages/client/src/launchdarkly_ai_server/__init__.py

@@ -41,10 +41,20 @@
     InMemorySkillStore,
     all_skills,
     get_skill,
+    get_skill_result,
     get_skills,
     skill_refs,
 )
 from .skills_core import SkillStore
+from .skills_fdv2 import FDv2SkillStore, StoreDiagnostics
+from .skills_fs import (
+    MANIFEST_FILENAME,
+    MANIFEST_VERSION,
+    SKILL_FILENAME,
+    OnUnavailable,
+    write_skills,
+)
+from .skills_watch import SkillWatcher, watch_skills
 from .tracking import execute_and_stream, execute_and_track, wrap_tool_handlers
 from .types import (
     NATIVE_TOOL_KEY,
@@ -74,7 +84,12 @@
     ProviderGraphResponse,
     ProviderHandler,
     ProviderResponse,
+    ReconcileAction,
+    ReconcileActionKind,
+    ReconcileReport,
     Skill,
+    SkillOutcome,
+    SkillOutcomeReason,
     SkillReference,
     StreamChunkEvent,
     StreamDoneEvent,
@@ -136,7 +151,11 @@
     "ProviderGraphResponse",
     "ProviderHandler",
     "ProviderResponse",
+    "ReconcileAction",
+    "ReconcileActionKind",
+    "ReconcileReport",
     "Skill",
+    "SkillOutcome",
     "SkillReference",
     "StreamChunkEvent",
     "StreamDoneEvent",
@@ -214,8 +233,23 @@
     # skills
     "skill_refs",
     "get_skill",
+    "get_skill_result",
     "get_skills",
     "all_skills",
+    "write_skills",
     "SkillStore",
     "InMemorySkillStore",
+    # skills — the FDv2 delivery transport, and the eager re-reconcile it enables
+    "FDv2SkillStore",
+    "StoreDiagnostics",
+    "watch_skills",
+    "SkillWatcher",
+    # skills — the three closed-set unions a typed consumer needs to name
+    "ReconcileActionKind",
+    "OnUnavailable",
+    "SkillOutcomeReason",
+    # skills — on-disk constants, identical across languages
+    "SKILL_FILENAME",
+    "MANIFEST_FILENAME",
+    "MANIFEST_VERSION",
 ]

packages/client/src/launchdarkly_ai_server/safe_fs.py

@@ -11,12 +11,36 @@
 re-resolving a name — which is what closes the swap window rather than merely
 narrowing it. Where the platform has no ``*at()`` syscall family (Windows) the
 identical sequence runs against full paths, the per-component ``lstat`` floor.
+
+**Platform bound — this guarantee is POSIX-only, deliberately.** On POSIX the
+descriptor walk closes the swap window. On Windows it does not exist: there is no
+``*at()`` family, so the ``lstat`` floor is all that runs, and a floor is a
+check-then-use race rather than a closed window. The remedy would be
+reparse-point checks (``GetFileAttributesW``, or opening with
+``FILE_FLAG_OPEN_REPARSE_POINT``) and it is **not implemented, by decision rather
+than by oversight**: Windows is not a supported or tested platform for this
+release, and neither SDK repository has a Windows CI runner, so the checks would
+ship untested — and the TypeScript SDK could not match them in any case, because
+Node exposes no ``*at()`` family on *any* platform. Shipping them in Python alone
+would break the cross-language parity the two SDKs are held to and would trade a
+documented bound for an unverified one.
+
+Two consequences worth stating plainly rather than discovering later. First, on
+Windows write permission on the managed root is the *only* boundary, so the
+privilege-separated deployment the README documents is not advice there but the
+mitigation. Second, this bound retroactively lowers the priority of the Windows
+reserved-device-name work in ``skills_fs.py`` (``_WINDOWS_RESERVED_NAMES``): that
+code stays, because it is cheap and it keeps a managed root written on Linux
+usable when read from Windows, but it should not be read as evidence that Windows
+is a hardened target. It is not. Revisit both together if Windows becomes
+supported.
 """
 
 from __future__ import annotations
 
 import errno
 import os
+import re
 import secrets
 import stat
 import tempfile
@@ -191,6 +215,53 @@ def unlink_file(directory: Path, name: str, *, dir_fd: int | None) -> None:
     os.unlink(name, dir_fd=dir_fd)
 
 
+_TEMP_SUFFIX = ".tmp"
+"""Suffix on every temp file this module creates."""
+
+_TEMP_TOKEN_BYTES = 8
+"""Bytes of randomness in a temp name, as ``secrets.token_hex`` takes them."""
+
+_TEMP_TOKEN_PATTERN = re.compile(
+    # Two producers, one recognizer. The descriptor path below names its temp
+    # file with ``secrets.token_hex(_TEMP_TOKEN_BYTES)`` — twice that many
+    # lowercase hex characters. The fallback path hands naming to
+    # ``tempfile.mkstemp``, whose sequence is eight characters drawn from
+    # ``[a-z0-9_]``. Matched with ``fullmatch``, which anchors both branches at
+    # both ends, so nothing longer or otherwise-shaped is ever recognized.
+    rf"[0-9a-f]{{{_TEMP_TOKEN_BYTES * 2}}}|[a-z0-9_]{{8}}"
+)
+
+
+def temp_name_prefix(name: str) -> str:
+    """
+    The prefix every temp file for *name* is created under.
+
+    Spelled once because two callers need to agree on it: ``atomic_write``
+    creates the name, and a caller sweeping orphaned temp files left by a crash
+    has to recognize it. A copy of the format string in the sweeper would be a
+    copy that can drift out of step with the writer.
+    """
+    return f".{name}."
+
+
+def is_temp_name(candidate: str, name: str) -> bool:
+    """
+    Whether *candidate* is a name this module could have created for *name*.
+
+    The recognizer for the orphan sweep: ``atomic_write`` unlinks its temp file
+    on any exception, but a ``SIGKILL`` between the create and the rename leaves
+    it behind, and nothing else on disk records that it exists. Deliberately
+    narrow — prefix, random token, and suffix must all match, with nothing
+    before or after — because the only thing a caller does with a ``True`` here
+    is delete the file.
+    """
+    prefix = temp_name_prefix(name)
+    if not candidate.startswith(prefix) or not candidate.endswith(_TEMP_SUFFIX):
+        return False
+    token = candidate[len(prefix) : -len(_TEMP_SUFFIX)]
+    return _TEMP_TOKEN_PATTERN.fullmatch(token) is not None
+
+
 def _mkstemp_at(dir_fd: int, prefix: str) -> tuple[int, str]:
     """
     ``tempfile.mkstemp`` for a directory descriptor.
@@ -202,7 +273,7 @@ def _mkstemp_at(dir_fd: int, prefix: str) -> tuple[int, str]:
     """
     flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
     for _ in range(tempfile.TMP_MAX):
-        name = f"{prefix}{secrets.token_hex(8)}.tmp"
+        name = f"{prefix}{secrets.token_hex(_TEMP_TOKEN_BYTES)}{_TEMP_SUFFIX}"
         try:
             return os.open(name, flags, 0o600, dir_fd=dir_fd), name
         except FileExistsError:
@@ -234,7 +305,7 @@ def atomic_write(
     semantics on Windows).
     """
     at_fd = dir_fd if dir_fd is not None and SUPPORTS_DIR_FD else None
-    prefix = f".{name}."
+    prefix = temp_name_prefix(name)
     target: str | Path
 
     if at_fd is not None:
@@ -243,7 +314,7 @@ def atomic_write(
     else:
         # mkstemp opens with O_CREAT|O_EXCL, so an existing temp path is never
         # reused.
-        fd, temp = tempfile.mkstemp(dir=directory, prefix=prefix, suffix=".tmp")
+        fd, temp = tempfile.mkstemp(dir=directory, prefix=prefix, suffix=_TEMP_SUFFIX)
         target = directory / name
 
     try:

packages/client/src/launchdarkly_ai_server/skills.py

@@ -37,7 +37,7 @@
     resolve_from_store,
     verify_raw_skill,
 )
-from .types import AiConfigRep, Skill, SkillReference
+from .types import AiConfigRep, Skill, SkillOutcome, SkillReference
 from .types_validation import (
     is_valid_skill_key,
     is_valid_skill_version,
@@ -170,6 +170,22 @@ def add_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None:
         """
         self._listeners.setdefault(kind, []).append(fn)
 
+    def remove_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None:
+        """
+        Unregisters *fn* from *kind*, so a subsequent ``put`` no longer calls it.
+
+        Removes one occurrence: a callable registered twice must be removed twice.
+        Removing a callable that is not registered is a no-op, not an error, so a
+        consumer that detaches on close can do so unconditionally.
+        """
+        listeners = self._listeners.get(kind)
+        if listeners is None:
+            return
+        try:
+            listeners.remove(fn)
+        except ValueError:
+            return
+
 
 # ---------------------------------------------------------------------------
 # Reference discovery
@@ -186,8 +202,8 @@ def skill_refs(config: AiConfigRep | None) -> list[SkillReference]:
 
     A config that came through ``parse_ai_config`` never contains an invalid
     entry — parsing fails closed on one. A hand-built dict can, and a silently
-    shortened projection would leave a caller materializing a skill set it
-    believes is complete, so every dropped entry is logged.
+    shortened projection would let ``write_skills`` prune the dropped skill's
+    on-disk copy, so every dropped entry is logged.
     """
     if not isinstance(config, dict):
         return []
@@ -250,6 +266,40 @@ async def get_skill(key: str, *, version: int | None = None) -> Skill | None:
     return resolve_from_store(require_store(), key, version).skill
 
 
+async def get_skill_result(key: str, *, version: int | None = None) -> SkillOutcome:
+    """
+    Retrieves one verified skill, reporting *why* when there is none.
+
+    Same retrieval, same verification, same telemetry as ``get_skill`` — the two
+    differ only in what they report. ``get_skill`` collapses "no such skill",
+    "the store raised", "that is not the version held", and "the content failed
+    integrity verification" to one ``None``; this returns a ``SkillOutcome``
+    whose ``reason`` names which of them happened, so a caller can fail closed on
+    suspected tampering while tolerating a merely-absent skill:
+
+    ```python
+    outcome = await get_skill_result("pdf-extraction")
+    if outcome.reason == "integrity_failure":
+        raise SystemExit(f"refusing to run: {outcome.detail}")
+    if outcome.skill is not None:
+        print(outcome.skill.content)
+    ```
+
+    ``detail`` is human-readable and safe to surface — it names the key and the
+    failure mode, never any skill content or filesystem path. Branch on
+    ``reason``, not on ``detail``.
+
+    Emits nothing of its own: an integrity failure has already recorded its log
+    record and its signal inside verification, and recording a second here would
+    double-count one failure. Raises ``RuntimeError`` only when no skill store is
+    configured, exactly as ``get_skill`` does.
+    """
+    resolved = resolve_from_store(require_store(), key, version)
+    return SkillOutcome(
+        skill=resolved.skill, reason=resolved.reason, detail=resolved.error
+    )
+
+
 async def get_skills(refs: Sequence[SkillReference | str]) -> list[Skill]:
     """
     Retrieves a batch of verified skills.