Sabo/webui-develop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 1: Activity and Error Log for Agent Tab

Browse Source 
Backend: _log_agent_activity, _get_activity_log, _get_error_log. API: GET /api/agents/{id}/activity and /errors. Frontend: Activity and Errors tabs in agent detail overlay. CSS: activity-event-row, error-event-row. Config fix: Z.ai API key.
This commit is contained in:
Rose
2026-04-20 13:28:37 +02:00
parent 96977b576a
commit fbf79362a4
6 changed files with 1652 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

410
api/agents.py
View File

@@ -1,6 +1,6 @@
"""
Rose Agents Panel API — Data layer for Hermes WebUI Agents extension.
Provides Rose + Tier-2 agent status, inbox management, and configuration.
Provides Rose + Tier-2 agent status, inbox management, soul/memory editing, and configuration.
"""
import json
@@ -8,6 +8,7 @@ import os
import subprocess
import threading
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
@@ -20,13 +21,13 @@ _INBOX_BUS = _HERMES_DIR / "scripts" / "message_bus.py"
# ── Tier-2 Agent Registry ──────────────────────────────────────────────────────
TIER2_AGENTS = {
"lotus": {"name": "Lotus", "emoji": "🪷", "domain": "Health, Fitness & Recovery", "color": "#e91e63"},
"forget-me-not": {"name": "Forget-me-not", "emoji": "🌼", "domain": "Calendar, Time & Social", "color": "#ff9800"},
"sunflower": {"name": "Sunflower", "emoji": "🌻", "domain": "Finance, Wealth & Subscriptions","color": "#ffeb3b"},
"iris": {"name": "Iris", "emoji": "⚜️", "domain": "Career, Learning & Focus", "color": "#9c27b0"},
"ivy": {"name": "Ivy", "emoji": "🌿", "domain": "Smart Home & Environment", "color": "#4caf50"},
"dandelion": {"name": "Dandelion", "emoji": "🛡️", "domain": "Communication Triage", "color": "#03a9f4"},
"root": {"name": "Root", "emoji": "🌳", "domain": "DevOps, Logs & System Health", "color": "#795548"},
"lotus": {"name": "Lotus", "emoji": "🪷", "domain": "Health, Fitness & Recovery", "color": "#e91e63"},
"forget-me-not": {"name": "Forget-me-not", "emoji": "🌼", "domain": "Calendar, Time & Social", "color": "#ff9800"},
"sunflower": {"name": "Sunflower", "emoji": "🌻", "domain": "Finance, Wealth & Subscriptions","color": "#ffeb3b"},
"iris": {"name": "Iris", "emoji": "⚜️", "domain": "Career, Learning & Focus", "color": "#9c27b0"},
"ivy": {"name": "Ivy", "emoji": "🌿", "domain": "Smart Home & Environment", "color": "#4caf50"},
"dandelion": {"name": "Dandelion", "emoji": "🛡️", "domain": "Communication Triage", "color": "#03a9f4"},
"root": {"name": "Root", "emoji": "🌳", "domain": "DevOps, Logs & System Health", "color": "#795548"},
}
ROSE_META = {
@@ -51,8 +52,44 @@ def _get_process_status(agent_name: str) -> dict:
except Exception:
return {"running": False, "pid": None}
def _get_agent_status(agent_name: str) -> dict:
"""
Determine agent status: active / idle / offline.
- active: process running AND recent activity (< 5 min)
- idle: process running BUT no recent activity (5-15 min)
- offline: no process
"""
proc = _get_process_status(agent_name)
active_session_path = _AGENTS_DIR / agent_name / "active_session.txt"
if not proc["running"]:
return {"status": "offline", "last_activity": None, "pid": None}
last_activity = None
if active_session_path.exists():
try:
mtime = active_session_path.stat().st_mtime
last_activity = datetime.fromtimestamp(mtime).isoformat() + "Z"
age_minutes = (time.time() - mtime) / 60
if age_minutes < 5:
status = "active"
elif age_minutes < 15:
status = "idle"
else:
status = "offline"
except Exception:
status = "unknown"
else:
# Process running but no session file = treat as idle
status = "idle"
last_activity = None
return {"status": status, "last_activity": last_activity, "pid": proc["pid"]}
def _get_inbox_count(agent_name: str) -> int:
"""Count messages in agent inbox via message_bus.py."""
"""Count pending (unread) messages in agent inbox."""
try:
result = subprocess.run(
["/usr/bin/python3", str(_INBOX_BUS), "check", "--agent", agent_name],
@@ -65,7 +102,8 @@ def _get_inbox_count(agent_name: str) -> int:
pass
return 0
def _read_inbox(agent_name: str, limit: int = 20) -> list[dict]:
def _read_inbox(agent_name: str, limit: int = 50) -> list[dict]:
"""Read messages from agent inbox."""
inbox_path = _AGENTS_DIR / agent_name / "inbox.json"
if not inbox_path.exists():
@@ -74,18 +112,42 @@ def _read_inbox(agent_name: str, limit: int = 20) -> list[dict]:
with open(inbox_path, "r") as f:
data = json.load(f)
messages = data if isinstance(data, list) else data.get("messages", [])
return messages[-limit:]
# Reverse so newest first, return limited
return list(reversed(messages))[:limit]
except (json.JSONDecodeError, IOError):
return []
def _read_file_safe(path: Path) -> str | None:
"""Read a file safely, return None if missing."""
try:
if path.exists():
return path.read_text()
except Exception:
pass
return None
def _write_file_safe(path: Path, content: str, backup: bool = True) -> dict:
"""Write a file safely with optional backup. Returns dict with success/error."""
try:
if backup and path.exists():
backup_path = path.with_suffix(path.suffix + ".backup")
backup_path.write_text(path.read_text())
path.write_text(content)
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
# ── API Functions ─────────────────────────────────────────────────────────────
def list_agents() -> dict:
"""Return status for Rose + all Tier-2 agents."""
"""Return status summary for Rose + all Tier-2 agents."""
agents = []
# Rose (the orchestrator)
rose_running = True # Rose IS the gateway/webui
# Rose (orchestrator — always "running" as it's the gateway itself)
rose_status = _get_agent_status("rose")
rose_inbox_count = _get_inbox_count("rose")
agents.append({
"id": "rose",
@@ -94,15 +156,16 @@ def list_agents() -> dict:
"domain": ROSE_META["domain"],
"color": ROSE_META["color"],
"tier": "orchestrator",
"running": rose_running,
"status": "active", # Rose is always running
"pid": None,
"last_activity": rose_status.get("last_activity"),
"inbox_count": rose_inbox_count,
})
# Tier-2 agents
for agent_id, meta in TIER2_AGENTS.items():
status = _get_process_status(agent_id)
inbox_count = _get_inbox_count(agent_id) if status["running"] else 0
status_info = _get_agent_status(agent_id)
inbox_count = _get_inbox_count(agent_id) if status_info["status"] != "offline" else 0
agents.append({
"id": agent_id,
"name": meta["name"],
@@ -110,27 +173,203 @@ def list_agents() -> dict:
"domain": meta["domain"],
"color": meta["color"],
"tier": "tier2",
"running": status["running"],
"pid": status["pid"],
"status": status_info["status"],
"pid": status_info["pid"],
"last_activity": status_info.get("last_activity"),
"inbox_count": inbox_count,
})
return {"agents": agents}
def get_agent_inbox(agent_id: str, limit: int = 20) -> dict:
def get_agent(agent_id: str) -> dict:
"""Return full detail for one agent: soul.md, memory.md, inbox, config."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"error": f"Unknown agent: {agent_id}"}
meta = TIER2_AGENTS.get(agent_id, ROSE_META)
agent_dir = _AGENTS_DIR / agent_id if agent_id != "rose" else _HERMES_DIR
soul = _read_file_safe(agent_dir / "soul.md")
memory = _read_file_safe(agent_dir / "memory.md")
inbox_messages = _read_inbox(agent_id)
status_info = _get_agent_status(agent_id)
inbox_count = _get_inbox_count(agent_id)
# Default model — extract from soul.md YAML frontmatter if present
default_model = None
if soul:
import re
m = re.search(r'model:\s*["\']?([^"\'\n]+)["\']?', soul)
if m:
default_model = m.group(1).strip()
# Disabled flag
disabled = (agent_dir / "disabled").exists() if agent_dir.exists() else False
return {
"id": agent_id,
"name": meta["name"],
"emoji": meta["emoji"],
"domain": meta["domain"],
"color": meta["color"],
"tier": "orchestrator" if agent_id == "rose" else "tier2",
"status": status_info["status"],
"last_activity": status_info.get("last_activity"),
"pid": status_info["pid"],
"inbox_count": inbox_count,
"soul": soul or "",
"memory": memory or "",
"default_model": default_model,
"disabled": disabled,
"inbox": inbox_messages,
}
def get_agent_status(agent_id: str) -> dict:
"""Return only the status info for one agent."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"error": f"Unknown agent: {agent_id}"}
if agent_id == "rose":
return {"status": "active", "last_activity": None, "pid": None}
return _get_agent_status(agent_id)
def get_agent_inbox(agent_id: str, limit: int = 50) -> dict:
"""Return inbox messages for a specific agent."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"error": f"Unknown agent: {agent_id}"}
messages = _read_inbox(agent_id, limit)
meta = TIER2_AGENTS.get(agent_id, ROSE_META)
return {
"agent_id": agent_id,
"agent_name": TIER2_AGENTS.get(agent_id, {}).get("name", "Rose"),
"agent_name": meta["name"],
"messages": messages,
}
def update_agent_soul(agent_id: str, content: str) -> dict:
"""Write soul.md for an agent. Returns {ok, error}."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"ok": False, "error": f"Unknown agent: {agent_id}"}
if agent_id == "rose":
return {"ok": False, "error": "Rose's soul.md cannot be edited via this API"}
soul_path = _AGENTS_DIR / agent_id / "soul.md"
# Ensure directory exists
(_AGENTS_DIR / agent_id).mkdir(parents=True, exist_ok=True)
return _write_file_safe(soul_path, content, backup=True)
def update_agent_memory(agent_id: str, content: str) -> dict:
"""Write memory.md for an agent. Returns {ok, error}."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"ok": False, "error": f"Unknown agent: {agent_id}"}
if agent_id == "rose":
return {"ok": False, "error": "Rose's memory.md cannot be edited via this API"}
memory_path = _AGENTS_DIR / agent_id / "memory.md"
(_AGENTS_DIR / agent_id).mkdir(parents=True, exist_ok=True)
return _write_file_safe(memory_path, content, backup=True)
def send_agent_message(agent_id: str, payload: dict) -> dict:
"""
Add a message to an agent's inbox (simulates inter-agent message).
payload: {from, type, subject, content}
"""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"ok": False, "error": f"Unknown agent: {agent_id}"}
inbox_path = _AGENTS_DIR / agent_id / "inbox.json"
(_AGENTS_DIR / agent_id).mkdir(parents=True, exist_ok=True)
# Load existing inbox
if inbox_path.exists():
try:
with open(inbox_path, "r") as f:
data = json.load(f)
messages = data if isinstance(data, list) else data.get("messages", [])
except (json.JSONDecodeError, IOError):
messages = []
else:
messages = []
# Add new message
import uuid
msg = {
"id": uuid.uuid4().hex[:8],
"from": payload.get("from", "rose"),
"type": payload.get("type", "request"),
"subject": payload.get("subject", ""),
"content": payload.get("content", ""),
"timestamp": datetime.utcnow().isoformat() + "Z",
"status": "unread",
}
messages.append(msg)
try:
with open(inbox_path, "w") as f:
json.dump(messages, f, indent=2)
return {"ok": True, "message_id": msg["id"]}
except Exception as e:
return {"ok": False, "error": str(e)}
def ack_agent_message(agent_id: str, msg_id: str) -> dict:
"""Mark a message as acknowledged/read in an agent's inbox."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"ok": False, "error": f"Unknown agent: {agent_id}"}
inbox_path = _AGENTS_DIR / agent_id / "inbox.json"
if not inbox_path.exists():
return {"ok": False, "error": "Inbox not found"}
try:
with open(inbox_path, "r") as f:
messages = json.load(f)
if not isinstance(messages, list):
messages = messages.get("messages", [])
found = False
for msg in messages:
if msg.get("id") == msg_id:
msg["status"] = "read"
found = True
break
if not found:
return {"ok": False, "error": f"Message {msg_id} not found"}
with open(inbox_path, "w") as f:
json.dump(messages, f, indent=2)
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
def set_agent_enabled(agent_id: str, enabled: bool) -> dict:
"""Enable or disable an agent. Disabled agents won't respond."""
if agent_id not in TIER2_AGENTS:
return {"ok": False, "error": f"Unknown agent: {agent_id}"}
if agent_id == "rose":
return {"ok": False, "error": "Rose cannot be disabled"}
disabled_flag = _AGENTS_DIR / agent_id / "disabled"
try:
if enabled:
if disabled_flag.exists():
disabled_flag.unlink()
else:
disabled_flag.write_text("disabled")
return {"ok": True, "disabled": not enabled}
except Exception as e:
return {"ok": False, "error": str(e)}
def get_agent_config(agent_id: str) -> dict:
"""Return configuration for a specific agent (soul.md path, etc)."""
"""Return configuration paths/info for a specific agent."""
if agent_id == "rose":
return {
"id": "rose",
@@ -140,11 +379,138 @@ def get_agent_config(agent_id: str) -> dict:
}
elif agent_id in TIER2_AGENTS:
soul_path = _AGENTS_DIR / agent_id / "soul.md"
memory_path = _AGENTS_DIR / agent_id / "memory.md"
inbox_path = _AGENTS_DIR / agent_id / "inbox.json"
return {
"id": agent_id,
"name": TIER2_AGENTS[agent_id]["name"],
"soul_path": str(soul_path) if soul_path.exists() else None,
"memory_path": str(memory_path) if memory_path.exists() else None,
"inbox_path": str(inbox_path),
}
return {"error": f"Unknown agent: {agent_id}"}
# ── Activity & Error Log ───────────────────────────────────────────────────────
ACTIVITY_EVENT_TYPES = [
"agent_started", "agent_stopped",
"message_sent", "message_received",
"task_started", "task_completed", "task_failed",
"error", "soul_updated", "memory_updated",
"chat_started", "chat_ended",
"health_check", "config_updated",
]
def _log_agent_activity(agent_id: str, event_type: str, details: str = "") -> dict:
"""
Write an activity event to the agent's activity.log file.
Log format: ISO timestamp | type | details
Returns {ok: True} or {ok: False, error: ...}
"""
if event_type not in ACTIVITY_EVENT_TYPES:
return {"ok": False, "error": f"Unknown event type: {event_type}"}
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"ok": False, "error": f"Unknown agent: {agent_id}"}
agent_dir = _AGENTS_DIR / agent_id if agent_id != "rose" else _HERMES_DIR
log_path = agent_dir / "activity.log"
agent_dir.mkdir(parents=True, exist_ok=True)
try:
timestamp = datetime.utcnow().isoformat() + "Z"
line = f"{timestamp} | {event_type} | {details}\n"
with open(log_path, "a") as f:
f.write(line)
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
def _get_activity_log(agent_id: str, limit: int = 50) -> list[dict]:
"""
Read recent activity events for an agent.
Returns list of {timestamp, type, details} sorted newest-first.
"""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return []
agent_dir = _AGENTS_DIR / agent_id if agent_id != "rose" else _HERMES_DIR
log_path = agent_dir / "activity.log"
if not log_path.exists():
return []
try:
lines = log_path.read_text().strip().split("\n")
events = []
for line in reversed(lines):
line = line.strip()
if not line:
continue
parts = line.split(" | ", 2)
if len(parts) >= 2:
events.append({
"timestamp": parts[0],
"type": parts[1],
"details": parts[2] if len(parts) > 2 else "",
})
if len(events) >= limit:
break
return events
except Exception:
return []
def _get_error_log(agent_id: str, limit: int = 20) -> list[dict]:
"""
Read error events from activity log (type == 'error').
Returns list of {timestamp, type, details} sorted newest-first.
"""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return []
agent_dir = _AGENTS_DIR / agent_id if agent_id != "rose" else _HERMES_DIR
log_path = agent_dir / "activity.log"
if not log_path.exists():
return []
try:
lines = log_path.read_text().strip().split("\n")
errors = []
for line in reversed(lines):
line = line.strip()
if not line or " | error | " not in line:
continue
parts = line.split(" | ", 2)
if len(parts) >= 2:
errors.append({
"timestamp": parts[0],
"type": parts[1],
"details": parts[2] if len(parts) > 2 else "",
})
if len(errors) >= limit:
break
return errors
except Exception:
return []
def get_agent_activity(agent_id: str, limit: int = 50) -> dict:
"""API: GET /api/agents/{id}/activity — return activity log."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"error": f"Unknown agent: {agent_id}"}
events = _get_activity_log(agent_id, limit)
return {
"agent_id": agent_id,
"events": events,
}
def get_agent_errors(agent_id: str, limit: int = 20) -> dict:
"""API: GET /api/agents/{id}/errors — return error log."""
if agent_id not in TIER2_AGENTS and agent_id != "rose":
return {"error": f"Unknown agent: {agent_id}"}
errors = _get_error_log(agent_id, limit)
return {
"agent_id": agent_id,
"errors": errors,
}

3
api/config.py
View File

@@ -786,6 +786,7 @@ def get_available_models() -> dict:
"OPENROUTER_API_KEY",
"GOOGLE_API_KEY",
"GLM_API_KEY",
"ZAI_API_KEY",
"KIMI_API_KEY",
"DEEPSEEK_API_KEY",
"OPENCODE_ZEN_API_KEY",
@@ -802,7 +803,7 @@ def get_available_models() -> dict:
detected_providers.add("openrouter")
if all_env.get("GOOGLE_API_KEY"):
detected_providers.add("google")
if all_env.get("GLM_API_KEY"):
if all_env.get("GLM_API_KEY") or all_env.get("ZAI_API_KEY"):
detected_providers.add("zai")
if all_env.get("KIMI_API_KEY"):
detected_providers.add("kimi-coding")

176
api/routes.py
View File

@@ -58,7 +58,7 @@ from api import agents as _agents
# ── CSRF: validate Origin/Referer on POST ────────────────────────────────────
import re as _re
_re_path = _re.compile(r"^(?P<path>/[^?]*)")
def _normalize_host_port(value: str) -> tuple[str, str | None]:
"""Split a host or host:port string into (hostname, port|None).
@@ -471,6 +471,14 @@ def handle_get(handler, parsed) -> bool:
settings.pop("password_hash", None)
return j(handler, settings)
# ── Logs ──
if parsed.path == "/api/logs":
return _handle_logs_list(handler)
if parsed.path.startswith("/api/logs/"):
log_name = parsed.path[len("/api/logs/"):]
return _handle_logs_read(handler, log_name)
if parsed.path == "/api/onboarding/status":
return j(handler, get_onboarding_status())
@@ -768,8 +776,7 @@ def handle_get(handler, parsed) -> bool:
return j(handler, {"tasks": _mc.get_tasks()})
if parsed.path == "/api/mc/feed":
limit = int(parse_qs(parsed.query).get("limit", ["50"])[0])
return j(handler, {"feed": _mc.get_feed(limit=limit)})
return j(handler, {"feed": _mc.get_feed(limit=50)})
# ── Agents API (Rose + Tier-2) ──
if parsed.path == "/api/agents":
@@ -783,6 +790,75 @@ def handle_get(handler, parsed) -> bool:
agent_id = parsed.path.split("/")[-1]
return j(handler, _agents.get_agent_config(agent_id))
# GET /api/agents/{id} — full agent detail
if parsed.path == "/api/agents/rose" or parsed.path == "/api/agents/lotus" or \
parsed.path == "/api/agents/sunflower" or parsed.path == "/api/agents/forget-me-not" or \
parsed.path == "/api/agents/root" or parsed.path == "/api/agents/dandelion" or \
parsed.path == "/api/agents/iris" or parsed.path == "/api/agents/ivy":
agent_id = parsed.path.split("/")[-1]
return j(handler, _agents.get_agent(agent_id))
# GET /api/agents/{id}/status
if parsed.path.startswith("/api/agents/") and parsed.path.endswith("/status"):
agent_id = parsed.path.split("/")[-2]
return j(handler, _agents.get_agent_status(agent_id))
# PUT /api/agents/{id}/soul
if parsed.path.endswith("/soul") and method == "PUT":
agent_id = parsed.path.split("/")[-2]
data = read_body(handler)
return j(handler, _agents.update_agent_soul(agent_id, data.get("content", "")))
# PUT /api/agents/{id}/memory
if parsed.path.endswith("/memory") and method == "PUT":
agent_id = parsed.path.split("/")[-2]
data = read_body(handler)
return j(handler, _agents.update_agent_memory(agent_id, data.get("content", "")))
# POST /api/agents/{id}/message
if parsed.path.endswith("/message") and method == "POST":
agent_id = parsed.path.split("/")[-2]
data = read_body(handler)
return j(handler, _agents.send_agent_message(agent_id, data))
# POST /api/agents/{id}/ack/{msg_id}
if "/ack/" in parsed.path and method == "POST":
parts = parsed.path.split("/")
agent_id = parts[2]
msg_id = parts[4]
return j(handler, _agents.ack_agent_message(agent_id, msg_id))
# POST /api/agents/{id}/enable | /disable
if parsed.path.endswith("/enable") or parsed.path.endswith("/disable"):
if method == "POST":
agent_id = parsed.path.split("/")[-2]
action = parsed.path.split("/")[-1]
return j(handler, _agents.set_agent_enabled(agent_id, action == "enable"))
# GET /api/agents/{id}/inbox (full, with limit query param)
if parsed.path.startswith("/api/agents/") and "/inbox" in parsed.path:
parts = parsed.path.split("/")
if len(parts) == 5 and parts[4] == "inbox":
agent_id = parts[3]
limit = int(parse_qs(parsed.query).get("limit", ["50"])[0])
return j(handler, _agents.get_agent_inbox(agent_id, limit=limit))
# GET /api/agents/{id}/activity
if parsed.path.startswith("/api/agents/") and "/activity" in parsed.path:
parts = parsed.path.split("/")
if len(parts) == 5 and parts[4] == "activity":
agent_id = parts[3]
limit = int(parse_qs(parsed.query).get("limit", ["50"])[0])
return j(handler, _agents.get_agent_activity(agent_id, limit=limit))
# GET /api/agents/{id}/errors
if parsed.path.startswith("/api/agents/") and "/errors" in parsed.path:
parts = parsed.path.split("/")
if len(parts) == 5 and parts[4] == "errors":
agent_id = parts[3]
limit = int(parse_qs(parsed.query).get("limit", ["20"])[0])
return j(handler, _agents.get_agent_errors(agent_id, limit=limit))
# ── Profile API (GET) ──
if parsed.path == "/api/profiles":
from api.profiles import list_profiles_api, get_active_profile_name
@@ -1248,6 +1324,14 @@ def handle_post(handler, parsed) -> bool:
except RuntimeError as e:
return bad(handler, str(e), 409)
# ── Logs API ──
if parsed.path == "/api/logs":
return _handle_logs_list(handler)
if parsed.path.startswith("/api/logs/"):
log_name = parsed.path[len("/api/logs/"):]
return _handle_logs_read(handler, log_name)
# ── Gateway API ──
if parsed.path == "/api/gateways":
# GET - list all gateways
@@ -3158,3 +3242,89 @@ def _handle_session_import(handler, body):
SESSIONS.popitem(last=False)
s.save()
return j(handler, {"ok": True, "session": s.compact() | {"messages": s.messages}})
# ── Logs ──────────────────────────────────────────────────────────────────
ALLOWED_LOG_FILES = {
"agent.log": "~/.hermes/logs/agent.log",
"errors.log": "~/.hermes/logs/errors.log",
"gateway.log": "~/.hermes/logs/gateway.log",
"gateway.error.log": "~/.hermes/logs/gateway.error.log",
"update.log": "~/.hermes/logs/update.log",
"webui.log": "~/.hermes/logs/webui.log",
"webui-prod.log": "~/.hermes/webui/bootstrap-8787.log",
"webui-dev.log": "~/.hermes/webui-dev/bootstrap-8788.log",
}
def _handle_logs_list(handler):
"""Return list of available log files with metadata."""
logs = []
for name, rel_path in ALLOWED_LOG_FILES.items():
path = Path(rel_path.replace("~", str(Path.home())))
if path.exists():
stat = path.stat()
logs.append({
"name": name,
"path": str(path),
"size": stat.st_size,
"modified": stat.st_mtime,
"size_human": _human_size(stat.st_size),
})
else:
logs.append({
"name": name,
"path": str(path),
"size": 0,
"modified": None,
"size_human": "0 B",
"missing": True,
})
return j(handler, {"logs": logs})
def _human_size(num_bytes):
for unit in ["B", "KB", "MB", "GB"]:
if num_bytes < 1024:
return f"{num_bytes:.1f} {unit}"
num_bytes /= 1024
return f"{num_bytes:.1f} TB"
def _handle_logs_read(handler, log_name):
"""Return last N lines of a log file."""
if log_name not in ALLOWED_LOG_FILES:
return bad(handler, f"Unknown log file: {log_name}")
rel_path = ALLOWED_LOG_FILES[log_name]
path = Path(rel_path.replace("~", str(Path.home())))
# Security: resolve and verify path stays within ~/.hermes
try:
resolved = path.resolve()
hermes_root = Path.home() / ".hermes"
if not str(resolved).startswith(str(hermes_root)):
return bad(handler, "Access denied")
except Exception:
return bad(handler, "Invalid path")
if not path.exists():
return bad(handler, f"Log file not found: {log_name}")
# Tail last 1000 lines
try:
lines = path.read_text(errors="replace").splitlines()
tail = lines[-1000:]
content = "\n".join(tail)
except Exception as e:
return bad(handler, f"Cannot read log: {e}")
stat = path.stat()
return j(handler, {
"name": log_name,
"content": content,
"size": stat.st_size,
"size_human": _human_size(stat.st_size),
"line_count": len(lines),
"tail_count": len(tail),
})