Replacement: validate forced skills at card creation

hermes_cli/kanban_db.py

@@ -1218,6 +1218,48 @@ def _normalize_task_skills(skills: Optional[Iterable[str]]) -> Optional[list[str
     return cleaned
 
 
+def _validate_task_skills_for_assignee(assignee: Optional[str], skills: Optional[list[str]]) -> None:
+    """Reject forced skills that the assigned profile cannot load at worker startup."""
+    if not skills:
+        return
+
+    from agent.skill_utils import is_excluded_skill_path, iter_skill_index_files, parse_frontmatter
+    from hermes_cli import profiles as profiles_mod
+
+    profile = profiles_mod.normalize_profile_name(assignee or "")
+    profile_home = profiles_mod.get_profile_dir(profile)
+    skills_dir = profile_home / "skills"
+    available: set[str] = set()
+    if skills_dir.is_dir():
+        for skill_md in iter_skill_index_files(skills_dir, "SKILL.md"):
+            if is_excluded_skill_path(skill_md):
+                continue
+            relative = skill_md.relative_to(skills_dir).parts[:-1]
+            if relative:
+                available.add(relative[-1])
+                available.add("/".join(relative))
+            try:
+                frontmatter, _body = parse_frontmatter(
+                    skill_md.read_text(encoding="utf-8", errors="replace")[:4000]
+                )
+                if frontmatter.get("name"):
+                    available.add(str(frontmatter["name"]))
+            except OSError:
+                continue
+        for flat_skill in skills_dir.rglob("*.md"):
+            if flat_skill.name != "SKILL.md" and not is_excluded_skill_path(flat_skill):
+                available.add(flat_skill.stem)
+
+    missing = [skill for skill in skills if skill not in available]
+    if missing:
+        quoted = ", ".join(repr(skill) for skill in missing)
+        raise ValueError(
+            f"forced skill(s) {quoted} are not installed for assignee profile {profile!r}. "
+            f"Install them in {skills_dir} or choose from `hermes -p {profile} skills list` "
+            "before creating the task."
+        )
+
+
 def create_task(
     conn: sqlite3.Connection, *, title: str, body: Optional[str] = None,
     assignee: Optional[str] = None, created_by: Optional[str] = None,
@@ -1280,6 +1322,7 @@ def create_task(
     )
     parents = tuple(p for p in parents if p)
     skills_list = _normalize_task_skills(skills)
+    _validate_task_skills_for_assignee(assignee, skills_list)
 
     # Idempotency check BEFORE the write txn (no lock held); a concurrent-create
     # race may insert twice, the next lookup stabilises on the newest.

tests/hermes_cli/test_kanban_review_lifecycle.py

@@ -481,6 +481,12 @@ def test_active_pr_guard_skipped_for_review_lane_but_defers_ready_lane(
 def test_review_dispatch_preserves_task_skills_and_adds_reviewer_skill(
     kanban_home: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    skill_dir = kanban_home / "profiles" / "reviewer" / "skills" / "domain-specific-review"
+    skill_dir.mkdir(parents=True)
+    (skill_dir / "SKILL.md").write_text(
+        "---\nname: domain-specific-review\ndescription: Review domain changes.\n---\n",
+        encoding="utf-8",
+    )
     import hermes_cli.config as cfgmod
     import hermes_cli.profiles as profmod
 

tests/hermes_cli/test_kanban_swarm.py

@@ -2,6 +2,7 @@
 
 from hermes_cli import kanban_db as kb
 from hermes_cli import kanban_db_connect as kbc
+from hermes_cli.profiles import get_profile_dir
 from hermes_cli.kanban_swarm import (
     SwarmWorkerSpec,
     create_swarm,
@@ -10,6 +11,24 @@
 )
 
 
+@pytest.fixture(autouse=True)
+def forced_skills(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None:
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
+    for profile, skill_name in (
+        ("reviewer", "requesting-code-review"),
+        ("writer", "humanizer"),
+    ):
+        skill_dir = get_profile_dir(profile) / "skills" / skill_name
+        skill_dir.mkdir(parents=True, exist_ok=True)
+        (skill_dir / "SKILL.md").write_text(
+            f"---\nname: {skill_name}\ndescription: Swarm test skill.\n---\n",
+            encoding="utf-8",
+        )
+
+
 def test_create_swarm_builds_parallel_workers_verifier_and_synthesizer(tmp_path):
     conn = kbc.connect(tmp_path / "kanban.db")
     try:

tests/tools/test_kanban_tools.py

@@ -429,6 +429,69 @@ def test_create_happy_path(worker_env):
         conn.close()
 
 
+def test_create_rejects_skill_missing_from_assignee_profile_without_persisting(
+    worker_env, tmp_path,
+):
+    """A default-profile skill must not validate for a named assignee."""
+    default_skill = tmp_path / ".hermes" / "skills" / "monitoring-incident-response"
+    default_skill.mkdir(parents=True)
+    (default_skill / "SKILL.md").write_text(
+        "---\nname: monitoring-incident-response\ndescription: Monitor incidents.\n---\n"
+    )
+    (tmp_path / ".hermes" / "profiles" / "foreman" / "skills").mkdir(parents=True)
+
+    from hermes_cli import kanban_db as kb
+    from hermes_cli import kanban_db_connect as kbc
+    from tools import kanban_tools as kt
+
+    conn = kbc.connect()
+    try:
+        before = len(kb.list_tasks(conn))
+    finally:
+        conn.close()
+
+    result = json.loads(kt._handle_create({
+        "title": "must not persist",
+        "assignee": "foreman",
+        "skills": ["monitoring-incident-response"],
+    }))
+
+    assert "error" in result
+    assert "monitoring-incident-response" in result["error"]
+    assert "foreman" in result["error"]
+    assert "hermes -p foreman skills list" in result["error"]
+    conn = kbc.connect()
+    try:
+        assert len(kb.list_tasks(conn)) == before
+    finally:
+        conn.close()
+
+
+def test_create_preserves_skill_resolvable_in_assignee_profile(worker_env, tmp_path):
+    skill = tmp_path / ".hermes" / "profiles" / "foreman" / "skills" / "review"
+    skill.mkdir(parents=True)
+    (skill / "SKILL.md").write_text(
+        "---\nname: review\ndescription: Review changes.\n---\n"
+    )
+
+    from tools import kanban_tools as kt
+
+    result = json.loads(kt._handle_create({
+        "title": "valid forced skill",
+        "assignee": "foreman",
+        "skills": ["review"],
+    }))
+
+    assert result["ok"] is True
+    from hermes_cli import kanban_db as kb
+    from hermes_cli import kanban_db_connect as kbc
+    conn = kbc.connect()
+    try:
+        assert kb.get_task(conn, result["task_id"]).skills == ["review"]
+    finally:
+        conn.close()
+
+
 def test_link_happy_path(worker_env):
     from hermes_cli import kanban_db as kb
     from hermes_cli import kanban_db_connect as kbc