ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

Merge pull request #68883 from afourniernv/feat/hermes-relay-skill-metrics

Browse Source 
feat(observability): aggregate bounded skill metrics
This commit is contained in:
Jeffrey Quesnelle
2026-08-05 13:20:57 -04:00
committed by GitHub
parent 25c7827ec9 451a078a50
commit 0531aad55d
24 changed files with 1500 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/skill_bundles.py
+1 -1
View File
@@ -321,7 +321,7 @@ def build_bundle_invocation_message(
try:
from tools.skill_usage import bump_use
bump_use(skill_name)
bump_use(skill_name, task_id=task_id)
except Exception:
pass
agent/skill_commands.py
+3 -3
View File
@@ -595,7 +595,7 @@ def build_skill_invocation_message(
# Track active usage for Curator lifecycle management (#17782)
try:
from tools.skill_usage import bump_use
bump_use(skill_name)
bump_use(skill_name, task_id=task_id)
except Exception:
pass # Non-critical — skill invocation proceeds regardless
@@ -703,7 +703,7 @@ def build_stacked_skill_invocation_message(
# Track active usage for Curator lifecycle management (#17782)
try:
from tools.skill_usage import bump_use
bump_use(skill_name)
bump_use(skill_name, task_id=task_id)
except Exception:
pass # Non-critical
@@ -790,7 +790,7 @@ def build_preloaded_skills_prompt(
# Track active usage for Curator lifecycle management (#17782)
try:
from tools.skill_usage import bump_use
bump_use(skill_name)
bump_use(skill_name, task_id=task_id)
except Exception:
pass # Non-critical
cron/scheduler.py
+1 -1
View File
@@ -2616,7 +2616,7 @@ def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str:
# Bump usage so the curator sees this skill as actively used.
try:
bump_use(skill_name)
bump_use(skill_name, task_id=str(job.get("id") or "") or None)
except Exception:
logger.debug("Cron job: failed to bump skill usage for '%s'", skill_name, exc_info=True)
docs/observability/relay-shared-metrics.md
+18 -5
View File
@@ -56,7 +56,7 @@ dependency does not change the collection or privacy policy.
## Current Slices
The current vertical slices record logical model calls, top-level task runs,
and tool and approval outcomes:
tool and approval outcomes, and skill lifecycle and reuse:
```text
Hermes turn, API, tool, and approval hooks
@@ -106,6 +106,17 @@ included in shared-metrics events or packages. A started tool that is still
open when its task terminates is closed as failed, timed out, or cancelled and
remains in the task's tool-count bucket.
Successful skill mutations emit `hermes.skill.lifecycle` marks with only a
bounded action and provenance. Successful loads emit `hermes.skill.load`
marks with bounded provenance, first-use or reuse state, reuse-after-patch
state, and a use-count bucket. Hermes derives reuse and patch-generation
continuity transactionally in its existing `skills/.usage.json` state; skill
names and exact counts or generations never enter Relay metrics events,
SQLite dimensions, or packages. A use after a new patch is counted once as
`reused_after_patch`; later uses remain ordinary reuse until another patch.
Task-outcome attribution after a patch remains deferred until its window and
multi-skill semantics are defined.
Local state is written under:
```text
@@ -148,7 +159,9 @@ The script uses the installed `nemo-relay` dependency by default. Pass
binding.
The smoke has the local model request a real `read_file` tool call before its
final response. It verifies model, provider, task, and bounded tool counters in
SQLite, validates the exported package against the closed schema, and checks
that prompt, response, tool-call ID, and tool-result canaries are absent from
the package.
final response, then drives create, load, reuse, patch, edit, stale, archive,
restore, and install skill transitions through the installed Relay binding. It
verifies model, provider, task, tool, and skill counters in SQLite, validates
all exported delta packages against the closed schema, and checks that prompt,
response, tool-call ID, tool-result, and skill-name canaries are absent from the
packages.
hermes_cli/observability/relay_shared_metrics.py
+77 -6
View File
@@ -20,11 +20,15 @@ from .shared_metrics_contract import (
MODEL_CALL_SCOPE,
SCHEMA_KEY,
SCHEMA_VERSION,
SKILL_LIFECYCLE_MARK,
SKILL_LOAD_MARK,
SUBSCRIBER_NAME,
TASK_SCOPE,
TOOL_APPROVAL_MARK,
TOOL_CALL_SCOPE,
model_call_fields,
skill_lifecycle_fields,
skill_load_fields,
task_start_fields,
task_terminal_fields,
task_terminal_state,
@@ -48,6 +52,7 @@ HANDLED_HOOKS = frozenset({
"post_approval_response",
"post_api_request",
"api_request_error",
"on_skill_lifecycle",
"subagent_stop",
})
@@ -189,6 +194,8 @@ class _Runtime:
return None
task = owner.tasks.get(task_id)
if task is not None:
if not self._event_matches_task_turn(task, event):
return None
self._remember_turn(owner, task, event)
return task
@@ -196,8 +203,10 @@ class _Runtime:
if session is None:
return None
with session.lock:
turn_id = str(event.get("turn_id") or "")
if (
session.closing
or (turn_id and turn_id in session.retired_turn_ids)
or session.relay_session.context is None
):
return None
@@ -258,6 +267,8 @@ class _Runtime:
if task is None:
task = self.start_task(event)
session = self._task_session(event) if task is not None else None
if task_id and task is None:
return
if session is None:
session = self.ensure_session(event)
if session is None:
@@ -272,6 +283,11 @@ class _Runtime:
if session.closing:
return
if task is not None:
if (
session.tasks.get(task.task_id) is not task
or not self._event_matches_task_turn(task, event)
):
return
self._remember_turn(session, task, event)
existing = session.model_calls.get(model_call_key)
if existing is not None:
@@ -465,6 +481,55 @@ class _Runtime:
tool_call = self._open_tool_call(task, event)
self._finish_tool_call(task, tool_call, event)
def record_skill_lifecycle(self, event: dict[str, Any]) -> None:
"""Emit one allowlisted skill fact without its local identity."""
action = str(event.get("action") or "").strip().lower()
if action == "loaded":
mark = SKILL_LOAD_MARK
fields = skill_load_fields(event)
else:
mark = SKILL_LIFECYCLE_MARK
fields = skill_lifecycle_fields(event)
if fields is None:
return
session_id = str(event.get("session_id") or "")
task_id = str(event.get("task_id") or "")
session = self._task_session(
event,
allow_task_id_fallback=not session_id,
)
task = session.tasks.get(task_id) if session is not None else None
if session is not None:
if task is None:
return
with session.lock:
if session.closing:
return
if (
session.tasks.get(task.task_id) is not task
or not self._event_matches_task_turn(task, event)
):
return
self._run_in_task(
task,
self.relay.scope.event,
mark,
handle=task.handle,
data=fields,
metadata=self._event_metadata(),
)
return
if session_id and task_id:
return
self.relay.get_scope_stack()
self.relay.scope.event(
mark,
data=fields,
metadata=self._event_metadata(),
)
def end_model_call(self, event: dict[str, Any]) -> None:
session = self._task_session(event, allow_task_id_fallback=True)
if session is None:
@@ -643,19 +708,23 @@ class _Runtime:
*,
allow_task_id_fallback: bool = False,
) -> _MetricsSession | None:
task_key = self._task_key(event)
if task_key is None:
session_id = str(event.get("session_id") or "")
task_id = str(event.get("task_id") or "")
if not task_id:
return None
task_key = (session_id, task_id) if session_id else None
turn_key = self._turn_key(event)
with self._task_sessions_lock:
if turn_key is not None:
owner = self._turn_sessions.get(turn_key)
if owner is not None:
return owner
owner = self._task_sessions.get(task_key)
if owner is not None or not allow_task_id_fallback:
return owner
task_id = task_key[1]
if task_key is not None:
owner = self._task_sessions.get(task_key)
if owner is not None:
return owner
if not allow_task_id_fallback:
return None
candidates: list[_MetricsSession] = []
for (_, candidate_task_id), session in self._task_sessions.items():
if candidate_task_id != task_id:
@@ -1031,6 +1100,8 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
runtime.record_tool_call(_with_runtime_toolset(kwargs))
elif hook_name == "post_approval_response":
runtime.record_approval(kwargs)
elif hook_name == "on_skill_lifecycle":
runtime.record_skill_lifecycle(kwargs)
elif hook_name == "post_api_request":
runtime.end_model_call(kwargs)
elif hook_name == "api_request_error":
hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json
+114
View File
@@ -73,6 +73,12 @@
},
{
"$ref": "#/$defs/tool_approval_counter"
},
{
"$ref": "#/$defs/skill_lifecycle_counter"
},
{
"$ref": "#/$defs/skill_load_counter"
}
]
}
@@ -539,6 +545,114 @@
"minimum": 1
}
}
},
"skill_lifecycle_counter": {
"type": "object",
"additionalProperties": false,
"required": [
"name",
"type",
"dimensions",
"value"
],
"properties": {
"name": {
"const": "hermes.skill.lifecycle.count"
},
"type": {
"const": "counter"
},
"dimensions": {
"type": "object",
"additionalProperties": false,
"required": [
"action",
"provenance"
],
"properties": {
"action": {
"enum": [
"archived",
"created",
"edited",
"installed",
"patched",
"restored",
"stale"
]
},
"provenance": {
"$ref": "#/$defs/skill_provenance"
}
}
},
"value": {
"type": "integer",
"minimum": 1
}
}
},
"skill_load_counter": {
"type": "object",
"additionalProperties": false,
"required": [
"name",
"type",
"dimensions",
"value"
],
"properties": {
"name": {
"const": "hermes.skill.load.count"
},
"type": {
"const": "counter"
},
"dimensions": {
"type": "object",
"additionalProperties": false,
"required": [
"post_patch_state",
"provenance",
"reuse_state",
"use_count_bucket"
],
"properties": {
"post_patch_state": {
"enum": [
"no_new_patch",
"not_applicable",
"reused_after_patch"
]
},
"provenance": {
"$ref": "#/$defs/skill_provenance"
},
"reuse_state": {
"enum": [
"first_use",
"reused"
]
},
"use_count_bucket": {
"$ref": "#/$defs/count_bucket"
}
}
},
"value": {
"type": "integer",
"minimum": 1
}
}
},
"skill_provenance": {
"enum": [
"agent_created",
"external",
"installed",
"local",
"unknown"
]
}
}
}
hermes_cli/observability/shared_metrics_contract.py
+118
View File
@@ -19,6 +19,8 @@ MODEL_CALL_PROFILE_MODEL = "unknown"
TASK_SCOPE = "hermes.task_run"
TOOL_CALL_SCOPE = "hermes.tool_call"
TOOL_APPROVAL_MARK = "hermes.tool_approval"
SKILL_LIFECYCLE_MARK = "hermes.skill.lifecycle"
SKILL_LOAD_MARK = "hermes.skill.load"
SUBSCRIBER_NAME = "hermes.nemo_relay.shared_metrics"
LEGACY_MODEL_CALL_METRIC = "hermes.model_call.count"
MODEL_ROUTE_METRIC = "hermes.model_route.count"
@@ -26,6 +28,8 @@ TASK_STARTED_METRIC = "hermes.task_run.started"
TASK_FINISHED_METRIC = "hermes.task_run.finished"
TOOL_CALL_METRIC = "hermes.tool_call.count"
TOOL_APPROVAL_METRIC = "hermes.tool_approval.count"
SKILL_LIFECYCLE_METRIC = "hermes.skill.lifecycle.count"
SKILL_LOAD_METRIC = "hermes.skill.load.count"
MODEL_IDENTIFIER_MAX_LENGTH = 256
PROVIDER_IDENTIFIER_MAX_LENGTH = 64
_METRIC_IDENTIFIER_CHARACTERS = frozenset(
@@ -152,6 +156,28 @@ TOOL_LATENCY_BUCKETS: frozenset[str] = frozenset({
"unknown",
})
TOOL_RETRY_BUCKETS: frozenset[str] = COUNT_BUCKETS | frozenset({"unknown"})
SKILL_LIFECYCLE_ACTIONS: frozenset[str] = frozenset({
"archived",
"created",
"edited",
"installed",
"patched",
"restored",
"stale",
})
SKILL_PROVENANCES: frozenset[str] = frozenset({
"agent_created",
"external",
"installed",
"local",
"unknown",
})
SKILL_REUSE_STATES: frozenset[str] = frozenset({"first_use", "reused"})
SKILL_POST_PATCH_STATES: frozenset[str] = frozenset({
"no_new_patch",
"not_applicable",
"reused_after_patch",
})
_LEGACY_PROVIDER_FAMILIES = frozenset({
"aggregator",
@@ -221,9 +247,21 @@ _COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = {
"attribution": TOOL_APPROVAL_ATTRIBUTIONS,
"outcome": TOOL_APPROVAL_OUTCOMES - {"not_required"},
},
SKILL_LIFECYCLE_METRIC: {
"action": SKILL_LIFECYCLE_ACTIONS,
"provenance": SKILL_PROVENANCES,
},
SKILL_LOAD_METRIC: {
"post_patch_state": SKILL_POST_PATCH_STATES,
"provenance": SKILL_PROVENANCES,
"reuse_state": SKILL_REUSE_STATES,
"use_count_bucket": COUNT_BUCKETS,
},
}
COUNTER_METRICS: frozenset[str] = frozenset({
MODEL_ROUTE_METRIC,
SKILL_LIFECYCLE_METRIC,
SKILL_LOAD_METRIC,
TASK_FINISHED_METRIC,
TASK_STARTED_METRIC,
TOOL_APPROVAL_METRIC,
@@ -447,6 +485,86 @@ def tool_approval_counter(event: Any) -> tuple[str, dict[str, str]] | None:
return TOOL_APPROVAL_METRIC, dimensions
def skill_counter(event: Any) -> tuple[str, dict[str, str]] | None:
"""Return one validated skill lifecycle or load counter from a safe mark."""
if not _event_metadata_is_valid(event):
return None
if (
str(getattr(event, "kind", "") or "") != "mark"
or getattr(event, "category", None) is not None
or getattr(event, "scope_category", None) is not None
or getattr(event, "category_profile", None) is not None
):
return None
name = str(getattr(event, "name", "") or "")
data = getattr(event, "data", None)
if name == SKILL_LIFECYCLE_MARK:
metric_name = SKILL_LIFECYCLE_METRIC
expected_fields = {"action", "provenance"}
elif name == SKILL_LOAD_MARK:
metric_name = SKILL_LOAD_METRIC
expected_fields = {
"post_patch_state",
"provenance",
"reuse_state",
"use_count_bucket",
}
else:
return None
if not isinstance(data, dict) or set(data) != expected_fields:
return None
dimensions = {field: data.get(field) for field in sorted(expected_fields)}
if not counter_dimensions_are_valid(metric_name, dimensions):
return None
return metric_name, dimensions
def skill_lifecycle_fields(kwargs: dict[str, Any]) -> dict[str, str] | None:
"""Build bounded fields for one successful non-load skill transition."""
action = str(kwargs.get("action") or "").strip().lower()
if action not in SKILL_LIFECYCLE_ACTIONS:
return None
return {
"action": action,
"provenance": skill_provenance(kwargs.get("provenance")),
}
def skill_load_fields(kwargs: dict[str, Any]) -> dict[str, str] | None:
"""Build bounded skill-use fields without exporting local skill identity."""
use_count = kwargs.get("use_count")
reused = kwargs.get("reused")
reuse_after_patch = kwargs.get("reuse_after_patch")
if (
isinstance(use_count, bool)
or not isinstance(use_count, int)
or use_count < 1
or not isinstance(reused, bool)
or not isinstance(reuse_after_patch, bool)
or (reuse_after_patch and not reused)
):
return None
return {
"post_patch_state": (
"not_applicable"
if not reused
else "reused_after_patch"
if reuse_after_patch
else "no_new_patch"
),
"provenance": skill_provenance(kwargs.get("provenance")),
"reuse_state": "reused" if reused else "first_use",
"use_count_bucket": count_bucket(use_count),
}
def skill_provenance(value: Any) -> str:
"""Normalize producer provenance to the closed shared-metrics taxonomy."""
normalized = str(value or "").strip().lower()
return normalized if normalized in SKILL_PROVENANCES else "unknown"
def execution_surface(kwargs: dict[str, Any]) -> str:
"""Normalize the safe session surface carried by the parent Relay scope."""
value = (
hermes_cli/observability/shared_metrics_subscriber.py
+6 -1
View File
@@ -13,6 +13,7 @@ from .shared_metrics_contract import (
MODEL_ROUTE_METRIC,
TOOL_CALL_METRIC,
model_call_dimensions,
skill_counter,
task_counter,
tool_approval_counter,
tool_call_dimensions,
@@ -56,7 +57,11 @@ class SharedMetricsSubscriber:
dimensions = tool_call_dimensions(event)
metric_name = TOOL_CALL_METRIC
if dimensions is None:
metric = task_counter(event) or tool_approval_counter(event)
metric = (
task_counter(event)
or tool_approval_counter(event)
or skill_counter(event)
)
if metric is None:
return
metric_name, dimensions = metric
hermes_cli/plugins.py
+3
View File
@@ -161,6 +161,9 @@ VALID_HOOKS: Set[str] = {
"on_session_end",
"on_session_finalize",
"on_session_reset",
# Successful skill lifecycle facts. The local skill name is available to
# plugins, while built-in shared metrics emit only bounded classifications.
"on_skill_lifecycle",
"subagent_start",
"subagent_stop",
# Gateway pre-dispatch hook. Fired once per incoming MessageEvent
scripts/smoke_nemo_relay_shared_metrics.py
+171 -19
View File
@@ -23,6 +23,8 @@ RESPONSE_CANARY = "relay-smoke-sensitive-response"
TOOL_CALL_CANARY = "relay-smoke-sensitive-tool-call"
TOOL_RESULT_CANARY = "relay-smoke-sensitive-tool-result"
TOOL_FILE = "relay-smoke-input.txt"
SKILL_CANARY = "relay-smoke-private-agent-skill"
INSTALLED_SKILL_CANARY = "relay-smoke-private-installed-skill"
def _resolve_hermes_executable(hermes_repo: Path) -> Path:
@@ -297,6 +299,8 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]:
by_name.setdefault(counter["name"], []).append(counter)
if set(by_name) != {
"hermes.model_route.count",
"hermes.skill.lifecycle.count",
"hermes.skill.load.count",
"hermes.task_run.finished",
"hermes.task_run.started",
"hermes.tool_call.count",
@@ -364,15 +368,54 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]:
or tool["packaged_value"] != 1
):
raise AssertionError(f"Unexpected tool counter: {tool}")
lifecycle = by_name["hermes.skill.lifecycle.count"]
expected_actions = {
"archived",
"created",
"edited",
"installed",
"patched",
"restored",
"stale",
}
if (
{counter["dimensions"]["action"] for counter in lifecycle} != expected_actions
or any(counter["value"] != 1 for counter in lifecycle)
or any(counter["packaged_value"] != 1 for counter in lifecycle)
):
raise AssertionError(f"Unexpected skill lifecycle counters: {lifecycle}")
loads = by_name["hermes.skill.load.count"]
expected_load_states = {
("first_use", "not_applicable", "1"),
("reused", "no_new_patch", "2"),
("reused", "reused_after_patch", "3_to_5"),
}
observed_load_states = {
(
counter["dimensions"]["reuse_state"],
counter["dimensions"]["post_patch_state"],
counter["dimensions"]["use_count_bucket"],
)
for counter in loads
}
if (
observed_load_states != expected_load_states
or any(counter["value"] != 1 for counter in loads)
or any(counter["packaged_value"] != 1 for counter in loads)
):
raise AssertionError(f"Unexpected skill load counters: {loads}")
return counters
def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str, Any]]:
packages = sorted(outbox.glob("*.json"))
if len(packages) != 1:
raise AssertionError(f"Expected one package in {outbox}, found {len(packages)}")
package_path = packages[0]
package = json.loads(package_path.read_text(encoding="utf-8"))
def _validate_packages(
outbox: Path,
schema_path: Path,
) -> tuple[list[Path], list[dict[str, Any]]]:
package_paths = sorted(outbox.glob("*.json"))
if len(package_paths) != 2:
raise AssertionError(
f"Expected two delta packages in {outbox}, found {len(package_paths)}"
)
try:
import jsonschema
except ImportError as exc:
@@ -380,30 +423,40 @@ def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str,
"The Hermes development environment requires jsonschema"
) from exc
schema = json.loads(schema_path.read_text(encoding="utf-8"))
jsonschema.validate(package, schema)
packages = [
json.loads(package_path.read_text(encoding="utf-8"))
for package_path in package_paths
]
for package in packages:
jsonschema.validate(package, schema)
serialized = json.dumps(package)
serialized = json.dumps(packages)
for prohibited in (
PROMPT_CANARY,
RESPONSE_CANARY,
TOOL_CALL_CANARY,
TOOL_RESULT_CANARY,
SKILL_CANARY,
INSTALLED_SKILL_CANARY,
):
if prohibited in serialized:
raise AssertionError(
f"Exported package leaked prohibited value: {prohibited!r}"
)
metrics: dict[str, list[dict[str, Any]]] = {}
for metric in package.get("metrics", []):
metrics.setdefault(metric["name"], []).append(metric)
for package in packages:
for metric in package.get("metrics", []):
metrics.setdefault(metric["name"], []).append(metric)
if set(metrics) != {
"hermes.model_route.count",
"hermes.skill.lifecycle.count",
"hermes.skill.load.count",
"hermes.task_run.finished",
"hermes.task_run.started",
"hermes.tool_call.count",
}:
raise AssertionError(
f"Unexpected package metrics:\n{json.dumps(package.get('metrics'), indent=2)}"
f"Unexpected package metrics:\n{json.dumps(metrics, indent=2)}"
)
[model] = metrics["hermes.model_route.count"]
if model["dimensions"] != {
@@ -439,7 +492,32 @@ def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str,
or tool["dimensions"]["latency_bucket"] == "unknown"
):
raise AssertionError(f"Unexpected tool metric: {tool}")
return package_path, package
lifecycle = metrics["hermes.skill.lifecycle.count"]
if {metric["dimensions"]["action"] for metric in lifecycle} != {
"archived",
"created",
"edited",
"installed",
"patched",
"restored",
"stale",
}:
raise AssertionError(f"Unexpected skill lifecycle metrics: {lifecycle}")
loads = metrics["hermes.skill.load.count"]
if {
(
metric["dimensions"]["reuse_state"],
metric["dimensions"]["post_patch_state"],
metric["dimensions"]["use_count_bucket"],
)
for metric in loads
} != {
("first_use", "not_applicable", "1"),
("reused", "no_new_patch", "2"),
("reused", "reused_after_patch", "3_to_5"),
}:
raise AssertionError(f"Unexpected skill load metrics: {loads}")
return package_paths, packages
def main() -> int:
@@ -468,6 +546,29 @@ def main() -> int:
(workdir / TOOL_FILE).write_text(TOOL_RESULT_CANARY, encoding="utf-8")
home.mkdir()
(home / ".no-bundled-skills").touch()
agent_skill = home / "skills" / SKILL_CANARY
agent_skill.mkdir(parents=True)
(agent_skill / "SKILL.md").write_text(
f"---\nname: {SKILL_CANARY}\ndescription: private smoke skill\n---\n",
encoding="utf-8",
)
installed_skill = home / "skills" / INSTALLED_SKILL_CANARY
installed_skill.mkdir(parents=True)
(installed_skill / "SKILL.md").write_text(
f"---\nname: {INSTALLED_SKILL_CANARY}\ndescription: installed smoke skill\n---\n",
encoding="utf-8",
)
hub_state = home / "skills" / ".hub"
hub_state.mkdir()
(hub_state / "lock.json").write_text(
json.dumps({
"version": 1,
"installed": {
INSTALLED_SKILL_CANARY: {"source": "smoke/local"},
},
}),
encoding="utf-8",
)
_ModelHandler.requests = []
server = ThreadingHTTPServer(("127.0.0.1", 0), _ModelHandler)
@@ -477,11 +578,11 @@ def main() -> int:
_write_config(home, server.server_port)
env = os.environ.copy()
env["HERMES_HOME"] = str(home)
python_paths = [str(hermes_repo)]
if relay_python is not None:
env["PYTHONPATH"] = os.pathsep.join([
str(relay_python),
env.get("PYTHONPATH", ""),
]).rstrip(os.pathsep)
python_paths.append(str(relay_python))
python_paths.append(env.get("PYTHONPATH", ""))
env["PYTHONPATH"] = os.pathsep.join(python_paths).rstrip(os.pathsep)
result = subprocess.run(
[
str(hermes),
@@ -532,9 +633,60 @@ def main() -> int:
if RESPONSE_CANARY not in result.stdout:
raise AssertionError("Hermes did not print the mock model response")
skill_result = subprocess.run(
[
sys.executable,
"-c",
"\n".join([
"from hermes_cli.observability import relay_shared_metrics",
"from tools.skill_usage import (",
" STATE_ACTIVE, STATE_ARCHIVED, STATE_STALE, bump_patch,",
" bump_use, record_created, record_installed, set_state,",
")",
f"skill = {SKILL_CANARY!r}",
f"installed = {INSTALLED_SKILL_CANARY!r}",
"record_created(skill, agent_created=True)",
"bump_use(skill)",
"bump_use(skill)",
"bump_patch(skill)",
"bump_use(skill)",
"bump_patch(skill, action='edit')",
"set_state(skill, STATE_STALE)",
"set_state(skill, STATE_ACTIVE)",
"set_state(skill, STATE_ARCHIVED)",
"set_state(skill, STATE_ACTIVE)",
"record_installed(installed)",
"runtime = relay_shared_metrics._get_runtime()",
"assert runtime is not None",
"runtime.shutdown()",
# Production leaves same-day deltas pending. Force a package so
# this smoke can validate them without waiting for the next day.
"runtime.subscriber.store.create_and_export_package()",
]),
],
cwd=workdir,
env=env,
text=True,
capture_output=True,
timeout=60,
)
(root / "skills.stdout.txt").write_text(
skill_result.stdout,
encoding="utf-8",
)
(root / "skills.stderr.txt").write_text(
skill_result.stderr,
encoding="utf-8",
)
if skill_result.returncode != 0:
raise AssertionError(
f"Skill lifecycle probe exited with {skill_result.returncode}\n"
f"stdout:\n{skill_result.stdout}\nstderr:\n{skill_result.stderr}"
)
telemetry = home / "telemetry" / "shared_metrics"
counters = _validate_store(telemetry / "metrics.sqlite3")
package_path, package = _validate_package(
package_paths, packages = _validate_packages(
telemetry / "outbox",
hermes_repo
/ "hermes_cli"
@@ -547,8 +699,8 @@ def main() -> int:
print(f"Artifact directory: {root}")
print(f"Model requests: {len(_ModelHandler.requests)}")
print(f"SQLite counters: {json.dumps(counters, indent=2)}")
print(f"Export package: {package_path}")
print(json.dumps(package, indent=2, sort_keys=True))
print(f"Export packages: {', '.join(str(path) for path in package_paths)}")
print(json.dumps(packages, indent=2, sort_keys=True))
return 0
tests/agent/test_skill_bundles.py
+23
View File
@@ -160,6 +160,29 @@ class TestBuildBundleInvocationMessage:
assert "Skill B content." in msg
assert "combo" in msg
def test_forwards_task_id_to_each_loaded_skill(self, bundles_env, monkeypatch):
bundles_dir, skills_dir = bundles_env
_make_skill(skills_dir, "skill-a")
_make_skill(skills_dir, "skill-b")
_make_bundle_yaml(bundles_dir, "combo", ["skill-a", "skill-b"])
scan_bundles()
calls = []
monkeypatch.setattr(
"tools.skill_usage.bump_use",
lambda skill_name, **kwargs: calls.append((skill_name, kwargs)),
)
result = build_bundle_invocation_message(
"/combo",
task_id="task-bundle",
)
assert result is not None
assert calls == [
("skill-a", {"task_id": "task-bundle"}),
("skill-b", {"task_id": "task-bundle"}),
]
def test_skips_missing_skills(self, bundles_env):
bundles_dir, skills_dir = bundles_env
_make_skill(skills_dir, "skill-a")
tests/agent/test_skill_commands.py
+57 -3
View File
@@ -333,6 +333,24 @@ class TestBuildPreloadedSkillsPrompt:
assert "second-skill" in prompt
assert "preloaded" in prompt.lower()
def test_forwards_task_id_to_skill_usage(self, tmp_path):
with (
patch("tools.skills_tool.SKILLS_DIR", tmp_path),
patch("tools.skill_usage.bump_use") as bump_use,
):
_make_skill(tmp_path, "preloaded-skill")
_prompt, loaded, missing = build_preloaded_skills_prompt(
["preloaded-skill"],
task_id="task-preloaded",
)
assert loaded == ["preloaded-skill"]
assert missing == []
bump_use.assert_called_once_with(
"preloaded-skill",
task_id="task-preloaded",
)
def test_skips_disabled_skill(self, tmp_path, monkeypatch):
"""A globally-disabled skill must not be force-loaded via -s /
@@ -361,6 +379,21 @@ class TestBuildSkillInvocationMessage:
def test_forwards_task_id_to_skill_usage(self, tmp_path):
with (
patch("tools.skills_tool.SKILLS_DIR", tmp_path),
patch("tools.skill_usage.bump_use") as bump_use,
):
_make_skill(tmp_path, "test-skill")
scan_skill_commands()
msg = build_skill_invocation_message(
"/test-skill",
task_id="task-slash",
)
assert msg is not None
bump_use.assert_called_once_with("test-skill", task_id="task-slash")
def test_uses_shared_skill_loader_for_secure_setup(self, tmp_path, monkeypatch):
monkeypatch.delenv("TENOR_API_KEY", raising=False)
@@ -620,6 +653,30 @@ class TestStackedSkillCommands:
def test_stacked_message_forwards_task_id_to_each_skill(self, tmp_path):
from agent.skill_commands import build_stacked_skill_invocation_message
with (
patch("tools.skills_tool.SKILLS_DIR", tmp_path),
patch("tools.skill_usage.bump_use") as bump_use,
):
self._setup_three_skills(tmp_path)
scan_skill_commands()
result = build_stacked_skill_invocation_message(
["/skill-a", "/skill-b"],
task_id="task-stacked",
)
assert result is not None
assert [call.args[0] for call in bump_use.call_args_list] == [
"skill-a",
"skill-b",
]
assert all(
call.kwargs == {"task_id": "task-stacked"}
for call in bump_use.call_args_list
)
def test_stacked_message_skips_missing_skills(self, tmp_path):
from agent.skill_commands import build_stacked_skill_invocation_message
with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
@@ -633,6 +690,3 @@ class TestStackedSkillCommands:
assert loaded == ["skill-a"]
assert missing == ["gone"]
assert "Skills missing (skipped): gone" in msg
tests/cron/test_scheduler.py
+9 -1
View File
@@ -1149,12 +1149,20 @@ class TestBuildJobPromptBumpUse:
with patch("tools.skills_tool.skill_view", side_effect=_skill_view), \
patch("tools.skill_usage.bump_use") as mock_bump:
_build_job_prompt({"skills": ["alpha", "beta"], "prompt": "go"})
_build_job_prompt({
"id": "cron-task",
"skills": ["alpha", "beta"],
"prompt": "go",
})
assert mock_bump.call_count == 2
calls = [c[0][0] for c in mock_bump.call_args_list]
assert "alpha" in calls
assert "beta" in calls
assert all(
call.kwargs == {"task_id": "cron-task"}
for call in mock_bump.call_args_list
)
class TestSendMediaViaAdapter:
tests/hermes_cli/test_relay_shared_metrics.py
+93
View File
@@ -32,6 +32,10 @@ from hermes_cli.observability.shared_metrics_contract import (
PROVIDER_IDENTIFIER_MAX_LENGTH,
SCHEMA_KEY,
SCHEMA_VERSION,
SKILL_LIFECYCLE_ACTIONS,
SKILL_POST_PATCH_STATES,
SKILL_PROVENANCES,
SKILL_REUSE_STATES,
TASK_END_REASONS,
TASK_ENTRYPOINTS,
TASK_OUTCOMES,
@@ -47,6 +51,9 @@ from hermes_cli.observability.shared_metrics_contract import (
execution_surface,
model_call_dimensions,
model_call_fields,
skill_counter,
skill_lifecycle_fields,
skill_load_fields,
task_counter,
task_start_fields,
task_terminal_fields,
@@ -368,6 +375,18 @@ def test_package_schema_matches_the_tool_contract():
)
def test_package_schema_matches_the_skill_contract():
schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
lifecycle = _tool_dimension_schema("skill_lifecycle_counter")["properties"]
load = _tool_dimension_schema("skill_load_counter")["properties"]
assert set(lifecycle["action"]["enum"]) == SKILL_LIFECYCLE_ACTIONS
assert set(schema["$defs"]["skill_provenance"]["enum"]) == SKILL_PROVENANCES
assert set(load["reuse_state"]["enum"]) == SKILL_REUSE_STATES
assert set(load["post_patch_state"]["enum"]) == SKILL_POST_PATCH_STATES
assert load["use_count_bucket"] == {"$ref": "#/$defs/count_bucket"}
@pytest.mark.parametrize(
("toolset", "expected"),
[
@@ -611,6 +630,80 @@ def test_tool_subscriber_contract_accepts_only_bounded_events():
assert tool_approval_counter(approval) is None
def test_skill_subscriber_contract_accepts_only_bounded_marks():
metadata = {"hermes.metrics.schema_version": "hermes.metrics.event.v2"}
lifecycle = SimpleNamespace(
kind="mark",
category=None,
category_profile=None,
name="hermes.skill.lifecycle",
scope_category=None,
metadata=metadata,
data={"action": "patched", "provenance": "agent_created"},
)
assert skill_counter(lifecycle) == (
"hermes.skill.lifecycle.count",
lifecycle.data,
)
load = SimpleNamespace(**{
**lifecycle.__dict__,
"name": "hermes.skill.load",
"data": {
"post_patch_state": "reused_after_patch",
"provenance": "agent_created",
"reuse_state": "reused",
"use_count_bucket": "3_to_5",
},
})
assert skill_counter(load) == ("hermes.skill.load.count", load.data)
load.data["skill_name"] = "privacy-canary"
assert skill_counter(load) is None
load.data.pop("skill_name")
load.data["provenance"] = "private-repository"
assert skill_counter(load) is None
lifecycle.metadata["skill_name"] = "privacy-canary"
assert skill_counter(lifecycle) is None
def test_skill_event_fields_are_bounded_and_reject_malformed_usage():
assert skill_lifecycle_fields({
"action": "patched",
"provenance": "agent_created",
"skill_name": "privacy-canary",
}) == {"action": "patched", "provenance": "agent_created"}
assert skill_lifecycle_fields({"action": "deleted"}) is None
assert skill_load_fields({
"provenance": "private-repository",
"use_count": 2,
"reused": True,
"reuse_after_patch": False,
"skill_name": "privacy-canary",
}) == {
"post_patch_state": "no_new_patch",
"provenance": "unknown",
"reuse_state": "reused",
"use_count_bucket": "2",
}
assert skill_load_fields({
"use_count": 1,
"reused": False,
"reuse_after_patch": False,
}) == {
"post_patch_state": "not_applicable",
"provenance": "unknown",
"reuse_state": "first_use",
"use_count_bucket": "1",
}
assert (
skill_load_fields({
"use_count": 1,
"reused": False,
"reuse_after_patch": True,
})
is None
)
def test_store_does_not_record_the_retired_model_metric(tmp_path):
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
tests/hermes_cli/test_relay_shared_metrics_runtime.py
+202
View File
@@ -2061,6 +2061,40 @@ def test_pending_tool_uses_the_outer_task_timeout_outcome(direct_runtime):
assert "private timeout detail" not in repr(direct_runtime.events)
def test_late_model_start_does_not_create_an_orphan_after_task_completion(
direct_runtime,
):
base = {
"session_id": "s1",
"task_id": "t1",
"turn_id": "turn-1",
"platform": "cli",
}
lifecycle.invoke_hook("pre_llm_call", **base)
lifecycle.invoke_hook(
"on_session_end",
**base,
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
lifecycle.invoke_hook(
"pre_api_request",
**base,
api_request_id="late-request",
provider="nvidia",
model="nvidia/nemotron-3-super-120b-a12b",
)
lifecycle.finalize_session(session_id="s1")
assert [event for event in direct_runtime.events if event[0] == "llm.call"] == []
assert [
event for event in direct_runtime.events if event[0] == "llm.call_end"
] == []
def test_approval_without_tool_context_is_counted_as_unattributed(direct_runtime):
base = {
"session_id": "s1",
@@ -2293,3 +2327,171 @@ def test_failed_flush_keeps_daily_export_open_for_later_task(
assert flush_attempts == 2
assert "Hermes shared-metrics task flush failed" in caplog.text
def test_skill_lifecycle_flows_through_relay_to_a_privacy_safe_package(
direct_runtime,
tmp_path,
):
common = {
"skill_name": "private-skill-name",
"provenance": "agent_created",
}
lifecycle.invoke_hook("on_skill_lifecycle", **common, action="created")
lifecycle.invoke_hook(
"on_skill_lifecycle",
**common,
action="loaded",
use_count=1,
reused=False,
reuse_after_patch=False,
)
lifecycle.invoke_hook("on_skill_lifecycle", **common, action="patched")
lifecycle.invoke_hook(
"on_skill_lifecycle",
**common,
action="loaded",
use_count=2,
reused=True,
reuse_after_patch=True,
)
runtime = relay_shared_metrics._get_runtime()
assert runtime is not None
runtime.shutdown()
marks = [event for event in direct_runtime.events if event[0] == "scope.event"]
assert [event[1] for event in marks] == [
"hermes.skill.lifecycle",
"hermes.skill.load",
"hermes.skill.lifecycle",
"hermes.skill.load",
]
assert "private-skill-name" not in json.dumps(marks)
outbox = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" / "outbox"
[package_path] = list(outbox.glob("*.json"))
package = json.loads(package_path.read_text(encoding="utf-8"))
skill_metrics = [
metric
for metric in package["metrics"]
if metric["name"].startswith("hermes.skill.")
]
assert {metric["name"] for metric in skill_metrics} == {
"hermes.skill.lifecycle.count",
"hermes.skill.load.count",
}
assert "private-skill-name" not in json.dumps(package)
def test_skill_lifecycle_with_only_task_id_uses_unique_task_scope(direct_runtime):
runtime = relay_shared_metrics._get_runtime()
assert runtime is not None
task = runtime.start_task({
"session_id": "session-1",
"task_id": "task-1",
"platform": "cli",
})
assert task is not None
lifecycle.invoke_hook(
"on_skill_lifecycle",
action="created",
skill_name="private-skill-name",
provenance="agent_created",
task_id="task-1",
)
[mark] = [
event
for event in direct_runtime.events
if event[0] == "scope.event" and event[1] == "hermes.skill.lifecycle"
]
assert mark[2]["handle"] == task.handle
def test_skill_task_only_correlation_does_not_guess_across_sessions(direct_runtime):
runtime = relay_shared_metrics._get_runtime()
assert runtime is not None
for session_id in ("session-1", "session-2"):
assert runtime.start_task({
"session_id": session_id,
"task_id": "shared-task",
"platform": "cli",
}) is not None
lifecycle.invoke_hook(
"on_skill_lifecycle",
action="created",
skill_name="private-skill-name",
provenance="agent_created",
task_id="shared-task",
)
[mark] = [
event
for event in direct_runtime.events
if event[0] == "scope.event" and event[1] == "hermes.skill.lifecycle"
]
assert "handle" not in mark[2]
def test_late_skill_lifecycle_is_not_reemitted_at_the_root(direct_runtime):
base = {
"session_id": "session-1",
"task_id": "task-1",
"turn_id": "turn-1",
"platform": "cli",
}
lifecycle.invoke_hook("pre_llm_call", **base)
lifecycle.invoke_hook(
"on_session_end",
**base,
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
lifecycle.invoke_hook(
"on_skill_lifecycle",
**base,
action="loaded",
skill_name="private-skill-name",
provenance="local",
use_count=2,
reused=True,
reuse_after_patch=False,
)
assert [
event
for event in direct_runtime.events
if event[0] == "scope.event" and event[1] == "hermes.skill.load"
] == []
def test_skill_lifecycle_does_not_fallback_across_an_explicit_session(
direct_runtime,
):
runtime = relay_shared_metrics._get_runtime()
assert runtime is not None
assert runtime.start_task({
"session_id": "session-1",
"task_id": "task-1",
"platform": "cli",
}) is not None
lifecycle.invoke_hook(
"on_skill_lifecycle",
action="created",
skill_name="private-skill-name",
provenance="local",
session_id="wrong-session",
task_id="task-1",
)
assert [
event
for event in direct_runtime.events
if event[0] == "scope.event" and event[1] == "hermes.skill.lifecycle"
] == []
tests/test_plugin_skills.py
+29
View File
@@ -178,6 +178,35 @@ class TestSkillViewQualifiedName:
assert result["name"] == "superpowers:writing-plans"
assert "writing-plans body." in result["content"]
def test_plugin_skill_usage_reports_installed_provenance(
self,
tmp_path,
monkeypatch,
):
from hermes_cli import lifecycle
from tools.skills_tool import _skill_view_with_bump
events = []
monkeypatch.setattr(lifecycle, "has_hook", lambda name: True)
monkeypatch.setattr(
lifecycle,
"invoke_hook",
lambda name, **kwargs: events.append((name, kwargs)),
)
self._register_skill(tmp_path)
result = json.loads(
_skill_view_with_bump(
{"name": "superpowers:writing-plans"},
task_id="task-1",
session_id="session-1",
)
)
assert result["success"] is True
[loaded] = [event for _, event in events if event["action"] == "loaded"]
assert loaded["provenance"] == "installed"
def test_invalid_namespace_returns_error(self, tmp_path):
from tools.skills_tool import skill_view
tests/tools/test_skill_manager_tool.py
+73
View File
@@ -360,6 +360,79 @@ class TestSkillManageDispatcher:
rec = usage.get("test-skill") or {}
assert rec.get("created_by") in {None, "", False}
def test_successful_mutations_emit_lifecycle_with_correlation(self, tmp_path):
with (
_skill_dir(tmp_path),
patch("tools.skill_provenance.is_background_review", return_value=False),
patch("tools.skill_usage.record_created") as record_created,
patch("tools.skill_usage.bump_patch") as bump_patch,
):
created = json.loads(skill_manage(
action="create",
name="test-skill",
content=VALID_SKILL_CONTENT,
task_id="task-mutation",
session_id="session-mutation",
))
patched = json.loads(skill_manage(
action="patch",
name="test-skill",
old_string="Step 1: Do the thing.",
new_string="Step 1: Do the thing safely.",
task_id="task-mutation",
session_id="session-mutation",
))
edited = json.loads(skill_manage(
action="edit",
name="test-skill",
content=VALID_SKILL_CONTENT_2,
task_id="task-mutation",
session_id="session-mutation",
))
assert created["success"] is True
assert patched["success"] is True
assert edited["success"] is True
record_created.assert_called_once_with(
"test-skill",
agent_created=False,
task_id="task-mutation",
session_id="session-mutation",
)
assert [call.kwargs for call in bump_patch.call_args_list] == [
{
"action": "patch",
"task_id": "task-mutation",
"session_id": "session-mutation",
},
{
"action": "edit",
"task_id": "task-mutation",
"session_id": "session-mutation",
},
]
assert all(call.args == ("test-skill",) for call in bump_patch.call_args_list)
def test_failed_mutations_do_not_emit_lifecycle(self, tmp_path):
with (
_skill_dir(tmp_path),
patch("tools.skill_usage.record_created") as record_created,
patch("tools.skill_usage.bump_patch") as bump_patch,
):
create_result = json.loads(skill_manage(
action="create",
name="test-skill",
))
patch_result = json.loads(skill_manage(
action="patch",
name="test-skill",
))
assert create_result["success"] is False
assert patch_result["success"] is False
record_created.assert_not_called()
bump_patch.assert_not_called()
def test_background_review_delete_refuses_bundled_even_with_absorbed_into(self, tmp_path):
from tools.skill_provenance import (
tests/tools/test_skill_usage.py
+190 -1
View File
@@ -105,6 +105,196 @@ def test_bump_view_increments_and_timestamps(skills_home):
assert rec["last_viewed_at"] is not None
def test_skill_reuse_and_post_patch_reuse_are_derived_atomically(
skills_home,
monkeypatch,
):
from hermes_cli import lifecycle
from tools.skill_usage import bump_patch, bump_use, get_record, record_created
events = []
monkeypatch.setattr(lifecycle, "has_hook", lambda name: True)
monkeypatch.setattr(
lifecycle,
"invoke_hook",
lambda name, **kwargs: events.append((name, kwargs)),
)
record_created("private-skill-name", agent_created=True, task_id="task")
bump_use("private-skill-name", task_id="task")
bump_use("private-skill-name", task_id="task")
bump_patch("private-skill-name", task_id="task")
bump_use("private-skill-name", task_id="task")
bump_use("private-skill-name", task_id="task")
loaded = [event for _, event in events if event["action"] == "loaded"]
assert [event["reused"] for event in loaded] == [False, True, True, True]
assert [event["reuse_after_patch"] for event in loaded] == [
False,
False,
True,
False,
]
assert all(event["provenance"] == "agent_created" for event in loaded)
record = get_record("private-skill-name")
assert record["use_count"] == 4
assert record["patch_generation"] == 1
assert record["last_reused_patch_generation"] == 1
def test_skill_state_events_emit_only_for_real_transitions(skills_home, monkeypatch):
from hermes_cli import lifecycle
from tools.skill_usage import (
STATE_ACTIVE,
STATE_ARCHIVED,
STATE_STALE,
record_created,
set_state,
)
events = []
monkeypatch.setattr(lifecycle, "has_hook", lambda name: True)
monkeypatch.setattr(
lifecycle,
"invoke_hook",
lambda name, **kwargs: events.append(kwargs),
)
record_created("my-skill", agent_created=True)
set_state("my-skill", STATE_STALE)
set_state("my-skill", STATE_STALE)
set_state("my-skill", STATE_ARCHIVED)
set_state("my-skill", STATE_ARCHIVED)
set_state("my-skill", STATE_ACTIVE)
set_state("my-skill", STATE_ACTIVE)
assert [event["action"] for event in events] == [
"created",
"stale",
"archived",
"restored",
]
def test_skill_event_is_not_emitted_when_usage_state_cannot_commit(
skills_home,
monkeypatch,
):
from hermes_cli import lifecycle
from tools import skill_usage
events = []
monkeypatch.setattr(lifecycle, "has_hook", lambda name: True)
monkeypatch.setattr(
lifecycle,
"invoke_hook",
lambda name, **kwargs: events.append(kwargs),
)
monkeypatch.setattr(skill_usage, "save_usage", lambda data: False)
skill_usage.bump_use("private-skill-name")
assert events == []
def test_installed_lifecycle_uses_persisted_provenance_when_hub_lookup_misses(
skills_home,
monkeypatch,
):
from hermes_cli import lifecycle
from tools import skill_usage
events = []
monkeypatch.setattr(lifecycle, "has_hook", lambda name: True)
monkeypatch.setattr(
lifecycle,
"invoke_hook",
lambda name, **kwargs: events.append(kwargs),
)
monkeypatch.setattr(skill_usage, "is_hub_installed", lambda _name: False)
monkeypatch.setattr(skill_usage, "is_bundled", lambda _name: False)
skill_usage.record_installed("private-installed-skill")
assert len(events) == 1
assert events[0]["action"] == "installed"
assert events[0]["provenance"] == "installed"
def test_created_skill_does_not_inherit_stale_identity_or_continuity(
skills_home,
monkeypatch,
):
from hermes_cli import lifecycle
from tools import skill_usage
events = []
monkeypatch.setattr(lifecycle, "has_hook", lambda name: True)
monkeypatch.setattr(
lifecycle,
"invoke_hook",
lambda name, **kwargs: events.append(kwargs),
)
skill_usage.save_usage({
"recreated": {
"created_by": "agent",
"use_count": 11,
"patch_count": 4,
"patch_generation": 4,
"last_reused_patch_generation": 3,
"pinned": True,
"state": skill_usage.STATE_ARCHIVED,
}
})
skill_usage.record_created("recreated", agent_created=False)
skill_usage.bump_use("recreated")
record = skill_usage.get_record("recreated")
assert record["created_by"] is None
assert record["use_count"] == 1
assert record["patch_count"] == 0
assert record["patch_generation"] == 0
assert record["last_reused_patch_generation"] == 0
assert record["pinned"] is False
assert record["state"] == skill_usage.STATE_ACTIVE
assert [event["provenance"] for event in events] == ["local", "local"]
assert events[-1]["reused"] is False
assert events[-1]["reuse_after_patch"] is False
def test_malformed_usage_counters_recover_without_losing_patch_reuse(
skills_home,
monkeypatch,
):
from hermes_cli import lifecycle
from tools import skill_usage
events = []
monkeypatch.setattr(lifecycle, "has_hook", lambda name: True)
monkeypatch.setattr(
lifecycle,
"invoke_hook",
lambda name, **kwargs: events.append(kwargs),
)
skill_usage.save_usage({
"damaged": {
"view_count": "not-a-number",
"use_count": "not-a-number",
"patch_generation": 1,
"last_reused_patch_generation": 999,
}
})
skill_usage.bump_view("damaged")
skill_usage.bump_use("damaged")
skill_usage.bump_patch("damaged")
skill_usage.bump_use("damaged")
record = skill_usage.get_record("damaged")
assert record["view_count"] == 1
assert record["use_count"] == 2
assert record["patch_generation"] == 2
assert record["last_reused_patch_generation"] == 2
loaded = [event for event in events if event["action"] == "loaded"]
assert [event["reused"] for event in loaded] == [False, True]
assert [event["reuse_after_patch"] for event in loaded] == [False, True]
def test_bumps_do_not_corrupt_other_skills(skills_home):
from tools.skill_usage import bump_view, bump_use, get_record
bump_view("skill-a")
@@ -363,4 +553,3 @@ def test_adopt_rejects_empty_name(skills_home):
from tools.skill_usage import adopt_skill
assert adopt_skill("")[0] is False
tests/tools/test_skills_hub.py
+42 -1
View File
@@ -1148,12 +1148,14 @@ class TestInstallPathSafety:
)
with patch.object(hub, "SKILLS_DIR", skills_dir), \
patch.object(hub, "QUARANTINE_DIR", quarantine_root):
patch.object(hub, "QUARANTINE_DIR", quarantine_root), \
patch("tools.skill_usage.record_installed") as record_installed:
with pytest.raises(ValueError, match="symlink"):
hub.install_from_quarantine(
q_dir, "bad-skill", "", bundle, scan_result,
)
record_installed.assert_not_called()
assert not (skills_dir / "bad-skill" / "leak.txt").exists()
assert secret.read_text() == "data exfiltration payload\n"
@@ -1434,6 +1436,45 @@ class TestInstallPathSafety:
assert (skills_dir / "notes").read_text() == "not a directory"
def test_install_from_quarantine_records_successful_install(self, tmp_path):
import tools.skills_hub as hub
from tools.skills_guard import ScanResult
skills_dir = tmp_path / "skills"
quarantine_root = skills_dir / ".hub" / "quarantine"
q_dir = quarantine_root / "pending"
q_dir.mkdir(parents=True)
skill_md = "---\nname: good-skill\n---\n\n# Good skill\n"
(q_dir / "SKILL.md").write_text(skill_md, encoding="utf-8")
bundle = hub.SkillBundle(
name="good-skill",
files={"SKILL.md": skill_md},
source="community",
identifier="good/source",
trust_level="community",
)
scan_result = ScanResult(
skill_name="good-skill",
source="community",
trust_level="community",
verdict="safe",
)
with patch.object(hub, "SKILLS_DIR", skills_dir), \
patch.object(hub, "QUARANTINE_DIR", quarantine_root), \
patch("tools.skill_usage.record_installed") as record_installed:
installed = hub.install_from_quarantine(
q_dir,
"good-skill",
"",
bundle,
scan_result,
)
assert installed == skills_dir / "good-skill"
assert installed.is_dir()
record_installed.assert_called_once_with("good-skill")
# ---------------------------------------------------------------------------
# parallel_search_sources — overall_timeout must be honoured even when a
tests/tools/test_skills_tool.py
+23
View File
@@ -334,6 +334,29 @@ class TestSkillView:
assert by_name["success"] is True
assert "Step 1" in by_name["content"]
def test_registered_view_tracks_use_with_task_and_session(self, tmp_path):
from tools.skills_tool import _skill_view_with_bump
with (
patch("tools.skills_tool.SKILLS_DIR", tmp_path),
patch("tools.skill_usage.bump_view") as bump_view,
patch("tools.skill_usage.bump_use") as bump_use,
):
_make_skill(tmp_path, "my-skill")
raw = _skill_view_with_bump(
{"name": "my-skill"},
task_id="task-view",
session_id="session-view",
)
assert json.loads(raw)["success"] is True
bump_view.assert_called_once_with("my-skill")
bump_use.assert_called_once_with(
"my-skill",
task_id="task-view",
session_id="session-view",
)
def test_view_reference_files(self, tmp_path):
with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
tools/skill_manager_tool.py
+18 -5
View File
@@ -1521,6 +1521,8 @@ def skill_manage(
new_string: str = None,
replace_all: bool = False,
absorbed_into: str = None,
task_id: str = None,
session_id: str = None,
) -> str:
"""
Manage user-created skills. Dispatches to the appropriate action handler.
@@ -1592,13 +1594,22 @@ def skill_manage(
# user-directed, and those skills belong to the user (the curator must
# not touch them). Best-effort; telemetry failures never break the tool.
try:
from tools.skill_usage import bump_patch, forget, mark_agent_created
from tools.skill_usage import bump_patch, forget, record_created
from tools.skill_provenance import is_background_review
if action == "create":
if is_background_review():
mark_agent_created(name)
record_created(
name,
agent_created=is_background_review(),
task_id=task_id,
session_id=session_id,
)
elif action in {"patch", "edit", "write_file", "remove_file"}:
bump_patch(name)
bump_patch(
name,
action=action,
task_id=task_id,
session_id=session_id,
)
elif action == "delete":
# A recoverable curator archive (routed through archive_skill)
# keeps its usage record as STATE_ARCHIVED so `hermes curator
@@ -1763,6 +1774,8 @@ registry.register(
old_string=args.get("old_string"),
new_string=args.get("new_string"),
replace_all=args.get("replace_all", False),
absorbed_into=args.get("absorbed_into")),
absorbed_into=args.get("absorbed_into"),
task_id=kw.get("task_id"),
session_id=kw.get("session_id")),
emoji="📝",
)
tools/skill_usage.py
+213 -18
View File
@@ -649,6 +649,8 @@ def _empty_record() -> Dict[str, Any]:
"last_used_at": None,
"last_viewed_at": None,
"patch_count": 0,
"patch_generation": 0,
"last_reused_patch_generation": 0,
"last_patched_at": None,
"created_at": _now_iso(),
"state": STATE_ACTIVE,
@@ -677,8 +679,8 @@ def load_usage() -> Dict[str, Dict[str, Any]]:
return clean
def save_usage(data: Dict[str, Dict[str, Any]]) -> None:
"""Write the usage map atomically. Best-effort — errors are logged, not raised."""
def save_usage(data: Dict[str, Dict[str, Any]]) -> bool:
"""Write the usage map atomically and report whether it committed."""
path = _usage_file()
try:
path.parent.mkdir(parents=True, exist_ok=True)
@@ -691,6 +693,7 @@ def save_usage(data: Dict[str, Dict[str, Any]]) -> None:
f.flush()
os.fsync(f.fileno())
os.replace(tmp_path, path)
return True
except BaseException:
try:
os.unlink(tmp_path)
@@ -699,6 +702,7 @@ def save_usage(data: Dict[str, Dict[str, Any]]) -> None:
raise
except Exception as e:
logger.debug("Failed to write %s: %s", path, e, exc_info=True)
return False
def get_record(skill_name: str) -> Dict[str, Any]:
@@ -736,7 +740,7 @@ def seed_record_if_missing(skill_name: str) -> None:
logger.debug("skill_usage.seed_record_if_missing(%s) failed: %s", skill_name, e, exc_info=True)
def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False) -> None:
def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False) -> Any:
"""Load, apply *mutator(record)* in place, save. Best-effort.
By default this records telemetry for ANY skill — bundled, hub-installed,
@@ -748,20 +752,97 @@ def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False
hub-installed skill).
"""
if not skill_name:
return
return None
try:
if require_curation_eligible and not is_curation_eligible(skill_name):
return
return None
with _usage_file_lock():
data = load_usage()
rec = data.get(skill_name)
if not isinstance(rec, dict):
rec = _empty_record()
mutator(rec)
result = mutator(rec)
data[skill_name] = rec
save_usage(data)
if not save_usage(data):
return None
return result
except Exception as e:
logger.debug("skill_usage._mutate(%s) failed: %s", skill_name, e, exc_info=True)
return None
def _non_negative_int(value: Any) -> int:
if isinstance(value, bool):
return 0
try:
return max(0, int(value or 0))
except (TypeError, ValueError):
return 0
def telemetry_provenance(
skill_name: str,
record: Optional[Dict[str, Any]] = None,
) -> str:
"""Return the bounded provenance used by shared skill metrics."""
if is_hub_installed(skill_name) or is_bundled(skill_name):
return "installed"
if ":" in skill_name:
try:
from hermes_cli.plugins import get_plugin_manager
if get_plugin_manager().find_plugin_skill(skill_name) is not None:
return "installed"
except Exception:
pass
if isinstance(record, dict):
created_by = record.get("created_by")
if created_by == "installed":
return "installed"
if created_by == "agent":
return "agent_created"
if _find_external_skill_dir(skill_name) is not None:
return "external"
if _find_skill_dir(skill_name) is not None or isinstance(record, dict):
return "local"
return "unknown"
def _emit_skill_lifecycle(
skill_name: str,
action: str,
*,
record: Optional[Dict[str, Any]] = None,
task_id: Optional[str] = None,
session_id: Optional[str] = None,
use_count: Optional[int] = None,
reused: Optional[bool] = None,
reuse_after_patch: Optional[bool] = None,
) -> None:
"""Emit one best-effort lifecycle fact after authoritative state changes."""
try:
from hermes_cli.lifecycle import has_hook, invoke_hook
if not has_hook("on_skill_lifecycle"):
return
invoke_hook(
"on_skill_lifecycle",
action=action,
skill_name=skill_name,
provenance=telemetry_provenance(skill_name, record),
task_id=task_id or "",
session_id=session_id or "",
use_count=use_count,
reused=reused,
reuse_after_patch=reuse_after_patch,
)
except Exception:
logger.debug(
"skill_usage lifecycle hook failed for %s/%s",
skill_name,
action,
exc_info=True,
)
# ---------------------------------------------------------------------------
@@ -775,32 +856,127 @@ def bump_view(skill_name: str) -> None:
included. Usage telemetry is observability, not a curation signal.
"""
def _apply(rec: Dict[str, Any]) -> None:
rec["view_count"] = int(rec.get("view_count") or 0) + 1
rec["view_count"] = _non_negative_int(rec.get("view_count")) + 1
rec["last_viewed_at"] = _now_iso()
_mutate(skill_name, _apply)
def bump_use(skill_name: str) -> None:
def bump_use(
skill_name: str,
*,
task_id: Optional[str] = None,
session_id: Optional[str] = None,
) -> None:
"""Bump use_count and last_used_at. Called when a skill is actively used
(e.g. loaded into the prompt path or referenced from an assistant turn).
Tracks every skill regardless of provenance.
"""
def _apply(rec: Dict[str, Any]) -> None:
rec["use_count"] = int(rec.get("use_count") or 0) + 1
def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
previous_use_count = _non_negative_int(rec.get("use_count"))
patch_generation = _non_negative_int(rec.get("patch_generation"))
last_reused_generation = min(
_non_negative_int(rec.get("last_reused_patch_generation")),
patch_generation,
)
reused = previous_use_count > 0
reuse_after_patch = reused and patch_generation > last_reused_generation
rec["use_count"] = previous_use_count + 1
rec["last_used_at"] = _now_iso()
_mutate(skill_name, _apply)
rec["patch_generation"] = patch_generation
rec["last_reused_patch_generation"] = last_reused_generation
if reuse_after_patch:
rec["last_reused_patch_generation"] = patch_generation
return {
"created_by": rec.get("created_by"),
"use_count": rec["use_count"],
"reused": reused,
"reuse_after_patch": reuse_after_patch,
}
facts = _mutate(skill_name, _apply)
if isinstance(facts, dict):
_emit_skill_lifecycle(
skill_name,
"loaded",
record=facts,
task_id=task_id,
session_id=session_id,
use_count=facts["use_count"],
reused=facts["reused"],
reuse_after_patch=facts["reuse_after_patch"],
)
def bump_patch(skill_name: str) -> None:
def bump_patch(
skill_name: str,
*,
action: str = "patch",
task_id: Optional[str] = None,
session_id: Optional[str] = None,
) -> None:
"""Bump patch_count and last_patched_at. Called from skill_manage (patch/edit).
Tracks every skill regardless of provenance.
"""
def _apply(rec: Dict[str, Any]) -> None:
rec["patch_count"] = int(rec.get("patch_count") or 0) + 1
lifecycle_action = "patched" if action == "patch" else "edited"
def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
rec["patch_count"] = _non_negative_int(rec.get("patch_count")) + 1
rec["patch_generation"] = _non_negative_int(rec.get("patch_generation")) + 1
rec["last_patched_at"] = _now_iso()
_mutate(skill_name, _apply)
return {"created_by": rec.get("created_by")}
facts = _mutate(skill_name, _apply)
if isinstance(facts, dict):
_emit_skill_lifecycle(
skill_name,
lifecycle_action,
record=facts,
task_id=task_id,
session_id=session_id,
)
def record_created(
skill_name: str,
*,
agent_created: bool,
task_id: Optional[str] = None,
session_id: Optional[str] = None,
) -> None:
"""Persist explicit creation provenance and emit a successful create fact."""
def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
# A successful create is a new logical skill even if stale sidecar
# state survived an earlier deletion or manual filesystem change.
rec.clear()
rec.update(_empty_record())
if agent_created:
rec["created_by"] = "agent"
return {"created_by": rec["created_by"]}
facts = _mutate(skill_name, _apply)
if isinstance(facts, dict):
_emit_skill_lifecycle(
skill_name,
"created",
record=facts,
task_id=task_id,
session_id=session_id,
)
def record_installed(skill_name: str) -> None:
"""Record a successful Skills Hub install without exporting its name."""
def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
rec["created_by"] = "installed"
rec["state"] = STATE_ACTIVE
rec["archived_at"] = None
return {"created_by": rec["created_by"]}
facts = _mutate(skill_name, _apply)
if isinstance(facts, dict):
_emit_skill_lifecycle(skill_name, "installed", record=facts)
def mark_agent_created(skill_name: str) -> None:
@@ -820,13 +996,32 @@ def set_state(skill_name: str, state: str) -> None:
if state not in _VALID_STATES:
logger.debug("set_state: invalid state %r for %s", state, skill_name)
return
def _apply(rec: Dict[str, Any]) -> None:
def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
previous_state = rec.get("state")
if previous_state == state:
return {"changed": False, "created_by": rec.get("created_by")}
rec["state"] = state
if state == STATE_ARCHIVED:
rec["archived_at"] = _now_iso()
elif state == STATE_ACTIVE:
rec["archived_at"] = None
_mutate(skill_name, _apply, require_curation_eligible=True)
return {
"changed": True,
"created_by": rec.get("created_by"),
"previous_state": previous_state,
}
facts = _mutate(skill_name, _apply, require_curation_eligible=True)
if not isinstance(facts, dict) or not facts.get("changed"):
return
action = {
STATE_ARCHIVED: "archived",
STATE_STALE: "stale",
}.get(state)
if state == STATE_ACTIVE and facts.get("previous_state") == STATE_ARCHIVED:
action = "restored"
if action is not None:
_emit_skill_lifecycle(skill_name, action, record=facts)
def set_pinned(skill_name: str, pinned: bool) -> None:
tools/skills_hub.py
+11
View File
@@ -3835,6 +3835,17 @@ def install_from_quarantine(
content_hash(install_dir),
)
try:
from tools.skill_usage import record_installed
record_installed(safe_skill_name)
except Exception:
logger.debug(
"Unable to record skill install lifecycle for %s",
safe_skill_name,
exc_info=True,
)
return install_dir
tools/skills_tool.py
+5 -1
View File
@@ -1943,7 +1943,11 @@ def _skill_view_with_bump(args, **kw):
# 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))
bump_use(
str(resolved),
task_id=kw.get("task_id"),
session_id=kw.get("session_id"),
)
except Exception:
pass
return result