Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,13 @@
_CONTEXT_KEY_FIELDS = ("session_id", "task_id", "run_id")
_HERMES_SESSION_ENV = "HERMES_SESSION_ID"
_UNSUPPORTED_STATUS = "unsupported"
# Managed skill roots that skill-ledger daemon tracks. When a skill in
# ~/.hermes/skills/ is reported as "unmanaged", we retry the query against
# these roots to pick up scan/certify results recorded there.
_MANAGED_SKILL_ROOTS = (
Path("/usr/share/anolisa/skills"),
Path("/usr/local/share/anolisa/skills"),
)
_UNSUPPORTED_HERMES_NOTICE = "暂不支持Hermes场景,请自行关注skill安全性。"
_SKILLFS_INPLACE_SENTINELS = (
Path(".skillfs-inbox"),
Expand Down Expand Up @@ -144,9 +151,31 @@ def _on_pre_tool_call(self, tool_name, args, **kwargs):
)
return None

# When the skill is unmanaged (not in managedSkillDirs), the CLI
# returns latestStatus="unmanaged" with message=null. But the skill
# may have been scanned/certified in a managed root (e.g.
# /usr/share/anolisa/skills/) and copied to ~/.hermes/skills/.
# Retry the query against managed roots before giving up.
latest_status = str(summary.get("latestStatus", ""))
if latest_status == "unmanaged":
managed_summary = self._query_managed_root(skill_dir)
if managed_summary is not None:
summary = managed_summary

message = summary.get("message")
if not isinstance(message, str) or not message.strip():
return None
# If still no message after managed-root retry, generate a
# synthetic warning for truly unmanaged skills so the user is
# alerted that the skill has no ledger attestation.
latest_status = str(summary.get("latestStatus", ""))
if latest_status == "unmanaged":
skill_name = str(summary.get("skillName") or skill_dir.name)
message = (
f"Skill '{skill_name}': unmanaged — not tracked by "
f"Skill Ledger. Security status unknown."
)
if not isinstance(message, str) or not message.strip():
return None

status = str(summary.get("latestStatus", "unknown"))
skill_name = str(summary.get("skillName") or skill_dir.name)
Expand All @@ -163,6 +192,38 @@ def _on_pre_tool_call(self, tool_name, args, **kwargs):
self._remember_warning(kwargs, skill_name, skill_dir, status, message)
return None

def _query_managed_root(self, skill_dir: Path) -> dict[str, Any] | None:
"""Retry skill-ledger show against managed skill roots.

When a skill in ~/.hermes/skills/ is unmanaged, it may have a
scanned/certified copy in a managed root. Try each managed root
and return the first non-unmanaged summary.
"""
skill_name = skill_dir.name
for managed_root in _MANAGED_SKILL_ROOTS:
managed_path = managed_root / skill_name
try:
if not managed_path.is_dir():
continue
except OSError:
continue
result = call_agent_sec_cli(
["skill-ledger", "show", str(managed_path)],
timeout=self._timeout,
trace_context=trace_context({}),
)
if not result.stdout.strip():
continue
try:
summary = json.loads(result.stdout)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(summary, dict):
continue
if str(summary.get("latestStatus", "")) != "unmanaged":
return summary
return None

def _on_transform_llm_output(
self,
response_text: str = "",
Expand Down
Loading