Compare commits
1 Commits
fix/agent-
...
fix/865
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
62925c294d |
15
.github/workflows/tests.yml
vendored
15
.github/workflows/tests.yml
vendored
@@ -47,6 +47,21 @@ jobs:
|
||||
OPENAI_API_KEY: ""
|
||||
NOUS_API_KEY: ""
|
||||
|
||||
lint-paths:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 5
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
|
||||
|
||||
- name: Install Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Check for hardcoded ~/.hermes paths
|
||||
run: python3 scripts/lint_hardcoded_paths.py
|
||||
|
||||
e2e:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
|
||||
@@ -1,135 +0,0 @@
|
||||
"""
|
||||
Agent Card — A2A-compliant agent discovery.
|
||||
Part of #843: fix: implement A2A agent card for fleet discovery (#819)
|
||||
|
||||
Provides metadata about the agent's identity, capabilities, and installed skills
|
||||
for discovery by other agents in the fleet.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from hermes_cli import __version__
|
||||
from hermes_cli.config import load_config, get_hermes_home
|
||||
from agent.skill_utils import (
|
||||
iter_skill_index_files,
|
||||
parse_frontmatter,
|
||||
get_all_skills_dirs,
|
||||
get_disabled_skill_names,
|
||||
skill_matches_platform
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@dataclass
|
||||
class AgentSkill:
|
||||
id: str
|
||||
name: str
|
||||
description: str = ""
|
||||
version: str = "1.0.0"
|
||||
|
||||
@dataclass
|
||||
class AgentCapabilities:
|
||||
streaming: bool = True
|
||||
tools: bool = True
|
||||
vision: bool = False
|
||||
reasoning: bool = False
|
||||
|
||||
@dataclass
|
||||
class AgentCard:
|
||||
name: str
|
||||
description: str
|
||||
url: str
|
||||
version: str = __version__
|
||||
capabilities: AgentCapabilities = field(default_factory=AgentCapabilities)
|
||||
skills: List[AgentSkill] = field(default_factory=list)
|
||||
defaultInputModes: List[str] = field(default_factory=lambda: ["text/plain"])
|
||||
defaultOutputModes: List[str] = field(default_factory=lambda: ["text/plain"])
|
||||
|
||||
def _load_skills() -> List[AgentSkill]:
|
||||
"""Scan all enabled skills and return metadata."""
|
||||
skills = []
|
||||
disabled = get_disabled_skill_names()
|
||||
|
||||
for skills_dir in get_all_skills_dirs():
|
||||
if not skills_dir.is_dir():
|
||||
continue
|
||||
for skill_file in iter_skill_index_files(skills_dir, "SKILL.md"):
|
||||
try:
|
||||
raw = skill_file.read_text(encoding="utf-8")
|
||||
frontmatter, _ = parse_frontmatter(raw)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
skill_name = frontmatter.get("name") or skill_file.parent.name
|
||||
if str(skill_name) in disabled:
|
||||
continue
|
||||
if not skill_matches_platform(frontmatter):
|
||||
continue
|
||||
|
||||
skills.append(AgentSkill(
|
||||
id=str(skill_name),
|
||||
name=str(frontmatter.get("name", skill_name)),
|
||||
description=str(frontmatter.get("description", "")),
|
||||
version=str(frontmatter.get("version", "1.0.0"))
|
||||
))
|
||||
return skills
|
||||
|
||||
def build_agent_card() -> AgentCard:
|
||||
"""Build the agent card from current configuration and environment."""
|
||||
config = load_config()
|
||||
|
||||
# Identity
|
||||
name = os.environ.get("HERMES_AGENT_NAME") or config.get("agent", {}).get("name") or "hermes"
|
||||
description = os.environ.get("HERMES_AGENT_DESCRIPTION") or config.get("agent", {}).get("description") or "Sovereign AI agent"
|
||||
|
||||
# URL - try to determine from environment or config
|
||||
port = os.environ.get("HERMES_WEB_PORT") or "9119"
|
||||
host = os.environ.get("HERMES_WEB_HOST") or "localhost"
|
||||
url = f"http://{host}:{port}"
|
||||
|
||||
# Capabilities
|
||||
# In a real scenario, we'd check model metadata for vision/reasoning
|
||||
capabilities = AgentCapabilities(
|
||||
streaming=True,
|
||||
tools=True,
|
||||
vision=False, # Default to false unless we can confirm
|
||||
reasoning=False
|
||||
)
|
||||
|
||||
# Skills
|
||||
skills = _load_skills()
|
||||
|
||||
return AgentCard(
|
||||
name=name,
|
||||
description=description,
|
||||
url=url,
|
||||
version=__version__,
|
||||
capabilities=capabilities,
|
||||
skills=skills
|
||||
)
|
||||
|
||||
def get_agent_card_json() -> str:
|
||||
"""Return the agent card as a JSON string."""
|
||||
try:
|
||||
card = build_agent_card()
|
||||
return json.dumps(asdict(card), indent=2)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to build agent card: {e}")
|
||||
# Minimal fallback card
|
||||
fallback = {
|
||||
"name": "hermes",
|
||||
"description": "Sovereign AI agent (fallback)",
|
||||
"version": __version__,
|
||||
"error": str(e)
|
||||
}
|
||||
return json.dumps(fallback, indent=2)
|
||||
|
||||
def validate_agent_card(card_data: Dict[str, Any]) -> bool:
|
||||
"""Check if the card data complies with the A2A schema."""
|
||||
required = ["name", "description", "url", "version"]
|
||||
return all(k in card_data for k in required)
|
||||
@@ -45,7 +45,6 @@ from hermes_cli.config import (
|
||||
redact_key,
|
||||
)
|
||||
from gateway.status import get_running_pid, read_runtime_status
|
||||
from agent.agent_card import get_agent_card_json
|
||||
|
||||
try:
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
@@ -97,9 +96,6 @@ _PUBLIC_API_PATHS: frozenset = frozenset({
|
||||
"/api/config/defaults",
|
||||
"/api/config/schema",
|
||||
"/api/model/info",
|
||||
"/api/agent-card",
|
||||
"/agent-card.json",
|
||||
"/.well-known/agent-card.json",
|
||||
})
|
||||
|
||||
|
||||
@@ -364,14 +360,6 @@ def _probe_gateway_health() -> tuple[bool, dict | None]:
|
||||
return False, None
|
||||
|
||||
|
||||
@app.get("/api/agent-card")
|
||||
@app.get("/agent-card.json")
|
||||
@app.get("/.well-known/agent-card.json")
|
||||
async def get_agent_card():
|
||||
"""Return the A2A agent card for fleet discovery."""
|
||||
return JSONResponse(content=json.loads(get_agent_card_json()))
|
||||
|
||||
|
||||
@app.get("/api/status")
|
||||
async def get_status():
|
||||
current_ver, latest_ver = check_config_version()
|
||||
|
||||
146
scripts/lint_hardcoded_paths.py
Normal file
146
scripts/lint_hardcoded_paths.py
Normal file
@@ -0,0 +1,146 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Lint for hardcoded ~/.hermes paths.
|
||||
|
||||
Detects patterns that break profile isolation by hardcoding ~/.hermes
|
||||
instead of using get_hermes_home() from hermes_constants.
|
||||
|
||||
Usage:
|
||||
python3 scripts/lint_hardcoded_paths.py # check all
|
||||
python3 scripts/lint_hardcoded_paths.py --fix # suggest fixes
|
||||
python3 scripts/lint_hardcoded_paths.py --json # JSON output
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass, asdict
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
# Patterns that indicate hardcoded ~/.hermes paths
|
||||
_PATTERNS = [
|
||||
(r'Path\.home\(\)\s*/\s*[\"\']\.hermes[\"\']', "Path.home() / '.hermes'"),
|
||||
(r'Path\.home\(\)\s*/\s*\"\.hermes\"', 'Path.home() / ".hermes"'),
|
||||
(r'[\"\']~[/\\]\.hermes[/\\]', "hardcoded ~/.hermes string"),
|
||||
(r'os\.path\.expanduser\([\"\']~[/\\]\.hermes', "expanduser('~/.hermes')"),
|
||||
(r'os\.path\.join\(.*expanduser.*\.hermes', "os.path.join with expanduser"),
|
||||
(r'HOME[\"\']\s*\+\s*[\"\'][/\\]\.hermes', "$HOME + .hermes concatenation"),
|
||||
]
|
||||
|
||||
# Files to skip
|
||||
_SKIP_DIRS = {
|
||||
".git", "__pycache__", ".venv", "venv", "node_modules",
|
||||
".mypy_cache", ".pytest_cache", "dist", "build",
|
||||
}
|
||||
_SKIP_FILES = {
|
||||
"hermes_constants.py", # source of truth
|
||||
}
|
||||
_SKIP_EXTENSIONS = {".md", ".rst", ".txt", ".json", ".yaml", ".yml", ".toml"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Finding:
|
||||
file: str
|
||||
line: int
|
||||
pattern: str
|
||||
content: str
|
||||
severity: str = "error"
|
||||
|
||||
|
||||
def scan_file(filepath: Path) -> List[Finding]:
|
||||
"""Scan a single file for hardcoded path patterns."""
|
||||
findings = []
|
||||
|
||||
try:
|
||||
content = filepath.read_text(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
return findings
|
||||
|
||||
for line_num, line in enumerate(content.split("\n"), 1):
|
||||
# Skip comments and docstrings (rough heuristic)
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("#") or stripped.startswith('"""') or stripped.startswith("'''"):
|
||||
continue
|
||||
|
||||
for pattern, description in _PATTERNS:
|
||||
if re.search(pattern, line):
|
||||
findings.append(Finding(
|
||||
file=str(filepath.relative_to(REPO_ROOT)),
|
||||
line=line_num,
|
||||
pattern=description,
|
||||
content=stripped[:120],
|
||||
))
|
||||
break # One finding per line
|
||||
|
||||
return findings
|
||||
|
||||
|
||||
def scan_repo(root: Path = None) -> List[Finding]:
|
||||
"""Scan the entire repo for hardcoded paths."""
|
||||
root = root or REPO_ROOT
|
||||
findings = []
|
||||
|
||||
for path in root.rglob("*.py"):
|
||||
# Skip directories
|
||||
rel = path.relative_to(root)
|
||||
parts = rel.parts
|
||||
if any(p in _SKIP_DIRS for p in parts):
|
||||
continue
|
||||
if path.name in _SKIP_FILES:
|
||||
continue
|
||||
if path.suffix in _SKIP_EXTENSIONS:
|
||||
continue
|
||||
|
||||
findings.extend(scan_file(path))
|
||||
|
||||
return findings
|
||||
|
||||
|
||||
def format_findings(findings: List[Finding]) -> str:
|
||||
"""Format findings as readable report."""
|
||||
if not findings:
|
||||
return "OK: No hardcoded ~/.hermes paths found."
|
||||
|
||||
lines = [
|
||||
f"FAIL: Found {len(findings)} hardcoded ~/.hermes path(s):",
|
||||
"",
|
||||
]
|
||||
for f in findings:
|
||||
lines.append(f" {f.file}:{f.line} [{f.severity}]")
|
||||
lines.append(f" Pattern: {f.pattern}")
|
||||
lines.append(f" Line: {f.content}")
|
||||
lines.append("")
|
||||
|
||||
lines.append("Fix: Use get_hermes_home() from hermes_constants instead.")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser(description="Lint for hardcoded ~/.hermes paths")
|
||||
parser.add_argument("--json", action="store_true", help="JSON output")
|
||||
parser.add_argument("--fix", action="store_true", help="Show fix suggestions")
|
||||
args = parser.parse_args()
|
||||
|
||||
findings = scan_repo()
|
||||
|
||||
if args.json:
|
||||
print(json.dumps([asdict(f) for f in findings], indent=2))
|
||||
elif args.fix and findings:
|
||||
print(format_findings(findings))
|
||||
print("\nSuggested fix pattern:")
|
||||
print(" from hermes_constants import get_hermes_home")
|
||||
print(" hermes_home = get_hermes_home()")
|
||||
else:
|
||||
print(format_findings(findings))
|
||||
|
||||
return 1 if findings else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user