Auto-update downloaded skills at agent launch

src/ucode/cli.py

@@ -121,6 +121,7 @@
     configure_selected_skills_download_command,
     configure_skills_download_picker_command,
     reconcile_managed_skills,
+    refresh_downloaded_skills_on_launch,
     remove_downloaded_skills_command,
 )
 from ucode.skills_list import list_configured_skills_command
@@ -2567,6 +2568,8 @@ def _launch_tool(
             user_pinned_model=model or forwarded_model,
             provider=provider,
         )
+        if not skip_preflight:
+            refresh_downloaded_skills_on_launch(state)
         print_success(f"Starting {TOOL_SPECS[tool]['display']}")
         with _managed_smart_routing_environment(managed, tool):
             launch_agent(tool, state, ctx.args, options=launch_options)

src/ucode/skills_download.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import os
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import UTC, datetime
 from pathlib import Path
 
 import questionary
@@ -20,13 +22,17 @@
     list_schema_skills,
 )
 from ucode.skills_state import (
+    SKILL_UPDATE_CHECK_INTERVAL,
     SkillInstall,
+    forget,
+    last_update_check,
     list_downloaded,
     record_downloads,
     records_for_fqns,
     records_for_schema,
     records_for_scope,
     remove_downloads,
+    set_last_update_check,
 )
 from ucode.state import load_state
 from ucode.ui import (
@@ -214,6 +220,29 @@ def _reject_bundle_name_collisions(refs: list[SkillRef]) -> list[SkillRef]:
     return kept
 
 
+def _fetch_and_write(
+    workspace: str, token: str, refs: list[SkillRef], roots: list[Path], *, label: str
+) -> list[SkillRef]:
+    """Fetch each ref's bundle concurrently, write it into ``roots``, and return those that
+    reached disk. A per-skill fetch failure or disk error warns and skips only that skill."""
+    if not refs:
+        return []
+    bundles = _fetch_bundles(workspace, token, refs, label=label)
+    written: list[SkillRef] = []
+    for ref in refs:
+        files, reason = bundles[ref.fqn]
+        if reason or files is None:
+            print_warning(f"Skipping `{ref.fqn}`: {reason}.")
+            continue
+        try:
+            write_skill(roots, ref, files)
+        except OSError as exc:
+            print_warning(f"Skipping `{ref.fqn}`: {exc}.")
+            continue
+        written.append(ref)
+    return written
+
+
 def _download_refs(
     workspace: str, token: str, refs: list[SkillRef], roots: list[Path], *, label: str
 ) -> tuple[list[SkillRef], int]:
@@ -222,22 +251,13 @@ def _download_refs(
     The shared download core: drop siblings claiming one directory
     (``_reject_bundle_name_collisions``), prompt before overwriting a skill already
     on disk (``should_download_skill``, so a declined skill is never fetched), then
-    fetch the survivors' bundles concurrently and write them. ``written`` are the
-    refs that reached disk, so a caller can record their attribution; ``total`` is
-    the count that could reach disk (dropped siblings excluded), so a caller's
-    summary denominator is right. A per-skill fetch failure warns and skips it.
+    fetch and write the survivors. ``written`` are the refs that reached disk, so a
+    caller can record their attribution; ``total`` is the count that could reach disk
+    (dropped siblings excluded), so a caller's summary denominator is right.
     """
     refs = _reject_bundle_name_collisions(refs)
     to_download = [ref for ref in refs if should_download_skill(roots, ref)]
-    bundles = _fetch_bundles(workspace, token, to_download, label=label)
-    written: list[SkillRef] = []
-    for ref in to_download:
-        files, reason = bundles[ref.fqn]
-        if reason or files is None:
-            print_warning(f"Skipping `{ref.fqn}`: {reason}.")
-            continue
-        write_skill(roots, ref, files)
-        written.append(ref)
+    written = _fetch_and_write(workspace, token, to_download, roots, label=label)
     console.print()
     return written, len(refs)
 
@@ -391,25 +411,105 @@ def reconcile_managed_skills(managed: dict) -> tuple[list[str], list[str]]:
             removed = [str(r["bundle_name"]) for r in stale if r.get("bundle_name")]
 
     missing = [ref for ref in refs if not existing_skill_on_disk(roots, ref.bundle_name)]
-    written: list[str] = []
-    if missing:
-        bundles = _fetch_bundles(workspace, token, missing, label="Fetching workspace skills")
-        installed: list[SkillRef] = []
-        for ref in missing:
-            files, reason = bundles[ref.fqn]
-            if reason or files is None:
-                print_warning(f"Skipping `{ref.fqn}`: {reason}.")
-                continue
-            try:
-                write_skill(roots, ref, files)
-            except OSError as exc:
-                # Best-effort per skill: a disk failure on one must not strand the rest.
-                print_warning(f"Skipping `{ref.fqn}`: {exc}.")
+    installed = _fetch_and_write(
+        workspace, token, missing, roots, label="Fetching workspace skills"
+    )
+    record_downloads(_skill_installs(installed, roots, None, workspace, scope="managed"))
+    return [ref.bundle_name for ref in installed], removed
+
+
+# --- Launch-time refresh ---------------------------------------------------
+
+
+def _eligible_launch_records(records: list[dict], workspace: str) -> list[dict]:
+    """The current workspace's own (non-managed) downloads, which a launch may refresh.
+
+    Managed skills are left to ``ug configure``, and other workspaces' downloads are skipped
+    because the launch token authenticates only this workspace.
+    """
+    return [
+        record
+        for record in records
+        if record.get("scope") != "managed" and record.get("workspace") == workspace
+    ]
+
+
+def _get_updated_refs(
+    workspace: str, token: str, records: list[dict]
+) -> list[tuple[dict, SkillRef]]:
+    """Pair each record whose UC source is newer than its download with the current skill.
+
+    Resolves every record concurrently; one that no longer resolves (deleted, unfinalized,
+    unauthorized) is skipped, leaving its on-disk copy alone. A record with no recorded
+    ``uc_update_time`` predates attribution, so it is refreshed once to backfill the field.
+    """
+    if not records:
+        return []
+    pairs: list[tuple[dict, SkillRef]] = []
+    with ThreadPoolExecutor(max_workers=min(_MAX_FETCH_WORKERS, len(records))) as pool:
+        futures = {pool.submit(get_skill, workspace, token, r["fqn"]): r for r in records}
+        for future in as_completed(futures):
+            ref = future.result()
+            if ref is None:
                 continue
-            installed.append(ref)
-            written.append(ref.bundle_name)
-        record_downloads(_skill_installs(installed, roots, None, workspace, scope="managed"))
-    return written, removed
+            record = futures[future]
+            stored = record.get("uc_update_time")
+            if stored is None or (ref.uc_update_time or "") > stored:
+                pairs.append((record, ref))
+    return pairs
+
+
+def _update_stale_skills(workspace: str, token: str, pairs: list[tuple[dict, SkillRef]]) -> int:
+    """Re-download each stale skill into its own base and refresh its manifest record.
+
+    Overwrites in place with no prompt, since the developer already chose to download these,
+    and only manifest-attributed directories are touched, so a user-authored skill of the same
+    name is never overwritten. Returns how many skills were rewritten.
+    """
+    home = os.path.normpath(str(Path.home()))
+    refs_by_base: dict[str, list[SkillRef]] = {}
+    for record, ref in pairs:
+        refs_by_base.setdefault(os.path.normpath(record["base"]), []).append(ref)
+
+    updated = 0
+    for base, refs in refs_by_base.items():
+        path = None if base == home else base
+        roots = skill_dir_roots(path)
+        written = _fetch_and_write(workspace, token, refs, roots, label="Updating skills")
+        record_downloads(_skill_installs(written, roots, path, workspace))
+        updated += len(written)
+    return updated
+
+
+def refresh_downloaded_skills_on_launch(state: dict) -> None:
+    """Update downloaded skills whose UC source changed, before an agent launches.
+
+    Rate-limited to once per ``SKILL_UPDATE_CHECK_INTERVAL`` via the manifest's
+    ``last_update_check`` stamp, so back-to-back launches make no network calls. A record
+    whose directories the user deleted is forgotten rather than re-downloaded. Best-effort:
+    any failure is reported and the launch proceeds on whatever is already on disk.
+    """
+    try:
+        now = datetime.now(UTC)
+        last = last_update_check()
+        if last is not None and now - last < SKILL_UPDATE_CHECK_INTERVAL:
+            return
+        workspace = state.get("workspace")
+        if not workspace:
+            return
+        deleted, present = [], []
+        for record in _eligible_launch_records(list_downloaded(), workspace):
+            (deleted if _record_dirs_missing(record) else present).append(record)
+        forget(deleted)
+        if present:
+            token = get_databricks_token(workspace, state.get("profile"))
+            pairs = _get_updated_refs(workspace, token, present)
+            updated = _update_stale_skills(workspace, token, pairs)
+            if updated:
+                print_success(f"Updated {updated} downloaded skill(s) from Unity Catalog.")
+        set_last_update_check(now)
+    except Exception as exc:  # noqa: BLE001 - a skill refresh must never block a launch
+        print_note(f"Skipped checking for skill updates: {exc}")
 
 
 def configure_location_skills_download_command(locations: list[str], *, path: str | None) -> int:

src/ucode/skills_state.py

@@ -13,13 +13,17 @@
 import shutil
 import time
 from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from ucode import config_io
 from ucode.ui import print_warning
 
 SKILLS_STATE_VERSION = 1
 
+# Matches Isaac's plugin marketplace staleness window (see plugin-marketplace/CLAUDE.md).
+SKILL_UPDATE_CHECK_INTERVAL = timedelta(hours=24)
+
 
 @dataclass(frozen=True)
 class SkillInstall:
@@ -57,8 +61,8 @@ def _quarantine_corrupt(path: Path) -> None:
         pass
 
 
-def _load() -> list[dict]:
-    """Records in the manifest; ``[]`` if it is absent, unreadable, or an unrecognized version.
+def _load_manifest() -> dict:
+    """The whole manifest; ``{}`` if it is absent, unreadable, or an unrecognized version.
 
     A file that fails to parse is quarantined (see ``_quarantine_corrupt``) rather than read as
     empty, so a single bad byte doesn't let the next write silently erase every tracked skill.
@@ -67,22 +71,48 @@ def _load() -> list[dict]:
     try:
         text = path.read_text(encoding="utf-8")
     except OSError:
-        return []
+        return {}
     try:
         data = json.loads(text)
     except json.JSONDecodeError:
         _quarantine_corrupt(path)
-        return []
+        return {}
     if not isinstance(data, dict) or data.get("version") != SKILLS_STATE_VERSION:
-        return []
-    downloads = data.get("skill_downloads")
+        return {}
+    return data
+
+
+def _load() -> list[dict]:
+    """The download records in the manifest, or ``[]`` when there are none."""
+    downloads = _load_manifest().get("skill_downloads")
     return [r for r in downloads if isinstance(r, dict)] if isinstance(downloads, list) else []
 
 
 def _save(downloads: list[dict]) -> None:
-    config_io.atomic_write_json(
-        _skills_state_path(), {"version": SKILLS_STATE_VERSION, "skill_downloads": downloads}
-    )
+    """Write the download records, preserving other manifest keys (e.g. ``last_update_check``)."""
+    manifest = _load_manifest()
+    manifest["version"] = SKILLS_STATE_VERSION
+    manifest["skill_downloads"] = downloads
+    config_io.atomic_write_json(_skills_state_path(), manifest)
+
+
+def last_update_check() -> datetime | None:
+    """When the launch-time update sweep last ran, or None if it never has."""
+    raw = _load_manifest().get("last_update_check")
+    if not isinstance(raw, str):
+        return None
+    try:
+        return datetime.strptime(raw, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=UTC)
+    except ValueError:
+        return None
+
+
+def set_last_update_check(when: datetime) -> None:
+    """Record when the launch-time update sweep last ran."""
+    manifest = _load_manifest()
+    manifest["version"] = SKILLS_STATE_VERSION
+    manifest["last_update_check"] = when.strftime("%Y-%m-%dT%H:%M:%SZ")
+    config_io.atomic_write_json(_skills_state_path(), manifest)
 
 
 def _norm(path: str) -> str: