diff --git a/src/ucode/cli.py b/src/ucode/cli.py index a7684e5fc..6e8219bd8 100644 --- a/src/ucode/cli.py +++ b/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) diff --git a/src/ucode/skills_download.py b/src/ucode/skills_download.py index 8dac9cc5f..4fc1508e2 100644 --- a/src/ucode/skills_download.py +++ b/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: diff --git a/src/ucode/skills_state.py b/src/ucode/skills_state.py index b45a9c63a..bb6a47670 100644 --- a/src/ucode/skills_state.py +++ b/src/ucode/skills_state.py @@ -13,6 +13,7 @@ import shutil import time from dataclasses import dataclass +from datetime import UTC, datetime, timedelta from pathlib import Path from ucode import config_io @@ -20,6 +21,9 @@ 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: diff --git a/tests/test_skills_download.py b/tests/test_skills_download.py index 015184012..2212c86fe 100644 --- a/tests/test_skills_download.py +++ b/tests/test_skills_download.py @@ -3,6 +3,9 @@ from __future__ import annotations +from datetime import UTC, datetime +from pathlib import Path + import pytest import ucode.skills_download as sd @@ -14,10 +17,21 @@ skill_dir_roots, write_skill, ) +from ucode.skills_state import SkillInstall WS = "https://example.databricks.com" +def _skill(securable_name: str, uc_update_time: str) -> SkillRef: + return SkillRef( + catalog="main", + schema="default", + securable_name=securable_name, + bundle_name=securable_name, + uc_update_time=uc_update_time, + ) + + def ref( securable_name: str, bundle_name: str | None = None, @@ -1064,3 +1078,152 @@ def test_managed_skill_is_not_offered_by_picker(self, tmp_path, monkeypatch): sd.remove_downloaded_skills_command([], path=None) assert [r["fqn"] for r in offered] == ["ml.prod.pii"] # managed skill withheld + + +class TestEligibleLaunchRecords: + def test_keeps_current_workspace_non_managed_downloads(self, tmp_path): + home, proj, other = tmp_path / "home", tmp_path / "proj", tmp_path / "other" + records = [ + {"fqn": "a.b.home", "workspace": WS, "scope": "user", "base": str(home)}, + {"fqn": "a.b.proj", "workspace": WS, "scope": "project", "base": str(proj)}, + {"fqn": "a.b.other", "workspace": WS, "scope": "project", "base": str(other)}, + {"fqn": "a.b.managed", "workspace": WS, "scope": "managed", "base": str(home)}, + {"fqn": "a.b.otherws", "workspace": "https://x", "scope": "user", "base": str(home)}, + ] + + eligible = sd._eligible_launch_records(records, WS) + + assert {r["fqn"] for r in eligible} == {"a.b.home", "a.b.proj", "a.b.other"} + + +class TestGetUpdatedRefs: + def test_flags_only_newer_or_unversioned(self, monkeypatch): + records = [ + {"fqn": "main.default.newer", "uc_update_time": "2026-01-01T00:00:00Z"}, + {"fqn": "main.default.same", "uc_update_time": "2026-01-01T00:00:00Z"}, + {"fqn": "main.default.older", "uc_update_time": "2026-06-01T00:00:00Z"}, + {"fqn": "main.default.unversioned"}, + {"fqn": "main.default.gone", "uc_update_time": "2026-01-01T00:00:00Z"}, + ] + current = { + "main.default.newer": _skill("newer", "2026-02-01T00:00:00Z"), + "main.default.same": _skill("same", "2026-01-01T00:00:00Z"), + "main.default.older": _skill("older", "2026-01-01T00:00:00Z"), + "main.default.unversioned": _skill("unversioned", "2026-01-01T00:00:00Z"), + "main.default.gone": None, + } + monkeypatch.setattr(sd, "get_skill", lambda ws, tok, fqn: current[fqn]) + + pairs = sd._get_updated_refs(WS, "token", records) + + assert {r["fqn"] for r, _ in pairs} == {"main.default.newer", "main.default.unversioned"} + + def test_empty_records_makes_no_pool(self, monkeypatch): + monkeypatch.setattr(sd, "get_skill", lambda *a: pytest.fail("should not fetch")) + assert sd._get_updated_refs(WS, "token", []) == [] + + +class TestUpdateStaleSkills: + def test_overwrites_both_roots_and_refreshes_record(self, tmp_path, monkeypatch): + home = tmp_path / "home" + monkeypatch.setattr(sd.Path, "home", classmethod(lambda cls: home)) + monkeypatch.setattr( + sd, + "_fetch_bundles", + lambda ws, tok, refs, label: {"main.default.triage": ({"SKILL.md": b"fresh"}, None)}, + ) + + updated = sd._update_stale_skills( + WS, "token", [({"base": str(home)}, _skill("triage", "2026-09-01T00:00:00Z"))] + ) + + assert updated == 1 + for family in (".claude/skills", ".agents/skills"): + assert (home / family / "triage" / "SKILL.md").read_bytes() == b"fresh" + stored = skills_state.list_downloaded() + assert stored[0]["fqn"] == "main.default.triage" + assert stored[0]["uc_update_time"] == "2026-09-01T00:00:00Z" + + +def _record_download(home, monkeypatch, *, uc_update_time="2026-01-01T00:00:00Z", on_disk=True): + """Record a home-scoped download of `triage`, writing its dirs when `on_disk`.""" + monkeypatch.setattr(sd.Path, "home", classmethod(lambda cls: home)) + dirs = tuple(str(home / family / "triage") for family in (".claude/skills", ".agents/skills")) + if on_disk: + for directory in dirs: + Path(directory).mkdir(parents=True) + (Path(directory) / "SKILL.md").write_bytes(b"old") + skills_state.record_downloads( + [ + SkillInstall( + fqn="main.default.triage", + bundle_name="triage", + workspace=WS, + scope="user", + base=str(home), + dirs=dirs, + uc_update_time=uc_update_time, + ) + ] + ) + + +class TestRefreshOnLaunch: + def test_rate_limited_skips_network(self, monkeypatch): + skills_state.set_last_update_check(datetime.now(UTC)) + monkeypatch.setattr(sd, "list_downloaded", lambda: pytest.fail("should not read")) + monkeypatch.setattr(sd, "get_databricks_token", lambda *a, **k: pytest.fail("no token")) + + sd.refresh_downloaded_skills_on_launch({"workspace": WS}) + + def test_no_eligible_records_stamps_and_skips_token(self, monkeypatch): + monkeypatch.setattr(sd, "list_downloaded", list) + monkeypatch.setattr(sd, "get_databricks_token", lambda *a, **k: pytest.fail("no token")) + + sd.refresh_downloaded_skills_on_launch({"workspace": WS}) + + assert skills_state.last_update_check() is not None + + def test_fail_open_reports_and_continues(self, monkeypatch): + def boom(): + raise RuntimeError("boom") + + monkeypatch.setattr(sd, "list_downloaded", boom) + notes: list[str] = [] + monkeypatch.setattr(sd, "print_note", notes.append) + + sd.refresh_downloaded_skills_on_launch({"workspace": WS}) + + assert notes and "boom" in notes[0] + + def test_manually_deleted_skill_is_forgotten_not_redownloaded(self, tmp_path, monkeypatch): + _record_download(tmp_path / "home", monkeypatch, on_disk=False) + monkeypatch.setattr(sd, "get_databricks_token", lambda *a, **k: pytest.fail("no token")) + monkeypatch.setattr(sd, "get_skill", lambda *a, **k: pytest.fail("no fetch")) + + sd.refresh_downloaded_skills_on_launch({"workspace": WS}) + + assert skills_state.list_downloaded() == [] + assert skills_state.last_update_check() is not None + + def test_updates_changed_skill_end_to_end(self, tmp_path, monkeypatch): + home = tmp_path / "home" + _record_download(home, monkeypatch) + monkeypatch.setattr(sd, "get_databricks_token", lambda *a, **k: "token") + monkeypatch.setattr( + sd, "get_skill", lambda ws, tok, fqn: _skill("triage", "2026-09-01T00:00:00Z") + ) + monkeypatch.setattr( + sd, + "_fetch_bundles", + lambda ws, tok, refs, label: {"main.default.triage": ({"SKILL.md": b"fresh"}, None)}, + ) + messages: list[str] = [] + monkeypatch.setattr(sd, "print_success", messages.append) + + sd.refresh_downloaded_skills_on_launch({"workspace": WS}) + + assert (home / ".claude/skills/triage/SKILL.md").read_bytes() == b"fresh" + assert skills_state.list_downloaded()[0]["uc_update_time"] == "2026-09-01T00:00:00Z" + assert skills_state.last_update_check() is not None + assert any("Updated 1" in m for m in messages) diff --git a/tests/test_skills_state.py b/tests/test_skills_state.py index 3ea12d5ef..08cca339a 100644 --- a/tests/test_skills_state.py +++ b/tests/test_skills_state.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +from datetime import UTC, datetime, timedelta from pathlib import Path from ucode import config_io, skills_state @@ -252,3 +253,37 @@ def boom(path): assert len(warnings) == 1 assert install.dirs[0] in warnings[0] assert skills_state.list_downloaded() == [] + + +class TestUpdateCheck: + def test_interval_matches_isaac(self): + assert skills_state.SKILL_UPDATE_CHECK_INTERVAL == timedelta(hours=24) + + def test_missing_reads_none(self): + assert skills_state.last_update_check() is None + + def test_round_trips(self): + when = datetime(2026, 9, 18, 17, 4, tzinfo=UTC) + skills_state.set_last_update_check(when) + assert skills_state.last_update_check() == when + + def test_malformed_reads_none(self): + (config_io.APP_DIR / "skills.json").write_text( + json.dumps({"version": 1, "last_update_check": "not-a-time"}) + ) + assert skills_state.last_update_check() is None + + def test_recording_downloads_preserves_stamp(self, tmp_path): + when = datetime(2026, 9, 18, 17, 4, tzinfo=UTC) + skills_state.set_last_update_check(when) + + skills_state.record_downloads([_install(tmp_path, "main.default.triage", "triage")]) + + assert skills_state.last_update_check() == when + + def test_stamping_preserves_downloads(self, tmp_path): + skills_state.record_downloads([_install(tmp_path, "main.default.triage", "triage")]) + + skills_state.set_last_update_check(datetime(2026, 9, 18, 17, 4, tzinfo=UTC)) + + assert [r["fqn"] for r in skills_state.list_downloaded()] == ["main.default.triage"]