curator: enforce skill lock/pin, drop the review fork's shell, stop self-inflating review telemetry

agent/curator.py

@@ -1923,7 +1923,15 @@ def _run_llm_review(prompt: str) -> Dict[str, Any]:
             credential_pool=_credential_pool,
             request_overrides=_request_overrides,
             **_agent_kwargs,
-            enabled_toolsets=["skills", "terminal"],
+            # CARRY: upstream ships ["skills", "terminal"] so the fork can
+            # `mv` support files into an umbrella's references/ while
+            # consolidating. We drop `terminal`: the prompt's "DO NOT call
+            # terminal to mv skill directories into .archive/" is advisory
+            # only, and this fork is autonomous with no user in the loop — a
+            # shell it can reach archives an operator-authored skill past
+            # every guard the skill_manage path enforces. Consolidation moves
+            # are the cost; we don't use them.
+            enabled_toolsets=["skills"],
             # Umbrella-building over a large skill collection is worth a
             # high iteration ceiling — the pass typically takes 50-100
             # API calls against hundreds of candidate skills. The

tests/agent/test_curator.py

@@ -701,17 +701,26 @@ def close(self):
 
 
 
-def test_review_fork_restricts_toolsets_to_skills_and_terminal(curator_env, monkeypatch):
-    """The curator LLM fork must advertise only the skills + terminal toolsets.
-
-    Without ``enabled_toolsets=["skills", "terminal"]`` on the AIAgent(...) call
-    in ``_run_llm_review``, ``enabled_toolsets`` defaults to None and init_agent
-    grants the fork the full default catalog (~30 tools) plus the context_engine
-    (lcm_*) tools, billing ~7K wasted schema tokens on every one of the fork's
-    50-100 API calls per consolidation pass. The prompt (curator.py:509-523)
-    confines the model to four tools in natural language, but only this kwarg
-    filters the advertised request schema. Capturing the constructor kwarg is
-    the sole assertion that distinguishes fixed from unfixed code.
+def test_review_fork_restricts_toolsets_to_skills_only(curator_env, monkeypatch):
+    """The curator LLM fork must advertise only the skills toolset.
+
+    Without ``enabled_toolsets`` on the AIAgent(...) call in
+    ``_run_llm_review`` it defaults to None and init_agent grants the fork the
+    full default catalog (~30 tools) plus the context_engine (lcm_*) tools,
+    billing ~7K wasted schema tokens on every one of the fork's 50-100 API
+    calls per consolidation pass. The prompt (curator.py:509-523) confines the
+    model to four tools in natural language, but only this kwarg filters the
+    advertised request schema. Capturing the constructor kwarg is the sole
+    assertion that distinguishes fixed from unfixed code.
+
+    CARRY DIVERGENCE from upstream: upstream pins the fork to
+    ``["skills", "terminal"]``, keeping ``terminal`` so the fork can `mv`
+    support files into an umbrella's references/ during consolidation. We drop
+    ``terminal`` entirely: the prompt's "DO NOT call terminal to mv skill
+    directories into .archive/" is advisory, and an autonomous fork that can
+    shell out can archive an operator-authored skill past every guard the
+    ``skill_manage`` path enforces. We do not use umbrella consolidation;
+    losing it is the cheaper side of that trade.
     """
     curator = curator_env["curator"]
 
@@ -742,35 +751,40 @@ def close(self):
 
     # error is None proves the fork was actually constructed (capture ran).
     assert meta.get("error") is None, meta.get("error")
-    assert captured.get("enabled_toolsets") == ["skills", "terminal"], (
-        "curator review fork did not pass enabled_toolsets=['skills', "
-        "'terminal'] to AIAgent; the full default tool catalog (plus lcm_* "
-        "context_engine tools) would be advertised; got "
-        f"{captured.get('enabled_toolsets')!r}"
+    assert captured.get("enabled_toolsets") == ["skills"], (
+        "curator review fork did not pass enabled_toolsets=['skills'] to "
+        "AIAgent; with 'terminal' back in the list the fork can shell out and "
+        "mv a skill directory into .archive/ past every skill_manage guard; "
+        f"got {captured.get('enabled_toolsets')!r}"
     )
 
 
-def test_review_fork_toolset_surface_is_skills_plus_terminal():
+def test_review_fork_toolset_surface_is_skills_without_terminal():
     """Documentary check on the static surface the fork's kwarg resolves to.
 
     Registry-independent (include_registry=False) so a plugin-registered tool
     tagged into these toolsets cannot flake the membership checks. This
-    documents the intended surface (the four prompt-named tools present, dead
+    documents the intended surface (the prompt-named skill tools present, dead
     default and lcm_* schema absent) but does not itself guard the call-site
-    kwarg. No exact-set pin: intentional additions to either toolset must not
+    kwarg. No exact-set pin: intentional additions to the toolset must not
     fail this test.
+
+    ``terminal`` absent is the carry divergence (see
+    test_review_fork_restricts_toolsets_to_skills_only): resolving the surface
+    is what proves the fork cannot shell out, since a toolset the kwarg omits
+    contributes no tool schema at all.
     """
     from toolsets import resolve_toolset
 
-    surface = set(resolve_toolset("skills", include_registry=False)) | set(
-        resolve_toolset("terminal", include_registry=False)
-    )
+    surface = set(resolve_toolset("skills", include_registry=False))
 
-    # The four prompt-named tools are all present.
+    # The prompt-named skill tools are all present.
     assert "skills_list" in surface
     assert "skill_view" in surface
     assert "skill_manage" in surface
-    assert "terminal" in surface
+
+    # The shell-out escape hatch is gone.
+    assert "terminal" not in surface
 
     # Representative dropped default + context_engine tools are absent.
     assert "read_file" not in surface

tests/tools/test_skill_usage.py

@@ -202,6 +202,48 @@ def test_is_agent_created(skills_home):
 # Archive / restore
 # ---------------------------------------------------------------------------
 
+def test_curation_eligible_honors_frontmatter_pinned(skills_home):
+    from tools.skill_usage import is_curation_eligible
+    skills_dir = skills_home / "skills"
+    d = skills_dir / "pinned-skill"
+    d.mkdir()
+    (d / "SKILL.md").write_text(
+        "---\n"
+        "name: pinned-skill\n"
+        "description: test skill\n"
+        "metadata:\n"
+        "  hermes:\n"
+        "    pinned: true\n"
+        "---\n\n# body\n",
+        encoding="utf-8",
+    )
+    assert is_curation_eligible("pinned-skill", d / "SKILL.md") is False
+
+
+def test_curation_eligible_honors_frontmatter_locked(skills_home):
+    from tools.skill_usage import is_curation_eligible
+    skills_dir = skills_home / "skills"
+    d = skills_dir / "locked-skill"
+    d.mkdir()
+    (d / "SKILL.md").write_text(
+        "---\n"
+        "name: locked-skill\n"
+        "description: test skill\n"
+        "metadata:\n"
+        "  hermes:\n"
+        "    locked: true\n"
+        "---\n\n# body\n",
+        encoding="utf-8",
+    )
+    assert is_curation_eligible("locked-skill", d / "SKILL.md") is False
+
+
+def test_curation_eligible_ignores_frontmatter_without_flags(skills_home):
+    from tools.skill_usage import is_curation_eligible
+    skills_dir = skills_home / "skills"
+    d = _write_skill(skills_dir, "plain-skill")
+    assert is_curation_eligible("plain-skill", d / "SKILL.md") is True
+
 
 # ---------------------------------------------------------------------------
 # Reporting

tests/tools/test_skills_tool.py

@@ -387,6 +387,44 @@ def test_view_finds_skill_in_symlinked_category_dir(self, tmp_path):
         assert result["name"] == "knowledge-brain"
 
 
+class TestSkillViewBump:
+    """`skill_view`'s telemetry bump must not fire from the curator's
+    background-review fork — else browsing candidates while judging them
+    inflates the very use_count/last_used_at signal the review reads (#77)."""
+
+    def test_bump_skipped_during_background_review(self, tmp_path):
+        with (
+            patch("tools.skills_tool.SKILLS_DIR", tmp_path),
+            patch("tools.skill_provenance.is_background_review", return_value=True),
+            patch("tools.skill_usage.bump_view") as mock_bump_view,
+            patch("tools.skill_usage.bump_use") as mock_bump_use,
+        ):
+            _make_skill(tmp_path, "my-skill")
+            result = json.loads(
+                skills_tool_module._skill_view_with_bump({"name": "my-skill"})
+            )
+
+        assert result["success"] is True
+        mock_bump_view.assert_not_called()
+        mock_bump_use.assert_not_called()
+
+    def test_bump_still_fires_for_foreground_calls(self, tmp_path):
+        with (
+            patch("tools.skills_tool.SKILLS_DIR", tmp_path),
+            patch("tools.skill_provenance.is_background_review", return_value=False),
+            patch("tools.skill_usage.bump_view") as mock_bump_view,
+            patch("tools.skill_usage.bump_use") as mock_bump_use,
+        ):
+            _make_skill(tmp_path, "my-skill")
+            result = json.loads(
+                skills_tool_module._skill_view_with_bump({"name": "my-skill"})
+            )
+
+        assert result["success"] is True
+        mock_bump_view.assert_called_once_with("my-skill")
+        mock_bump_use.assert_called_once_with("my-skill")
+
+
 class TestSkillViewSecureSetupOnLoad:
     def test_requests_missing_required_env_and_continues(self, tmp_path, monkeypatch):
         monkeypatch.delenv("TENOR_API_KEY", raising=False)

tools/skill_usage.py

@@ -474,14 +474,38 @@ def is_curation_eligible(skill_name: str, skill_path: Optional[Path] = None) ->
         return False
     if is_bundled(skill_name):
         return _prune_builtins_enabled()
+    if skill_path is not None and _frontmatter_locks_curation(skill_path):
+        return False
     local_dir = _find_skill_dir(skill_name)
     if local_dir is not None:
+        if _frontmatter_locks_curation(local_dir):
+            return False
         return not is_external_skill_path(local_dir)
     if _find_external_skill_dir(skill_name) is not None:
         return False
     return True
 
 
+def _frontmatter_locks_curation(skill_path: Path) -> bool:
+    """True when the skill's own SKILL.md opts out of curation via
+    ``metadata.hermes.pinned`` or ``metadata.hermes.locked``.
+
+    Fail-open like ``_locked_guard`` in skill_manager_tool.py: a missing or
+    unparseable SKILL.md never blocks curation on its own.
+    """
+    try:
+        skill_md = skill_path / "SKILL.md" if skill_path.is_dir() else skill_path
+        if not skill_md.exists():
+            return False
+        from agent.skill_utils import parse_frontmatter
+        fm, _body = parse_frontmatter(skill_md.read_text(encoding="utf-8"))
+        hermes_meta = (fm.get("metadata") or {}).get("hermes") or {}
+        return hermes_meta.get("pinned") is True or hermes_meta.get("locked") is True
+    except Exception:
+        logger.debug("frontmatter lock lookup failed for %s", skill_path, exc_info=True)
+        return False
+
+
 def _is_curator_managed_record(record: Any) -> bool:
     """Return True when a usage record opts a skill into curator management.
 

tools/skills_tool.py

@@ -1938,12 +1938,18 @@ def _skill_view_with_bump(args, **kw):
             # qualified forms ("plugin:skill") return with the canonical name.
             resolved = parsed.get("name") or name
             if resolved:
-                from tools.skill_usage import bump_use, bump_view
-                bump_view(str(resolved))
-                # A skill_view tool call is the agent actively loading the skill
-                # to act on it — that counts as use, not just a browse/view.
-                # Curator's stale timer keys off last_used_at (see agent/curator.py).
-                bump_use(str(resolved))
+                from tools.skill_provenance import is_background_review
+                # The curator's review fork calls skill_view to inspect
+                # candidates it's judging — that must not itself count as
+                # use, or the inspection inflates the very signal the
+                # review reads (#77).
+                if not is_background_review():
+                    from tools.skill_usage import bump_use, bump_view
+                    bump_view(str(resolved))
+                    # A skill_view tool call is the agent actively loading the skill
+                    # to act on it — that counts as use, not just a browse/view.
+                    # Curator's stale timer keys off last_used_at (see agent/curator.py).
+                    bump_use(str(resolved))
     except Exception:
         pass
     return result