Fix Kanban worker archived-skill detection and diagnostics

hermes_cli/kanban_db.py

@@ -6636,9 +6636,16 @@ def _kanban_worker_skill_available(hermes_home: Optional[str]) -> bool:
     # profiles that have it nested elsewhere.
     if (skills_root / "devops" / "kanban-worker" / "SKILL.md").is_file():
         return True
+    def _is_archived_or_hidden(path: _Path) -> bool:
+        try:
+            rel_parts = path.relative_to(skills_root).parts
+        except ValueError:
+            rel_parts = path.parts
+        return any(part.startswith(".") or part == "archive" for part in rel_parts)
+
     try:
         for skill_md in skills_root.rglob("kanban-worker/SKILL.md"):
-            if skill_md.is_file():
+            if skill_md.is_file() and not _is_archived_or_hidden(skill_md):
                 return True
     except OSError:
         pass

hermes_cli/kanban_diagnostics.py

@@ -536,6 +536,14 @@ def _rule_repeated_failures(task, events, runs, now, cfg) -> list[Diagnostic]:
         cfg.get("spawn_failure_threshold", 3),
     ), 3)
     failure_limit = _positive_int(cfg.get("failure_limit"), threshold)
+    limit_source = "dispatcher"
+    task_max_retries = _task_field(task, "max_retries", None)
+    if task_max_retries is not None:
+        parsed_task_limit = _positive_int(task_max_retries, 0)
+        if parsed_task_limit >= 1:
+            failure_limit = parsed_task_limit
+            limit_source = "task"
+    threshold = min(threshold, failure_limit)
     # Read the new unified counter name, with a fallback to the legacy
     # column name so this rule keeps working against old DB rows the
     # caller somehow materialised without running the migration.
@@ -633,6 +641,7 @@ def _rule_repeated_failures(task, events, runs, now, cfg) -> list[Diagnostic]:
             "last_error": last_err,
             "failure_threshold": threshold,
             "failure_limit": failure_limit,
+            "failure_limit_source": limit_source,
         },
     )]
 

tests/hermes_cli/test_kanban_cli_dispatch_passthrough.py

@@ -148,3 +148,25 @@ def test_kanban_swarm_uses_existing_humanizer_skill():
         f"humanizer skill missing at {humanizer_path}; the kanban_swarm fix "
         "for #29415 requires this bundled skill to exist"
     )
+
+
+def test_kanban_worker_skill_ignores_archived_copy(tmp_path):
+    """Archived profile skills are not loadable by name.
+
+    If the dispatcher treats ``skills/.archive/kanban-worker/SKILL.md`` as
+    available, Foreman workers launch with ``--skills kanban-worker`` and die
+    before the task starts with "Unknown skill(s): kanban-worker".
+    """
+    from hermes_cli import kanban_db
+
+    archived = tmp_path / "skills" / ".archive" / "kanban-worker"
+    archived.mkdir(parents=True)
+    (archived / "SKILL.md").write_text("# archived copy\n", encoding="utf-8")
+
+    assert kanban_db._kanban_worker_skill_available(str(tmp_path)) is False
+
+    live = tmp_path / "skills" / "devops" / "kanban-worker"
+    live.mkdir(parents=True)
+    (live / "SKILL.md").write_text("# live copy\n", encoding="utf-8")
+
+    assert kanban_db._kanban_worker_skill_available(str(tmp_path)) is True

tests/hermes_cli/test_kanban_diagnostics.py

@@ -198,6 +198,26 @@ def test_repeated_failures_default_matches_dispatcher_failure_limit():
     assert "configured for 2" in d.detail
 
 
+def test_repeated_failures_honours_task_max_retries_below_dispatcher_limit():
+    """A per-task retry cap can trip before the dispatcher default."""
+    task = _task(
+        status="blocked",
+        consecutive_failures=1,
+        max_retries=1,
+        last_failure_error="pid 10400 not alive",
+    )
+    runs = [_run(outcome="crashed", run_id=54)]
+    diags = kd.compute_task_diagnostics(task, [], runs, config={"failure_limit": 2})
+    repeated = [d for d in diags if d.kind == "repeated_failures"]
+    assert len(repeated) == 1
+    d = repeated[0]
+    assert d.data["failure_threshold"] == 1
+    assert d.data["failure_limit"] == 1
+    assert d.data["failure_limit_source"] == "task"
+    assert d.data["most_recent_outcome"] == "crashed"
+    assert "configured for 1" in d.detail
+
+
 def test_repeated_failures_derives_threshold_from_kanban_failure_limit():
     task = _task(status="ready", consecutive_failures=2,
                  last_failure_error="Profile 'debugger' does not exist")