Add ug skill status for downloaded and MCP skill state

src/ucode/cli.py

@@ -15,7 +15,7 @@
 from rich.panel import Panel
 from typer.core import TyperCommand
 
-from ucode import custom_oauth
+from ucode import custom_oauth, skills_status
 from ucode.agents import (
     TOOL_SPECS,
     LaunchOptions,
@@ -106,7 +106,6 @@
     remove_skills_command,
     remove_skills_locations_command,
     revert_mcp_configs,
-    skill_locations_for_client,
 )
 from ucode.skills_download import (
     configure_location_skills_download_command,
@@ -968,29 +967,26 @@ def status() -> int:
         console.print()
 
     print_heading("Skills")
-    skill_mcp_entry = next((s for s in mcp_servers if s.get("kind") == SKILLS_MCP_KIND), None)
-    if not skill_mcp_entry:
-        print_kv("Skills", "not configured")
+    skills = skills_status.collect(state)
+    print_kv("Downloaded skills", str(len(skills.downloaded)) if skills.downloaded else "none")
+    scopes = skills.mcp.by_agent
+    if not skills.mcp.configured:
+        print_kv("Skill MCP", "not configured")
+    elif agents_share_one_scope(scopes):
+        locations = next(iter(scopes.values()), [])
+        print_kv(
+            "Skill MCP Locations",
+            ", ".join(locations) if locations else "none — utility tools only",
+        )
+        configured_agents = [str(MCP_CLIENTS[client]["display"]) for client in scopes]
+        print_kv("Configured", ", ".join(configured_agents) if configured_agents else "none")
     else:
-        scopes = {
-            client: skill_locations_for_client(skill_mcp_entry, client)
-            for client in (skill_mcp_entry.get("clients") or [])
-            if client in MCP_CLIENTS
-        }
-        if agents_share_one_scope(scopes):
-            locations = next(iter(scopes.values()), [])
+        for client, locations in scopes.items():
             print_kv(
-                "Skill MCP Locations",
+                f"{MCP_CLIENTS[client]['display']} skill MCP locations",
                 ", ".join(locations) if locations else "none — utility tools only",
             )
-            configured_agents = [str(MCP_CLIENTS[client]["display"]) for client in scopes]
-            print_kv("Configured", ", ".join(configured_agents) if configured_agents else "none")
-        else:
-            for client, locations in scopes.items():
-                print_kv(
-                    f"{MCP_CLIENTS[client]['display']} skill MCP locations",
-                    ", ".join(locations) if locations else "none — utility tools only",
-                )
+    print_note("Run `ug skill status` (or `--json`) for downloaded skills and change commands.")
 
     print_heading("Tracing")
     tracing = state.get("tracing") or {}
@@ -1435,6 +1431,36 @@ def skills_remove(
         raise typer.Exit(130) from None
 
 
+@skill_app.command("status")
+def skills_status_cmd(
+    as_json: Annotated[
+        bool,
+        typer.Option("--json", help="Emit machine-readable JSON to stdout instead of a report."),
+    ] = False,
+    path: Annotated[
+        str | None,
+        typer.Option("--path", help="Limit downloaded skills to this base directory."),
+    ] = None,
+) -> None:
+    """Show configured skills: downloaded skills and the skills MCP scope.
+
+    Reports the exact ``ug skill add``/``ug skill remove`` inputs for each, so an agent can act on
+    the output without the interactive picker. Stale download records (skill dir deleted out of
+    band) are pruned as a side effect.
+    """
+    try:
+        status = skills_status.collect(load_state(), base=path)
+    except (RuntimeError, ValueError) as exc:
+        print_err(str(exc))
+        raise typer.Exit(1) from None
+    if as_json:
+        import json
+
+        print(json.dumps(skills_status.to_json(status), indent=2))
+    else:
+        skills_status.render(status)
+
+
 @app.command("mcp-proxy", hidden=True)
 def mcp_proxy_cmd(
     url: Annotated[

src/ucode/skills_download.py

@@ -22,6 +22,7 @@
 from ucode.skills_state import (
     SkillInstall,
     list_downloaded,
+    record_dirs_missing,
     record_downloads,
     records_for_fqns,
     records_for_schema,
@@ -443,14 +444,9 @@ def configure_skills_download_picker_command(path: str | None = None) -> int:
 # --- Removing and listing downloaded skills ---------------------------------
 
 
-def _record_dirs_missing(record: dict) -> bool:
-    """Whether any of a record's on-disk directories no longer exists."""
-    return any(not Path(directory).exists() for directory in record.get("dirs") or [])
-
-
 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
+    return f"{label}  (missing)" if record_dirs_missing(record) else label
 
 
 def _removal_choice(record: dict, index: int) -> questionary.Choice:

src/ucode/skills_state.py

@@ -169,6 +169,11 @@ def list_downloaded() -> list[dict]:
     return _load()
 
 
+def record_dirs_missing(record: dict) -> bool:
+    """Whether any of a record's on-disk directories no longer exists."""
+    return any(not Path(directory).exists() for directory in record.get("dirs") or [])
+
+
 def attribution_for_dir(path: str | Path) -> dict | None:
     """The install whose directories include ``path``, or None if unattributed."""
     target = _norm(str(path))

src/ucode/skills_status.py

@@ -0,0 +1,227 @@
+"""Read-side view of configured skills, backing ``ug skill status`` and the ``ug status`` summary.
+
+Skills reach a coding agent two independent ways: downloaded bundles (the ``~/.ucode/skills.json``
+manifest plus their on-disk dirs) and schemas attached to the skills MCP connection (``state.json``).
+:func:`collect` gathers both into one :class:`SkillStatus`, which :func:`render` prints for humans and
+:func:`to_json` emits for agents, so every surface reads the same data. Collecting also reconciles the
+manifest: a record whose on-disk directory is gone is pruned (see :func:`skills_state.forget`), so the
+listing only ever shows skills that are actually installed.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from ucode import skills_state
+from ucode.mcp import (
+    MCP_CLIENTS,
+    SKILLS_MCP_SERVER_NAME,
+    _skill_locations_by_client,
+    _skills_entry,
+    _skills_workspace,
+    agents_share_one_scope,
+)
+from ucode.ui import console, heading, print_heading, print_kv, print_note
+
+_DOWNLOAD_ADD = (
+    "Add:    ug skill add    --skills <fqn>  or  --location <catalog.schema>  [--path <base>]"
+)
+_DOWNLOAD_REMOVE = (
+    "Remove: ug skill remove --skills <fqn>  or  --location <catalog.schema>  [--path <base>]"
+)
+_MCP_ADD = "Add:    ug skill add    --mcp --location <catalog.schema>  [--agents <agent>]"
+_MCP_REMOVE = "Remove: ug skill remove --mcp --location <catalog.schema>  [--agents <agent>]"
+
+
+@dataclass(frozen=True)
+class DownloadedSkill:
+    fqn: str
+    schema: str
+    skill_dir: str
+    base: str
+    scope: str
+    dirs: tuple[str, ...]
+    workspace: str | None
+    workspace_id: str | None
+    downloaded_at: str | None
+    uc_update_time: str | None
+
+
+@dataclass(frozen=True)
+class McpScope:
+    configured: bool
+    server: str | None = None
+    workspace: str | None = None
+    by_agent: dict[str, list[str]] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class SkillStatus:
+    workspace: str | None
+    downloaded: list[DownloadedSkill]
+    pruned: list[str]
+    mcp: McpScope
+
+
+def collect(state: dict, base: str | None = None) -> SkillStatus:
+    """Assemble skill status from the manifest and MCP state, pruning stale download records.
+
+    ``base`` limits the downloaded listing to one download base, mirroring ``--path``; the MCP scope
+    is global and unaffected.
+    """
+    downloaded, pruned = _reconciled_downloads(base)
+    return SkillStatus(
+        workspace=state.get("workspace"),
+        downloaded=downloaded,
+        pruned=pruned,
+        mcp=_mcp_scope(state),
+    )
+
+
+def _reconciled_downloads(base: str | None) -> tuple[list[DownloadedSkill], list[str]]:
+    stale: list[dict] = []
+    live: list[dict] = []
+    for record in skills_state.list_downloaded():
+        (stale if skills_state.record_dirs_missing(record) else live).append(record)
+    skills_state.forget(stale)
+    pruned = sorted(str(record.get("fqn", "")) for record in stale)
+    skills = sorted(
+        (_downloaded_skill(record) for record in live if _under_base(record, base)),
+        key=lambda skill: (skill.fqn, skill.base),
+    )
+    return skills, pruned
+
+
+def _under_base(record: dict, base: str | None) -> bool:
+    return base is None or os.path.normpath(record.get("base", "")) == os.path.normpath(base)
+
+
+def _downloaded_skill(record: dict) -> DownloadedSkill:
+    fqn = str(record.get("fqn", ""))
+    return DownloadedSkill(
+        fqn=fqn,
+        schema=fqn.rsplit(".", 1)[0] if "." in fqn else fqn,
+        skill_dir=str(record.get("bundle_name", "")),
+        base=str(record.get("base", "")),
+        scope=str(record.get("scope", "")),
+        dirs=tuple(record.get("dirs") or []),
+        workspace=record.get("workspace"),
+        workspace_id=record.get("workspace_id"),
+        downloaded_at=record.get("downloaded_at"),
+        uc_update_time=record.get("uc_update_time"),
+    )
+
+
+def _mcp_scope(state: dict) -> McpScope:
+    entry = _skills_entry(list(state.get("mcp_servers") or []))
+    if entry is None:
+        return McpScope(configured=False)
+    return McpScope(
+        configured=True,
+        server=str(entry.get("name") or SKILLS_MCP_SERVER_NAME),
+        workspace=_skills_workspace(entry) or state.get("workspace"),
+        by_agent=_skill_locations_by_client(entry),
+    )
+
+
+def to_json(status: SkillStatus) -> dict:
+    return {
+        "workspace": status.workspace,
+        "downloaded": {
+            "count": len(status.downloaded),
+            "pruned": status.pruned,
+            "skills": [_skill_json(skill) for skill in status.downloaded],
+        },
+        "mcp": _mcp_json(status.mcp),
+    }
+
+
+def _skill_json(skill: DownloadedSkill) -> dict:
+    return {
+        "fqn": skill.fqn,
+        "schema": skill.schema,
+        "skill_dir": skill.skill_dir,
+        "base": skill.base,
+        "scope": skill.scope,
+        "dirs": list(skill.dirs),
+        "workspace": skill.workspace,
+        "workspace_id": skill.workspace_id,
+        "downloaded_at": skill.downloaded_at,
+        "uc_update_time": skill.uc_update_time,
+    }
+
+
+def _mcp_json(mcp: McpScope) -> dict:
+    if not mcp.configured:
+        return {"configured": False}
+    return {
+        "configured": True,
+        "server": mcp.server,
+        "workspace": mcp.workspace,
+        "by_agent": mcp.by_agent,
+    }
+
+
+def render(status: SkillStatus) -> None:
+    console.print(heading("ug skill status"))
+    _render_downloaded(status)
+    _render_mcp(status.mcp)
+
+
+def _render_downloaded(status: SkillStatus) -> None:
+    print_heading(f"Downloaded skills ({len(status.downloaded)})")
+    for base, scope, skills in _group_by_base(status.downloaded):
+        console.print(f"  Base {_base_label(base)} ({scope})")
+        for skill in skills:
+            foreign = skill.workspace and skill.workspace != status.workspace
+            origin = f"  [dim]· from[/dim] {skill.workspace}" if foreign else ""
+            console.print(
+                f"    {skill.fqn}  [dim]· skill dir[/dim] {skill.skill_dir}"
+                f"  [dim]· downloaded[/dim] {skill.downloaded_at or 'unknown'}{origin}"
+            )
+    if not status.downloaded:
+        console.print("  None downloaded.")
+    if status.pruned:
+        print_note(
+            f"Pruned {len(status.pruned)} stale record(s) whose skill dir was deleted: "
+            + ", ".join(status.pruned)
+        )
+    print_note(_DOWNLOAD_ADD)
+    if status.downloaded:
+        print_note(_DOWNLOAD_REMOVE)
+
+
+def _render_mcp(mcp: McpScope) -> None:
+    print_heading("Skill MCP scope")
+    if not mcp.configured:
+        console.print("  Not configured.")
+        print_note(_MCP_ADD)
+        return
+    print_kv("Server", mcp.server or SKILLS_MCP_SERVER_NAME)
+    if agents_share_one_scope(mcp.by_agent):
+        locations = next(iter(mcp.by_agent.values()), [])
+        agents = ", ".join(str(MCP_CLIENTS[client]["display"]) for client in mcp.by_agent)
+        print_kv("Configured", agents or "none")
+        print_kv("Schemas", ", ".join(locations) if locations else "none — utility tools only")
+    else:
+        for client, locations in mcp.by_agent.items():
+            print_kv(
+                str(MCP_CLIENTS[client]["display"]),
+                ", ".join(locations) if locations else "none — utility tools only",
+            )
+    print_note(_MCP_ADD)
+    print_note(_MCP_REMOVE)
+
+
+def _group_by_base(skills: list[DownloadedSkill]) -> list[tuple[str, str, list[DownloadedSkill]]]:
+    groups: dict[str, list[DownloadedSkill]] = {}
+    for skill in skills:
+        groups.setdefault(skill.base, []).append(skill)
+    ordered = sorted(groups.items(), key=lambda item: (item[1][0].scope != "user", item[0]))
+    return [(base, members[0].scope, members) for base, members in ordered]
+
+
+def _base_label(base: str) -> str:
+    return "~" if base == str(Path.home()) else base