diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 44d2b053..18b65a4c 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -125,6 +125,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 configured_skill_counts_by_agent, list_configured_skills_command @@ -2791,6 +2792,8 @@ def _launch_tool( # Claude re-adds an out-of-catalog saved model to /model even when built-ins are # replaced. Keep the managed catalog launch-scoped and leave the user's settings alone. state["_claude_launch_picker_models"] = picker_catalog.model_ids + if not skip_preflight: + refresh_downloaded_skills_on_launch(state) # Relayed = a Claude subscription: forward the model to Claude Code's own flag, like `-- --model X`. should_forward_relayed_model = ( tool == "claude" @@ -2817,8 +2820,8 @@ def _launch_tool( ) if recommendation is not None: _print_budget_panel(recommendation, tool, managed) - # The managed config's MCP servers and skills are both applied at `ug configure`, not here, - # so the launch hot path makes no per-launch discovery calls for them. + # The managed config's MCP servers and skills are applied at `ug configure`, not here. + # Downloaded skills get a rate-limited refresh above (refresh_downloaded_skills_on_launch). if tool == "claude": if provider: state["_claude_launch_provider"] = provider diff --git a/src/ucode/managed_config.py b/src/ucode/managed_config.py index fe2e88cf..886dbac7 100644 --- a/src/ucode/managed_config.py +++ b/src/ucode/managed_config.py @@ -37,6 +37,7 @@ fetch_model_recommendation, get_databricks_token, ) +from ucode.string_utils import parse_update_time from ucode.ui import console, print_warning MANAGED_CONFIG_PATH = config_io.APP_DIR / "managed-config.json" @@ -447,20 +448,6 @@ def managed_update_time(managed: dict | None) -> str | None: return _str(_as_dict(managed).get("update_time")) -def _parse_update_time(value: str | None) -> datetime | None: - if not value: - return None - try: - dt = datetime.fromisoformat(value.replace("Z", "+00:00")) - except ValueError: - return None - # An offset-less timestamp (e.g. a stub value) parses tz-naive; pin it to UTC so it can be - # compared against the tz-aware persisted watermark without raising. - if dt.tzinfo is None: - dt = dt.replace(tzinfo=UTC) - return dt - - def managed_config_is_newer(fetched: dict | None, applied_update_time: str | None) -> bool: """True when ``fetched`` is a newer version than the last one applied locally. @@ -468,8 +455,8 @@ def managed_config_is_newer(fetched: dict | None, applied_update_time: str | Non re-applies it rather than trusting possibly-stale local settings; no previously-applied watermark also counts as newer (the first apply). """ - fetched_ut = _parse_update_time(managed_update_time(fetched)) - applied_ut = _parse_update_time(applied_update_time) + fetched_ut = parse_update_time(managed_update_time(fetched)) + applied_ut = parse_update_time(applied_update_time) if fetched_ut is None or applied_ut is None: return True return fetched_ut > applied_ut @@ -731,7 +718,7 @@ def _cached_result_if_fresh(workspace: str) -> ManagedConfigResult | None: if data.get("workspace") != workspace: return None # Reuses the RFC-3339 parser the update-time watermark uses; None (missing/unparseable) is stale. - retrieved_at = _parse_update_time(_str(data.get("retrieved_at"))) + retrieved_at = parse_update_time(_str(data.get("retrieved_at"))) if retrieved_at is None: return None age = _utcnow() - retrieved_at diff --git a/src/ucode/skills_download.py b/src/ucode/skills_download.py index 0cd6fff1..b6d07a54 100644 --- a/src/ucode/skills_download.py +++ b/src/ucode/skills_download.py @@ -2,9 +2,11 @@ from __future__ import annotations +import os import shutil from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import UTC, datetime from pathlib import Path import questionary @@ -21,15 +23,20 @@ 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.string_utils import parse_update_time from ucode.ui import ( console, picker_style, @@ -423,6 +430,106 @@ def reconcile_managed_skills(managed: dict) -> tuple[list[str], list[str]]: return [ref.bundle_name for ref in installed], removed +# --- Launch-time refresh --------------------------------------------------- + + +def _eligible_launch_refresh_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 to re-download with its current skill: one whose UC source is newer than + its download, or one whose on-disk copy is only partly present and needs restoring to mirror UC. + + Resolves every record concurrently; one that no longer resolves (deleted, unfinalized, + unauthorized) is skipped, leaving its on-disk copy alone. Times are parsed before comparing + so the two RFC-3339 forms UC emits sort chronologically. A record with no parseable 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 + record = futures[future] + stored = parse_update_time(record.get("uc_update_time")) + current = parse_update_time(ref.uc_update_time) + is_newer = stored is None or (current is not None and current > stored) + if is_newer or _record_dirs_missing(record): + 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_bundles_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 entirely is forgotten; one only partly deleted is re-downloaded + to restore the mirror. 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 + set_last_update_check(now) + deleted, present = [], [] + for record in _eligible_launch_refresh_records(list_downloaded(), workspace): + (deleted if _record_dirs_all_missing(record) else present).append(record) + forget(deleted) + if present: + print_note("Checking Unity Catalog for downloaded skill updates...") + 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.") + 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: """Download every skill in each schema to disk and register the skills connection. @@ -537,6 +644,12 @@ def _record_dirs_missing(record: dict) -> bool: return any(not Path(directory).exists() for directory in record.get("dirs") or []) +def _record_dirs_all_missing(record: dict) -> bool: + """Whether every one of a record's on-disk directories no longer exists.""" + dirs = record.get("dirs") or [] + return bool(dirs) and all(not Path(directory).exists() for directory in dirs) + + def _download_label(record: dict) -> str: label = f"{record.get('fqn')} ({record.get('scope')}: {record.get('base')})" return f"{label} (missing)" if _record_dirs_missing(record) else label diff --git a/src/ucode/skills_state.py b/src/ucode/skills_state.py index b45a9c63..c46519c7 100644 --- a/src/ucode/skills_state.py +++ b/src/ucode/skills_state.py @@ -13,13 +13,18 @@ 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.string_utils import parse_update_time 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 +62,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 +72,43 @@ 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") + return parse_update_time(raw) if isinstance(raw, str) else 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.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + config_io.atomic_write_json(_skills_state_path(), manifest) def _norm(path: str) -> str: diff --git a/src/ucode/string_utils.py b/src/ucode/string_utils.py index 7d518769..14c97378 100644 --- a/src/ucode/string_utils.py +++ b/src/ucode/string_utils.py @@ -1,7 +1,25 @@ -"""Shared string validation helpers.""" +"""Shared string helpers.""" from __future__ import annotations +from datetime import UTC, datetime + + +def parse_update_time(value: str | None) -> datetime | None: + """Parse an RFC-3339 ``update_time`` into an aware UTC datetime, or None if absent/unparseable. + + UC serializes fractional seconds only when non-zero (protobuf JSON), so ``...:25Z`` and + ``...:25.400Z`` both occur; parsing before comparing avoids the wrong lexicographic ordering of + those two forms. An offset-less value is pinned to UTC. + """ + if not value: + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) + def is_valid_catalog_schema(value: str) -> bool: """Return whether value is a safe ``.`` reference.""" diff --git a/tests/test_skills_download.py b/tests/test_skills_download.py index f39d25ec..7edd5a75 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, @@ -1173,3 +1187,191 @@ 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_refresh_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"}, + # Mixed RFC-3339 precision: stored whole-second, UC edited to a sub-second later time. + {"fqn": "main.default.subsecond", "uc_update_time": "2026-06-26T05:58:25Z"}, + # Stored sub-second, UC now on an earlier whole second: must NOT be judged newer. + {"fqn": "main.default.roundup", "uc_update_time": "2026-06-26T05:58:25.400Z"}, + ] + 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, + "main.default.subsecond": _skill("subsecond", "2026-06-26T05:58:25.400Z"), + "main.default.roundup": _skill("roundup", "2026-06-26T05:58:25Z"), + } + 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", + "main.default.subsecond", + } + + 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_is_silent_and_skips_token(self, monkeypatch): + monkeypatch.setattr(sd, "list_downloaded", list) + monkeypatch.setattr(sd, "get_databricks_token", lambda *a, **k: pytest.fail("no token")) + notes: list[str] = [] + monkeypatch.setattr(sd, "print_note", notes.append) + + sd.refresh_downloaded_skills_on_launch({"workspace": WS}) + + assert skills_state.last_update_check() is not None + assert notes == [] # no "Checking..." line for users with no downloaded skills + + 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 any("boom" in note for note in notes) + assert skills_state.last_update_check() is not None + + 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_partially_deleted_skill_is_redownloaded_to_restore_mirror(self, tmp_path, monkeypatch): + home = tmp_path / "home" + _record_download(home, monkeypatch) + removed = home / ".agents/skills/triage" + (removed / "SKILL.md").unlink() + removed.rmdir() + monkeypatch.setattr(sd, "get_databricks_token", lambda *a, **k: "token") + monkeypatch.setattr( + sd, "get_skill", lambda ws, tok, fqn: _skill("triage", "2026-01-01T00:00:00Z") + ) + monkeypatch.setattr( + sd, + "_fetch_bundles", + lambda ws, tok, refs, label: {"main.default.triage": ({"SKILL.md": b"fresh"}, None)}, + ) + + sd.refresh_downloaded_skills_on_launch({"workspace": WS}) + + assert (home / ".claude/skills/triage/SKILL.md").read_bytes() == b"fresh" + assert (home / ".agents/skills/triage/SKILL.md").read_bytes() == b"fresh" + assert skills_state.list_downloaded()[0]["fqn"] == "main.default.triage" + + 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) + notes: list[str] = [] + monkeypatch.setattr(sd, "print_note", notes.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("Checking Unity Catalog" in note for note in notes) + 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 3ea12d5e..81abebd0 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, timezone from pathlib import Path from ucode import config_io, skills_state @@ -252,3 +253,50 @@ 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_reads_subsecond_form(self): + (config_io.APP_DIR / "skills.json").write_text( + json.dumps({"version": 1, "last_update_check": "2026-09-18T17:04:25.400Z"}) + ) + assert skills_state.last_update_check() == datetime( + 2026, 9, 18, 17, 4, 25, 400000, tzinfo=UTC + ) + + def test_normalizes_non_utc_stamp_to_utc(self): + when = datetime(2026, 9, 18, 22, 4, tzinfo=timezone(timedelta(hours=5))) + skills_state.set_last_update_check(when) + assert skills_state.last_update_check() == datetime(2026, 9, 18, 17, 4, tzinfo=UTC) + + 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"] diff --git a/tests/test_string_utils.py b/tests/test_string_utils.py index 9c369df8..18e2fe5b 100644 --- a/tests/test_string_utils.py +++ b/tests/test_string_utils.py @@ -1,8 +1,27 @@ -"""Tests for string validation helpers.""" +"""Tests for string helpers.""" + +from datetime import UTC, datetime import pytest -from ucode.string_utils import is_valid_catalog_schema +from ucode.string_utils import is_valid_catalog_schema, parse_update_time + + +class TestParseUpdateTime: + def test_whole_second_and_subsecond_forms_compare_chronologically(self): + earlier = parse_update_time("2026-06-26T05:58:25Z") + later = parse_update_time("2026-06-26T05:58:25.400Z") + assert earlier is not None and later is not None + assert later > earlier # the case a raw string compare gets wrong + + def test_offsetless_value_is_pinned_to_utc(self): + assert parse_update_time("2026-06-26T05:58:25") == datetime( + 2026, 6, 26, 5, 58, 25, tzinfo=UTC + ) + + @pytest.mark.parametrize("value", [None, "", "not-a-time"]) + def test_missing_or_unparseable_is_none(self, value): + assert parse_update_time(value) is None @pytest.mark.parametrize("value", ["system.ai", "main.default", "my-catalog.my_schema"])