fix(skills): create reusing a dead skill's name gets archived by the curator (#65992)

agent/curator.py

@@ -369,7 +369,16 @@ def apply_automatic_transitions(now: Optional[datetime] = None) -> Dict[str, int
             continue
 
         if anchor <= archive_cutoff and current != _u.STATE_ARCHIVED:
-            ok, _msg = _u.archive_skill(name)
+            # Revalidate the live usage record under the usage lock inside
+            # archive_skill — a concurrent skill_manage(create) may have
+            # discarded/replaced the stale clock this snapshot still holds
+            # (#65992).
+            ok, _msg = _u.archive_skill(
+                name,
+                require_inactive_before=archive_cutoff,
+                stale_before=stale_cutoff,
+                now=now,
+            )
             if ok:
                 counts["archived"] += 1
         elif anchor <= stale_cutoff and current == _u.STATE_ACTIVE:

tests/agent/test_curator_activity.py

@@ -1,6 +1,8 @@
 """Regression tests for curator skill activity timestamps."""
 
 import importlib
+import json
+import threading
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
@@ -54,3 +56,166 @@ def test_recent_view_activity_prevents_false_stale_transition(curator_modules, m
 
     assert counts["marked_stale"] == 0
     assert skill_usage.get_record("recently-viewed")["state"] == "active"
+
+
+def test_fresh_create_reusing_dead_skill_name_is_not_archived(curator_modules, monkeypatch):
+    """#65992: skill_manage(create) reports skills/<category>/<name>/SKILL.md
+    but the file ends up in skills/.archive/<name>/.
+
+    A usage record left behind by a previous skill with the same name (removed
+    without skill_manage(delete), so ``forget()`` never ran) carries an
+    expired inactivity clock. The next automatic-transition pass reads that
+    clock, decides the seconds-old skill is long-inactive, and
+    ``archive_skill()`` relocates it — flattening the category, which is
+    exactly the reported filesystem shape. A successful create must start a
+    new life: the stale record is discarded, so the pass leaves the new
+    skill alone.
+    """
+    home, skill_usage, curator = curator_modules
+    skills_dir = home / "skills"
+
+    now = datetime.now(timezone.utc)
+    stale = (now - timedelta(days=200)).isoformat()
+    skill_usage.save_usage({
+        "fresh-skill": {
+            "created_by": "agent",
+            "use_count": 3,
+            "last_used_at": stale,
+            "created_at": stale,
+            "state": "active",
+        }
+    })
+
+    from tools.skill_manager_tool import skill_manage
+
+    raw = skill_manage(
+        action="create",
+        name="fresh-skill",
+        category="devops",
+        content="---\nname: fresh-skill\ndescription: test skill\n---\n\n# fresh-skill\n",
+    )
+    result = json.loads(raw)
+    assert result["success"] is True
+    assert (skills_dir / "devops" / "fresh-skill" / "SKILL.md").exists()
+
+    monkeypatch.setattr(curator, "get_stale_after_days", lambda: 30)
+    monkeypatch.setattr(curator, "get_archive_after_days", lambda: 90)
+
+    counts = curator.apply_automatic_transitions(now=now)
+
+    assert counts["archived"] == 0
+    assert (skills_dir / "devops" / "fresh-skill" / "SKILL.md").exists(), (
+        "freshly created skill was relocated away from its reported path"
+    )
+    assert not (skills_dir / ".archive" / "fresh-skill").exists(), (
+        "freshly created skill was archived by the automatic-transition pass"
+    )
+
+
+def test_in_flight_curator_snapshot_create_does_not_archive(curator_modules, monkeypatch):
+    """#65992 race: curator already holds a stale snapshot row, then create
+    resets the usage record, then archive_skill runs.
+
+    Without archive-time revalidation under the usage lock, archive_skill
+    resolves the newly written directory and moves it to .archive/. Barriers
+    force snapshot → create/forget → archive ordering.
+    """
+    home, skill_usage, curator = curator_modules
+    skills_dir = home / "skills"
+
+    now = datetime.now(timezone.utc)
+    stale = (now - timedelta(days=200)).isoformat()
+    skill_name = "race-skill"
+    skill_usage.save_usage({
+        skill_name: {
+            "created_by": "agent",
+            "use_count": 3,
+            "last_used_at": stale,
+            "created_at": stale,
+            "state": "active",
+        }
+    })
+
+    # Pre-create the on-disk skill so agent_created_report can snapshot it,
+    # then remove it before the create thread reuses the name (mirrors a
+    # dead skill whose .usage.json row survived).
+    skill_dir = skills_dir / "devops" / skill_name
+    skill_dir.mkdir(parents=True, exist_ok=True)
+    (skill_dir / "SKILL.md").write_text(
+        f"---\nname: {skill_name}\ndescription: dead\n---\n\n# {skill_name}\n",
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(curator, "get_stale_after_days", lambda: 30)
+    monkeypatch.setattr(curator, "get_archive_after_days", lambda: 90)
+
+    snapshot_ready = threading.Barrier(2, timeout=15)
+    create_done = threading.Barrier(2, timeout=15)
+    real_report = skill_usage.agent_created_report
+
+    def stalled_report():
+        rows = real_report()
+        assert any(r["name"] == skill_name for r in rows), rows
+        # Snapshot captured with the stale clock. Let create discard it and
+        # land a fresh skill before the pass proceeds to archive_skill.
+        snapshot_ready.wait()
+        create_done.wait()
+        return rows
+
+    monkeypatch.setattr(skill_usage, "agent_created_report", stalled_report)
+
+    from tools.skill_manager_tool import skill_manage
+
+    counts_holder: list = []
+    errors: list = []
+
+    def run_curator():
+        try:
+            counts_holder.append(curator.apply_automatic_transitions(now=now))
+        except Exception as exc:  # pragma: no cover - surfaced via errors
+            errors.append(exc)
+
+    def run_create():
+        try:
+            snapshot_ready.wait()
+            # Dead skill gone; create reuses the name (forget runs before write).
+            import shutil
+            shutil.rmtree(skill_dir, ignore_errors=True)
+            raw = skill_manage(
+                action="create",
+                name=skill_name,
+                category="devops",
+                content=(
+                    f"---\nname: {skill_name}\ndescription: fresh\n---\n\n"
+                    f"# {skill_name}\n"
+                ),
+            )
+            result = json.loads(raw)
+            assert result["success"] is True, result
+            assert (skills_dir / "devops" / skill_name / "SKILL.md").exists()
+            create_done.wait()
+        except Exception as exc:  # pragma: no cover - surfaced via errors
+            errors.append(exc)
+            # Unstick the curator thread if create failed mid-barrier.
+            for barrier in (snapshot_ready, create_done):
+                try:
+                    barrier.abort()
+                except Exception:
+                    pass
+
+    t_curator = threading.Thread(target=run_curator, name="curator-pass")
+    t_create = threading.Thread(target=run_create, name="skill-create")
+    t_curator.start()
+    t_create.start()
+    t_curator.join(timeout=30)
+    t_create.join(timeout=30)
+
+    assert not errors, errors
+    assert counts_holder, "curator thread did not finish"
+    assert counts_holder[0]["archived"] == 0, counts_holder[0]
+    assert (skills_dir / "devops" / skill_name / "SKILL.md").exists(), (
+        "in-flight curator pass archived a skill created after its snapshot"
+    )
+    assert not (skills_dir / ".archive" / skill_name).exists(), (
+        "fresh skill landed in .archive under snapshot/create/archive interleaving"
+    )

tests/tools/test_skill_manager_tool.py

@@ -563,6 +563,52 @@ def test_full_create_via_dispatcher(self, tmp_path):
         rec = usage.get("test-skill") or {}
         assert rec.get("created_by") in {None, "", False}
 
+    def test_create_discards_usage_record_left_by_previous_life(self, tmp_path):
+        """#65992: reusing the name of a dead skill must reset its usage record.
+
+        Usage records are keyed by skill name and survive any deletion that
+        did not go through skill_manage(delete) (manual rm, crashed delete).
+        If a create leaves the stale record in place, the curator's next
+        automatic-transition pass reads the expired inactivity clock and
+        relocates the brand-new skill to skills/.archive/<name>/ while the
+        create response still points at the requested category path.
+        """
+        from datetime import datetime, timedelta, timezone
+        from tools.skill_usage import load_usage, save_usage
+
+        stale = (datetime.now(timezone.utc) - timedelta(days=200)).isoformat()
+        save_usage({
+            "test-skill": {
+                "created_by": "agent",
+                "use_count": 3,
+                "last_used_at": stale,
+                "created_at": stale,
+                "state": "active",
+            }
+        })
+        with _skill_dir(tmp_path):
+            raw = skill_manage(action="create", name="test-skill", content=VALID_SKILL_CONTENT)
+            usage = load_usage()
+        result = json.loads(raw)
+        assert result["success"] is True
+        assert "test-skill" not in usage
+
+    def test_failed_create_keeps_existing_usage_record(self, tmp_path):
+        """A duplicate-name create fails and must NOT touch the live skill's
+        usage record — only a successful create starts a new life (#65992)."""
+        from tools.skill_usage import load_usage, save_usage
+
+        with _skill_dir(tmp_path):
+            raw = skill_manage(action="create", name="test-skill", content=VALID_SKILL_CONTENT)
+            assert json.loads(raw)["success"] is True
+            save_usage({"test-skill": {"created_by": "agent", "use_count": 7, "state": "active"}})
+            raw = skill_manage(action="create", name="test-skill", content=VALID_SKILL_CONTENT)
+            usage = load_usage()
+        result = json.loads(raw)
+        assert result["success"] is False
+        assert "already exists" in result["error"]
+        assert usage["test-skill"]["use_count"] == 7
+
     def test_create_from_background_review_marks_agent_created(self, tmp_path):
         """Background-review fork creates ARE marked as agent-created."""
         from tools.skill_provenance import set_current_write_origin, BACKGROUND_REVIEW

tools/skill_manager_tool.py

@@ -838,6 +838,20 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
             "error": f"A skill named '{name}' already exists at {existing['path']}."
         }
 
+    # Discard any leftover usage record BEFORE the new directory is visible.
+    # Records are keyed by name and carry the curator's inactivity clock; a
+    # stale leftover would make an automatic-transition pass (including one
+    # already mid-walk with a snapshot row) archive the seconds-old skill
+    # into skills/.archive/ while create still reports the live path (#65992).
+    # Clearing the clock before mkdir closes the write-then-forget race with
+    # archive_skill's locked revalidation. Collision check above guarantees
+    # no live skill owns this record. Best-effort: telemetry never breaks create.
+    try:
+        from tools.skill_usage import forget
+        forget(name)
+    except Exception:
+        logger.debug("usage-record reset failed for %s", name, exc_info=True)
+
     # Create the skill directory
     skill_dir = _resolve_skill_dir(name, category)
     skill_dir.mkdir(parents=True, exist_ok=True)

tools/skill_usage.py

@@ -676,7 +676,13 @@ def _apply(rec: Dict[str, Any]) -> None:
 
 
 def forget(skill_name: str) -> None:
-    """Drop a skill's usage entry entirely. Called when the skill is deleted."""
+    """Drop a skill's usage entry entirely.
+
+    Called when the skill is hard-deleted, and by ``skill_manage(create)``
+    when a new skill reuses the name of a dead one — the record's inactivity
+    clock belongs to the previous life and must not carry over, or the
+    curator archives the brand-new skill on its next pass (#65992).
+    """
     if not skill_name:
         return
     try:
@@ -693,13 +699,70 @@ def forget(skill_name: str) -> None:
 # Archive / restore
 # ---------------------------------------------------------------------------
 
-def archive_skill(skill_name: str) -> Tuple[bool, str]:
+def _record_still_warrants_archive(
+    skill_name: str,
+    record: Any,
+    *,
+    archive_cutoff: datetime,
+    stale_cutoff: Optional[datetime] = None,
+    now: Optional[datetime] = None,
+) -> bool:
+    """Whether a *live* usage record still justifies automatic archival.
+
+    Used to revalidate a curator snapshot decision under ``_usage_file_lock``
+    immediately before the destructive rename. A ``skill_manage(create)`` that
+    reused this name may have ``forget()``-cleared the stale clock (or replaced
+    it with a fresh one) while the pass was mid-walk (#65992).
+
+    Agent-authored skills must still carry a curator-managed marker; bundled
+    built-ins (when ``curator.prune_builtins`` is on) only need a persisted
+    inactivity record — they never set ``created_by=agent``.
+    """
+    if not isinstance(record, dict):
+        return False
+    if is_bundled(skill_name):
+        if not _prune_builtins_enabled():
+            return False
+    elif not _is_curator_managed_record(record):
+        return False
+    if record.get("pinned"):
+        return False
+    if record.get("state") == STATE_ARCHIVED:
+        return False
+
+    if now is None:
+        now = datetime.now(timezone.utc)
+    last_activity = _parse_iso_timestamp(latest_activity_at(record))
+    anchor = last_activity or _parse_iso_timestamp(record.get("created_at")) or now
+    if anchor.tzinfo is None:
+        anchor = anchor.replace(tzinfo=timezone.utc)
+
+    # Mirror apply_automatic_transitions: never-used skills get a grace floor
+    # at stale_after_days before archival is allowed.
+    never_used = int(record.get("use_count", 0) or 0) == 0
+    if stale_cutoff is not None and never_used and anchor > stale_cutoff:
+        return False
+    return anchor <= archive_cutoff
+
+
+def archive_skill(
+    skill_name: str,
+    *,
+    require_inactive_before: Optional[datetime] = None,
+    stale_before: Optional[datetime] = None,
+    now: Optional[datetime] = None,
+) -> Tuple[bool, str]:
     """Move a curator-eligible skill directory to ~/.hermes/skills/.archive/.
 
     Returns (ok, message). Never archives hub-installed skills. Bundled
     built-ins are only archivable when ``curator.prune_builtins`` is enabled;
     when one is archived, its name is added to the suppression list so the
     update-time re-seeder leaves it archived instead of restoring it.
+
+    When *require_inactive_before* is set (automatic curator transitions), the
+    live usage record is revalidated under ``_usage_file_lock`` immediately
+    before the rename. Manual / consolidation callers omit it so an explicit
+    archive still proceeds even if the inactivity clock was reset.
     """
     local_skill_dir = _find_skill_dir(skill_name)
     if local_skill_dir is None and _find_external_skill_dir(skill_name) is not None:
@@ -736,15 +799,34 @@ def archive_skill(skill_name: str) -> Tuple[bool, str]:
     if dest.exists():
         dest = archive_root / f"{skill_dir.name}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
 
-    try:
-        skill_dir.rename(dest)
-    except OSError as e:
-        # Cross-device — fall back to shutil.move
-        import shutil
+    # Hold the usage lock across revalidation + rename so a concurrent
+    # skill_manage(create) forget()/mark cannot slip a fresh life in between
+    # the check and the destructive move (#65992).
+    with _usage_file_lock():
+        if require_inactive_before is not None:
+            data = load_usage()
+            rec = data.get(skill_name)
+            if not _record_still_warrants_archive(
+                skill_name,
+                rec,
+                archive_cutoff=require_inactive_before,
+                stale_cutoff=stale_before,
+                now=now,
+            ):
+                return False, (
+                    f"skill '{skill_name}' usage record no longer warrants "
+                    "automatic archival"
+                )
+
         try:
-            shutil.move(str(skill_dir), str(dest))
-        except Exception as e2:
-            return False, f"failed to archive: {e2}"
+            skill_dir.rename(dest)
+        except OSError as e:
+            # Cross-device — fall back to shutil.move
+            import shutil
+            try:
+                shutil.move(str(skill_dir), str(dest))
+            except Exception as e2:
+                return False, f"failed to archive: {e2}"
 
     # Pruning a built-in only sticks if the re-seeder is told to leave it alone.
     if is_bundled(skill_name):