fix(skills): resolve skill-name collisions deterministically (local > external) + tighten kanban liveness check

hermes_cli/kanban_db.py

@@ -5628,26 +5628,17 @@ def _kanban_worker_skill_available(hermes_home: Optional[str]) -> bool:
     worker before the agent loop runs. Gate the flag on actual resolvability;
     the kanban lifecycle contract is still injected via ``KANBAN_GUIDANCE``, so
     omitting the flag only drops the supplementary pattern library.
-    """
-    from pathlib import Path as _Path
 
-    # An unset HERMES_HOME means the worker falls back to the default root
-    # home (``~/.hermes``), which ships the bundled skill.
-    base = _Path(hermes_home) if hermes_home else (_Path.home() / ".hermes")
-    skills_root = base / "skills"
-    if not skills_root.is_dir():
-        return False
-    # Canonical bundled location first (cheap), then a bounded scan for
-    # profiles that have it nested elsewhere.
-    if (skills_root / "devops" / "kanban-worker" / "SKILL.md").is_file():
-        return True
-    try:
-        for skill_md in skills_root.rglob("kanban-worker/SKILL.md"):
-            if skill_md.is_file():
-                return True
-    except OSError:
-        pass
-    return False
+    Reuses ``_resolve_skill_under_home`` so the liveness check is exactly the
+    same lookup the spawned worker will perform — including a recursive scan
+    of ``<home>/skills`` and every ``skills.external_dirs`` entry declared in
+    the profile's config.yaml. The previous bespoke check only looked at the
+    canonical bundled path and a bounded rglob of ``<home>/skills``, missing
+    cases where the skill lived in an external dir (the profile pattern used
+    by ``braintrusteng``, which keeps a single shared skills tree under
+    ``~/.hermes/skills`` and points every profile's ``external_dirs`` at it).
+    """
+    return _resolve_skill_under_home("kanban-worker", hermes_home)
 
 
 def _worker_terminal_timeout_env(

tests/hermes_cli/test_kanban_db.py

@@ -2981,3 +2981,79 @@ def test_detect_stale_does_not_tick_failure_counter(kanban_home, monkeypatch):
         assert "stale" in kinds, (
             f"Expected 'stale' event in task_events; got {kinds!r}"
         )
+
+
+# ---------------------------------------------------------------------------
+# Skill liveness check (collision-resolution regression)
+# ---------------------------------------------------------------------------
+
+
+class TestKanbanWorkerSkillAvailable:
+    """Regression coverage for ``_kanban_worker_skill_available``.
+
+    The bare check used to look only at ``<home>/skills`` (canonical
+    bundled path + bounded rglob). That missed two real-world cases that
+    crashed worker spawn:
+
+      1. The skill lives under an ``external_dirs`` entry (the
+         ``braintrusteng`` profile pattern — every profile points at
+         ``~/.hermes/skills`` instead of duplicating the tree).
+      2. A stale per-profile copy collides with the external canonical
+         copy and the bare loader bails out as "ambiguous". This is now
+         resolved deterministically by ``skill_view`` (local wins, WARN
+         logged), so this helper must return True in that case too.
+
+    The helper now delegates to ``_resolve_skill_under_home``, which
+    walks the same dir set (``<home>/skills`` + ``skills.external_dirs``
+    from ``<home>/config.yaml``) as the spawned worker.
+    """
+
+    def _write_skill(self, root: Path, *, category: str, body: str = "stub") -> Path:
+        skill_dir = root / category / "kanban-worker"
+        skill_dir.mkdir(parents=True, exist_ok=True)
+        skill_md = skill_dir / "SKILL.md"
+        skill_md.write_text(
+            f"---\nname: kanban-worker\ndescription: stub\n---\n\n{body}\n",
+            encoding="utf-8",
+        )
+        return skill_md
+
+    def test_returns_true_when_skill_in_local_skills_dir(self, tmp_path):
+        home = tmp_path / ".hermes"
+        (home / "skills").mkdir(parents=True)
+        self._write_skill(home / "skills", category="devops")
+        assert kb._kanban_worker_skill_available(str(home)) is True
+
+    def test_returns_false_when_no_skill_anywhere(self, tmp_path):
+        home = tmp_path / ".hermes"
+        (home / "skills").mkdir(parents=True)
+        assert kb._kanban_worker_skill_available(str(home)) is False
+
+    def test_returns_true_when_skill_only_in_external_dir(self, tmp_path):
+        """Profile-with-external-dirs pattern: the profile's own
+        ``skills/`` is empty, but ``skills.external_dirs`` points at a
+        shared root that hosts the skill. The helper must see it."""
+        home = tmp_path / ".hermes"
+        (home / "skills").mkdir(parents=True)
+        shared = tmp_path / "shared-skills"
+        shared.mkdir()
+        self._write_skill(shared, category="devops")
+        (home / "config.yaml").write_text(
+            f"skills:\n  external_dirs:\n    - {shared}\n", encoding="utf-8",
+        )
+        assert kb._kanban_worker_skill_available(str(home)) is True
+
+    def test_returns_true_when_local_and_external_collide(self, tmp_path):
+        """Today's incident shape: stale per-profile copy (v2.0.0) +
+        canonical external copy (v2.2.0). The bare loader resolves the
+        collision deterministically; the helper must report True."""
+        home = tmp_path / ".hermes"
+        (home / "skills").mkdir(parents=True)
+        shared = tmp_path / "shared-skills"
+        shared.mkdir()
+        self._write_skill(home / "skills", category="devops", body="STALE V2.0.0")
+        self._write_skill(shared, category="devops", body="CANONICAL V2.2.0")
+        (home / "config.yaml").write_text(
+            f"skills:\n  external_dirs:\n    - {shared}\n", encoding="utf-8",
+        )
+        assert kb._kanban_worker_skill_available(str(home)) is True

tests/tools/test_skills_tool.py

@@ -1108,17 +1108,18 @@ class TestSkillViewCollisionDetection:
     """Regression tests for skill_view name collision handling.
 
     When a skill name resolves to multiple paths across the local skills
-    dir and external_dirs, skill_view must refuse to guess. Silent
-    shadowing — where ``/skills`` shows the local version but
-    ``skill_view`` loads the external one — is the bug class this guards
-    against. Reproduces with `skills.external_dirs` registered in
-    config.yaml and a same-name skill nested under a category locally.
-
-    Adapted from a regression suite originally proposed by @polkn in PR
-    #6136 (which used local-first precedence). The collision-refusal
-    behavior preserves the same protection without silently picking a
-    side, and gives the user an actionable hint (use the categorized
-    path) to recover.
+    dir and external_dirs, skill_view now resolves deterministically:
+    local SKILLS_DIR wins over external_dirs (which are ranked in
+    declaration order), and within a tier the most-recently modified
+    SKILL.md wins. The collision is logged at WARN so an operator can
+    spot and clean up the stale copy.
+
+    This replaces the previous "refuse and surface error" behavior, which
+    permanently crashed the worker preload path (``--skills <name>``)
+    whenever a stale per-profile copy collided with the shared canonical
+    copy under ``external_dirs``. There is no human at CLI startup to
+    disambiguate, so picking deterministically (and loudly) is strictly
+    better than refusing.
     """
 
     def _patch_dirs(self, local_dir, external_dirs):
@@ -1131,9 +1132,10 @@ def _patch_dirs(self, local_dir, external_dirs):
             ),
         )
 
-    def test_nested_local_collides_with_top_level_external(self, tmp_path):
-        """The original bug scenario: nested local + top-level external,
-        same name. Now refuses with both paths surfaced."""
+    def test_nested_local_wins_over_top_level_external(self, tmp_path, caplog):
+        """Stale or competing external skill of the same name does NOT
+        crash the loader; SKILLS_DIR wins by tier and a WARN log is
+        emitted naming the shadowed candidate."""
         local_dir = tmp_path / "local"
         external_dir = tmp_path / "external"
         local_dir.mkdir()
@@ -1148,22 +1150,21 @@ def test_nested_local_collides_with_top_level_external(self, tmp_path):
         _make_skill(external_dir, "explore-codebase", body="EXTERNAL VERSION")
 
         p1, p2 = self._patch_dirs(local_dir, [external_dir])
-        with p1, p2:
+        with caplog.at_level("WARNING", logger="tools.skills_tool"), p1, p2:
             raw = skill_view("explore-codebase")
 
         result = json.loads(raw)
-        assert result["success"] is False
-        assert "Ambiguous skill name 'explore-codebase'" in result["error"]
-        assert "matches" in result
-        assert len(result["matches"]) == 2
-        # Both paths surfaced
-        assert any("foundations/runtime" in p for p in result["matches"])
-        assert any("external" in p for p in result["matches"])
-        assert "hint" in result
-
-    def test_top_level_local_collides_with_external(self, tmp_path):
-        """Top-level local + top-level external with the same name also
-        refuses — same-name shadowing is ambiguous regardless of nesting."""
+        assert result["success"] is True
+        assert "LOCAL VERSION" in result["content"]
+        # Operator-facing breadcrumb: WARN log identifies the chosen path
+        # and the shadowed candidate.
+        warn_messages = [r.getMessage() for r in caplog.records if r.levelname == "WARNING"]
+        assert any("Skill name collision for 'explore-codebase'" in m for m in warn_messages)
+        assert any("external" in m for m in warn_messages)
+
+    def test_top_level_local_wins_over_external(self, tmp_path):
+        """Top-level local + top-level external with the same name —
+        local wins by tier."""
         local_dir = tmp_path / "local"
         external_dir = tmp_path / "external"
         local_dir.mkdir()
@@ -1177,13 +1178,14 @@ def test_top_level_local_collides_with_external(self, tmp_path):
             raw = skill_view("shared-name")
 
         result = json.loads(raw)
-        assert result["success"] is False
-        assert "Ambiguous" in result["error"]
-        assert len(result["matches"]) == 2
+        assert result["success"] is True
+        assert "LOCAL VERSION" in result["content"]
 
     def test_collision_resolvable_via_categorized_path(self, tmp_path):
-        """User can recover from a collision by passing the full
-        categorized path — the bare name is ambiguous, the path is not."""
+        """User can still pin a specific skill by passing its full
+        categorized path. The bare name resolves to the local copy by
+        tier; the explicit path bypasses the collision logic and loads
+        exactly the requested file."""
         local_dir = tmp_path / "local"
         external_dir = tmp_path / "external"
         local_dir.mkdir()
@@ -1223,9 +1225,9 @@ def test_external_skill_resolves_when_no_collision(self, tmp_path):
         assert result["success"] is True
         assert "EXTERNAL BODY" in result["content"]
 
-    def test_two_externals_same_name_also_refuse(self, tmp_path):
-        """Collision detection is symmetric — two external dirs with
-        same-name skills also trigger the refusal."""
+    def test_two_externals_same_name_resolve_by_declaration_order(self, tmp_path):
+        """Same-name skills in two external dirs: the first declared
+        external dir wins (mirrors config.external_dirs order)."""
         local_dir = tmp_path / "local"
         ext_a = tmp_path / "ext_a"
         ext_b = tmp_path / "ext_b"
@@ -1241,9 +1243,8 @@ def test_two_externals_same_name_also_refuse(self, tmp_path):
             raw = skill_view("pr")
 
         result = json.loads(raw)
-        assert result["success"] is False
-        assert "Ambiguous" in result["error"]
-        assert len(result["matches"]) == 2
+        assert result["success"] is True
+        assert "EXT_A VERSION" in result["content"]
 
     def test_local_only_skill_loads_normally(self, tmp_path):
         """Sanity: a single local skill (no external collision) loads
@@ -1267,3 +1268,33 @@ def test_local_only_skill_loads_normally(self, tmp_path):
         result = json.loads(raw)
         assert result["success"] is True
         assert "LOCAL BODY" in result["content"]
+
+    def test_same_tier_collision_resolves_by_mtime(self, tmp_path):
+        """Two same-name skills nested differently inside the SAME
+        SKILLS_DIR (same tier): the most recently modified wins.
+        Reproduces the original incident — two ``kanban-worker``
+        SKILL.md files under the same dir at different versions, where
+        the newer (v2.2.0) one should be loaded."""
+        import os
+        import time
+
+        local_dir = tmp_path / "local"
+        local_dir.mkdir()
+
+        _make_skill(local_dir, "kanban-worker", category="devops", body="V2.0.0 OLD")
+        # Bump second skill's mtime explicitly so the test is independent
+        # of filesystem write ordering.
+        _make_skill(local_dir, "kanban-worker", category="legacy", body="V2.2.0 NEW")
+        old_path = local_dir / "devops" / "kanban-worker" / "SKILL.md"
+        new_path = local_dir / "legacy" / "kanban-worker" / "SKILL.md"
+        now = time.time()
+        os.utime(old_path, (now - 3600, now - 3600))
+        os.utime(new_path, (now, now))
+
+        p1, p2 = self._patch_dirs(local_dir, [])
+        with p1, p2:
+            raw = skill_view("kanban-worker")
+
+        result = json.loads(raw)
+        assert result["success"] is True
+        assert "V2.2.0 NEW" in result["content"]

tools/skills_tool.py

@@ -1008,32 +1008,64 @@ def _record(sd: Optional[Path], smd: Path) -> None:
                 if found_md.name != "SKILL.md":
                     _record(None, found_md)
 
-        if len(candidates) > 1:
-            paths = [str(smd) for _, smd in candidates]
-            logging.getLogger(__name__).warning(
-                "Skill name collision for '%s': %d candidates — %s",
-                name, len(candidates), "; ".join(paths),
-            )
-            return json.dumps(
-                {
-                    "success": False,
-                    "error": (
-                        f"Ambiguous skill name '{name}': {len(candidates)} skills "
-                        "match across your local skills dir and external_dirs. "
-                        "Refusing to guess — load one explicitly by its categorized path."
-                    ),
-                    "matches": paths,
-                    "hint": (
-                        "Pass the full relative path instead of the bare name "
-                        "(e.g., 'category/skill-name'), or rename one of the "
-                        "colliding skills so each name is unique."
-                    ),
-                },
-                ensure_ascii=False,
-            )
-
         if candidates:
-            skill_dir, skill_md = candidates[0]
+            # Deterministic resolution: prefer the highest-priority tier
+            # (SKILLS_DIR first, then each external_dir in declaration order).
+            # Within a tier, if multiple candidates collide, pick the
+            # most-recently modified SKILL.md. Surfacing a *crashing* error
+            # on collision proved too brittle for the worker preload path
+            # (see issue: stale profile skill copy crashed every kanban
+            # worker spawn). The CLI has no human to disambiguate, so we
+            # pick — loudly — and surface the collision via a WARN log so
+            # an operator can clean up the stale copy.
+            #
+            # Tier assignment: a candidate's tier is the index of the
+            # first ``all_dirs`` entry it lives under. Candidates outside
+            # any known dir (shouldn't happen with the strategies above)
+            # are sorted to the end.
+            try:
+                resolved_dirs = [d.resolve() for d in all_dirs]
+            except OSError:
+                resolved_dirs = list(all_dirs)
+
+            def _tier_of(smd: Path) -> int:
+                try:
+                    rmd = smd.resolve()
+                except OSError:
+                    rmd = smd
+                for i, d in enumerate(resolved_dirs):
+                    try:
+                        rmd.relative_to(d)
+                        return i
+                    except ValueError:
+                        continue
+                return len(resolved_dirs)
+
+            def _mtime(smd: Path) -> float:
+                try:
+                    return smd.stat().st_mtime
+                except OSError:
+                    return 0.0
+
+            # Sort: by tier ascending, then mtime descending (newest first).
+            ranked = sorted(
+                candidates,
+                key=lambda item: (_tier_of(item[1]), -_mtime(item[1])),
+            )
+            skill_dir, skill_md = ranked[0]
+
+            if len(candidates) > 1:
+                chosen_path = str(skill_md)
+                other_paths = [str(smd) for _, smd in ranked[1:]]
+                logging.getLogger(__name__).warning(
+                    "Skill name collision for '%s' resolved to %s "
+                    "(tier=%d). Shadowed candidates: %s. "
+                    "Remove the stale copies to silence this warning.",
+                    name,
+                    chosen_path,
+                    _tier_of(skill_md),
+                    "; ".join(other_paths),
+                )
 
         if not skill_md or not skill_md.exists():
             available = [s["name"] for s in _sort_skills(_find_all_skills())[:20]]