[AIGTWY-4977] Cover managed skills in CUJ 2, 3 and 6

tests/README.md

@@ -327,7 +327,7 @@ These are unit/component checks; they do not establish live sudo password-prompt
 | Scenario | Status / requirement |
 | --- | --- |
 | Broad live MCP functionality | CUJ2 covers sandbox MCP configuration and generated client listings. Live sandbox execution is deferred because MAS cannot downscope the CI service principal. Installation tests cover the local web-search MCP handshake and tool listing, not upstream proxying or a real search request. Other MCP services, live parent/child search, and permission decisions remain deferred |
-| Dedicated CUJ Skill coverage | Deferred because of a backend storage-path issue; CUJ2 has no Skill configuration, download, invocation, or assertion path |
+| Dedicated CUJ Skill coverage | CUJ6 covers the named managed skill `ug_e2e.skills.fixture-summary`: download to both skill roots, an unchanged repeat configure, and removal from disk and from both agents' `/skills` once the selector is dropped, including after relaunch. Skill fixtures need a schema on customer S3 storage because Default Storage rejects skill uploads. CUJ2 checks that only the named skill downloads, byte-identical to the live fixture, is listed in `/skills` without its siblings, and is invoked by both agents. CUJ3 checks that every skill in `ug_e2e.skills` downloads and is listed, the `ug_e2e.other_skills` decoy stays absent, `ug status` counts the skills, and both agents invoke one |
 | Broad configure flags, multiple workspaces, and PAT flows | Deferred while focusing on basic CUJs |
 | Workspace-switch MCP cleanup | The `workspace_switch` CUJ covers real registration, cleanup, repeat configure, and a completed Claude task. Unit/component tests cover duplicate attempts and injected removal failures; the CUJ does not force an agent timeout. It runs in the required managed CI lane for full/live runs. |
 | Relayed/subscription MPS discovery | Not covered by the scoped discovery journeys |

tests/e2e_cuj/AGENTS.md

@@ -49,7 +49,8 @@ Model evidence combines observed gateway requests/responses with native complete
 records; neither routing banners nor native records alone prove an applied decision.
 
 `test_cuj2_mps_mcp.py` runs three independently collected cases against a permanently
-preconfigured workspace: configuration, Codex inference, and Claude inference. Each case
+preconfigured workspace: configuration, Codex inference, and Claude inference. `test_cuj2_skills.py`
+adds the named-skill cases against the same workspace. Each case
 reads and validates the exact two-agent MPS/MCP shape. No case creates, updates, deletes,
 or restores CodingAgentConfig. The fixture's final read-only equality check is the only
 workspace mutation guard. The test uses the shared `UserSession`, `Workspace`, and

tests/e2e_cuj/helpers/constants.py

@@ -41,3 +41,9 @@ class CodingAgent(StrEnum):
 )
 UC_MODEL_LOCATION_FIXTURE = ("ug_e2e.models", "ug_e2e.models.codex_primary")
 FIXTURE_READER_MCP_SERVICE_NAME = "ug_e2e.tools.fixture_reader"
+# UC skill names reject underscores, so the plan's `fixture_summary` is hyphenated.
+FIXTURE_SUMMARY_SKILL_NAME = "ug_e2e.skills.fixture-summary"
+FIXTURE_NOTES_SKILL_NAME = "ug_e2e.skills.fixture-notes"
+FIXTURE_CATALOG_LOOKUP_SKILL_NAME = "ug_e2e.skills.fixture-catalog-lookup"
+FIXTURE_DECOY_SKILL_NAME = "ug_e2e.other_skills.fixture-decoy"
+SKILLS_LOCATION = "ug_e2e.skills"

tests/e2e_cuj/helpers/skills.py

@@ -0,0 +1,94 @@
+"""Managed-skill checks shared by the CUJs that publish a skills selector."""
+
+from tests.integration.utils.evidence import assistant_answer_contains
+from tests.integration.utils.terminal import AgentTerminal
+
+from .constants import CLAUDE, CODEX, FIXTURE_SUMMARY_SKILL_NAME, INFERENCE_PATHS
+
+# Claude reads the first root and Codex the second; ug writes managed skills to both.
+SKILL_ROOTS = (".claude/skills", ".agents/skills")
+
+
+def skill_name(fqn: str) -> str:
+    """The securable leaf, which is also the bundle directory and the name agents list."""
+    return fqn.rsplit(".", 1)[1]
+
+
+FIXTURE_SUMMARY = skill_name(FIXTURE_SUMMARY_SKILL_NAME)
+# Only reference/fixture-facts.md holds the token, so a reply with it proves the bundle was read.
+FIXTURE_SUMMARY_TOKEN = "ugsum-4c1e9a"
+
+
+def skills_view(
+    session,
+    recorder,
+    agent: str,
+    listed: tuple[str, ...],
+    name: str | None = None,
+    codex_filter: str = "",
+) -> str:
+    """The agent's /skills screen once every ``listed`` skill renders, so absence checks are real.
+
+    Codex's skill popup shows eight rows, so ``codex_filter`` narrows it; it must also match any
+    skill the caller asserts is absent.
+    """
+    recorder.prepare_launch()
+    with AgentTerminal(
+        session, agent, [str(session.binary), agent], name or f"{agent}-skills"
+    ) as tui:
+        tui.boot()
+        tui.send("/skills", "type the /skills command")
+        tui.send("\r", "open the skills list")
+        if agent == CODEX:
+            tui.send("\r", "confirm the skills selection so the list renders")
+            if codex_filter:
+                tui.send(codex_filter, "filter the skills popup")
+        tui.wait_for(
+            lambda s: all(name in s for name in listed),
+            "the /skills view to list skills",
+            timeout=60,
+        )
+        return tui.visible
+
+
+def bundles_on_disk(session, root: str) -> set[str]:
+    return {path.name for path in (session.home / root).iterdir()}
+
+
+def bundle_files(session, root: str, bundle: str) -> dict[str, bytes]:
+    base = session.home / root / bundle
+    return {
+        str(path.relative_to(base)): path.read_bytes() for path in base.rglob("*") if path.is_file()
+    }
+
+
+def assert_fixture_summary_invoked(session, recorder, agent: str) -> None:
+    """Run a headless prompt that must load fixture-summary and answer from its reference file."""
+    recorder.prepare_launch()
+    # Codex's `$name` mention loads the skill explicitly; small models skip implicit skill matches.
+    skill = f"${FIXTURE_SUMMARY}" if agent == CODEX else f"the {FIXTURE_SUMMARY} skill"
+    prompt = (
+        f"Use {skill} and reply with only the one-line summary it produces. Do not use subagents."
+    )
+    checkpoint = recorder.checkpoint()
+    if agent == CLAUDE:
+        session.run(
+            CLAUDE,
+            "--",
+            "-p",
+            prompt,
+            "--output-format",
+            "json",
+            "--allowedTools",
+            "Skill,Read",
+            timeout=240,
+        )
+    else:
+        session.run(CODEX, "--", "exec", "--skip-git-repo-check", "--json", prompt, timeout=240)
+    assert assistant_answer_contains(session, agent, FIXTURE_SUMMARY_TOKEN), (
+        f"{agent} never answered with the token from the skill's reference file"
+    )
+    request = recorder.expect_request(
+        method="POST", path=INFERENCE_PATHS[agent], after=checkpoint, timeout=240
+    )
+    assert recorder.response_for(request, timeout=240).status_code == 200

tests/e2e_cuj/test_cuj2_mps_mcp.py

@@ -22,6 +22,7 @@
     CLAUDE,
     CODEX,
     CODING_AGENT_BY_CLI_NAME,
+    FIXTURE_SUMMARY_SKILL_NAME,
     INFERENCE_PATHS,
     MODEL_PROVIDER_SERVICE_FIXTURES,
     SANDBOX_MCP_SERVICE_NAME,
@@ -53,6 +54,7 @@ def _assert_cuj2_config(config: dict) -> None:
         "default_agent": CODING_AGENT_BY_CLI_NAME[CODEX],
         "enabled_agents": [expected_agent_config(CLAUDE), expected_agent_config(CODEX)],
         "mcp_servers": {"names": [SANDBOX_MCP_SERVICE_NAME]},
+        "skills": {"names": [FIXTURE_SUMMARY_SKILL_NAME]},
     }
     assert actual == expected, config
 

tests/e2e_cuj/test_cuj2_skills.py

@@ -0,0 +1,76 @@
+"""CUJ2 skills: the one named managed skill is downloaded, listed, and invoked by both agents."""
+
+from __future__ import annotations
+
+import pytest
+
+from ucode.skills_api import fetch_skill_bundle
+
+from .helpers.constants import (
+    CLAUDE,
+    CODEX,
+    FIXTURE_CATALOG_LOOKUP_SKILL_NAME,
+    FIXTURE_NOTES_SKILL_NAME,
+    SKILLS_LOCATION,
+)
+from .helpers.skills import (
+    FIXTURE_SUMMARY,
+    SKILL_ROOTS,
+    assert_fixture_summary_invoked,
+    bundle_files,
+    bundles_on_disk,
+    skill_name,
+    skills_view,
+)
+from .test_cuj2_mps_mcp import _assert_cuj2_config, _Cuj2Base
+
+OTHER_SKILLS = tuple(map(skill_name, (FIXTURE_NOTES_SKILL_NAME, FIXTURE_CATALOG_LOOKUP_SKILL_NAME)))
+
+
+class TestCuj2Skills(_Cuj2Base):
+    @pytest.fixture(scope="class")
+    def configured(self, cuj):
+        session, workspace, recorder = cuj
+        _assert_cuj2_config(workspace.config())
+        recorder.configure_session(
+            session, ["configure", "--skip-upgrade", "--disable-databricks-ai-tools"]
+        )
+        return cuj
+
+    def test_named_skill_matches_live_fixture_and_siblings_are_absent(self, configured):
+        """Scenario: configure with `skills.names` selecting only fixture-summary.
+
+        Expected: both roots hold exactly that bundle, byte-identical to the live fixture.
+        """
+        session, workspace, _ = configured
+        live, reason = fetch_skill_bundle(
+            workspace.url,
+            session.env["DATABRICKS_BEARER"],
+            *SKILLS_LOCATION.split("."),
+            FIXTURE_SUMMARY,
+        )
+        assert live and "reference/fixture-facts.md" in live, reason
+        for root in SKILL_ROOTS:
+            assert bundles_on_disk(session, root) == {FIXTURE_SUMMARY}
+            assert bundle_files(session, root, FIXTURE_SUMMARY) == live
+
+    @pytest.mark.parametrize("agent", [CLAUDE, CODEX])
+    def test_agent_lists_only_the_named_skill(self, configured, agent):
+        """Scenario: open /skills in each agent.
+
+        Expected: fixture-summary is listed and the schema's other skills are not.
+        """
+        session, _, recorder = configured
+        screen = skills_view(session, recorder, agent, (FIXTURE_SUMMARY,))
+        for other in OTHER_SKILLS:
+            assert other not in screen, f"{agent} /skills:\n{screen}"
+
+    @pytest.mark.parametrize("agent", [CLAUDE, CODEX])
+    def test_agent_invokes_the_named_skill(self, configured, agent):
+        """Scenario: ask each agent, headless, to use fixture-summary.
+
+        Expected: the reply carries the token found only in the bundle's reference file, after a
+        200 inference response through the recorder.
+        """
+        session, _, recorder = configured
+        assert_fixture_summary_invoked(session, recorder, agent)

tests/e2e_cuj/test_cuj3_skills.py

@@ -0,0 +1,101 @@
+"""CUJ3 skills: a `unity_catalog_location` downloads every skill in the schema and nothing else."""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+
+from .base import BaseCujTest
+from .helpers.constants import (
+    CLAUDE,
+    CODEX,
+    FIXTURE_CATALOG_LOOKUP_SKILL_NAME,
+    FIXTURE_DECOY_SKILL_NAME,
+    FIXTURE_NOTES_SKILL_NAME,
+    FIXTURE_SUMMARY_SKILL_NAME,
+    SKILLS_LOCATION,
+)
+from .helpers.skills import (
+    SKILL_ROOTS,
+    assert_fixture_summary_invoked,
+    bundle_files,
+    bundles_on_disk,
+    skill_name,
+    skills_view,
+)
+
+SCHEMA_SKILLS = tuple(
+    map(
+        skill_name,
+        (FIXTURE_SUMMARY_SKILL_NAME, FIXTURE_NOTES_SKILL_NAME, FIXTURE_CATALOG_LOOKUP_SKILL_NAME),
+    )
+)
+DECOY = skill_name(FIXTURE_DECOY_SKILL_NAME)
+STATUS_PANEL_TITLES = {CLAUDE: "Claude Code", CODEX: "Codex"}
+
+
+class TestCuj3Skills(BaseCujTest):
+    WORKSPACE_URL = "https://dbc-bbdd5508-648e.cloud.databricks.com"
+
+    @pytest.fixture(scope="class")
+    def configured(self, cuj):
+        session, workspace, recorder = cuj
+        assert workspace.config()["skills"] == {"unity_catalog_location": SKILLS_LOCATION}
+        session.configure(
+            [
+                "configure",
+                "--workspace",
+                workspace.url,
+                "--skip-upgrade",
+                "--disable-databricks-ai-tools",
+            ]
+        )
+        # `ug status` takes no workspace flag, so read it before retargeting to the recorder.
+        status = session.run("status", timeout=120).stdout
+        recorder.retarget_session(session.home)
+        return session, recorder, status
+
+    def test_every_schema_skill_is_on_disk_and_the_decoy_is_not(self, configured):
+        """Scenario: configure with `skills.unity_catalog_location` set to the skills schema.
+
+        Expected: both roots hold exactly the schema's three skills with their supporting files,
+        and nothing from the other schema.
+        """
+        session, _, _ = configured
+        for root in SKILL_ROOTS:
+            assert bundles_on_disk(session, root) == set(SCHEMA_SKILLS)
+            assert "reference/fixture-facts.md" in bundle_files(session, root, SCHEMA_SKILLS[0])
+            assert "scripts/lookup_schema.py" in bundle_files(session, root, SCHEMA_SKILLS[2])
+
+    @pytest.mark.parametrize("agent", [CLAUDE, CODEX])
+    def test_agent_lists_every_schema_skill_and_not_the_decoy(self, configured, agent):
+        """Scenario: open /skills in each agent.
+
+        Expected: all three schema skills are listed and the out-of-scope decoy is not.
+        """
+        session, recorder, _ = configured
+        screen = skills_view(session, recorder, agent, SCHEMA_SKILLS, codex_filter="fixture")
+        assert DECOY not in screen, f"{agent} /skills:\n{screen}"
+
+    @pytest.mark.parametrize("agent", [CLAUDE, CODEX])
+    def test_status_counts_the_skills(self, configured, agent):
+        """Scenario: run `ug status` after configure.
+
+        Expected: each agent's panel reports one Skills row counting the three schema skills.
+        """
+        *_, status = configured
+        panel = re.search(
+            rf"╭[─ ]*{STATUS_PANEL_TITLES[agent]}[─ ]*╮.*?Skills:\s+(\d+)", status, re.DOTALL
+        )
+        assert panel and int(panel.group(1)) == len(SCHEMA_SKILLS), status
+
+    @pytest.mark.parametrize("agent", [CLAUDE, CODEX])
+    def test_agent_invokes_a_downloaded_skill(self, configured, agent):
+        """Scenario: ask each agent, headless, to use fixture-summary.
+
+        Expected: the reply carries the token found only in the bundle's reference file, so
+        listing alone is not the pass condition.
+        """
+        session, recorder, _ = configured
+        assert_fixture_summary_invoked(session, recorder, agent)