refactor(agent/curator+insights): compact literals/signatures, fold YAML entry parsing and transition diff into comprehensions
This commit is contained in:
Teknium
+48
-106
@@ -44,15 +44,11 @@ def _state_file() -> Path:
|
||||
return get_hermes_home() / "skills" / ".curator_state"
|
||||
|
||||
|
||||
def _default_state() -> Dict[str, Any]:
|
||||
return {
|
||||
def load_state() -> Dict[str, Any]:
|
||||
base: Dict[str, Any] = {
|
||||
"last_run_at": None, "last_run_duration_seconds": None, "last_run_summary": None,
|
||||
"last_run_summary_shown_at": None, "last_report_path": None, "paused": False, "run_count": 0,
|
||||
}
|
||||
|
||||
|
||||
def load_state() -> Dict[str, Any]:
|
||||
base = _default_state()
|
||||
path = _state_file()
|
||||
if not path.exists():
|
||||
return base
|
||||
@@ -136,8 +132,8 @@ def get_archive_after_days() -> int:
|
||||
|
||||
|
||||
def get_prune_builtins() -> bool:
|
||||
"""Bundled built-ins are curation candidates (ON by default); a suppression
|
||||
list keeps them archived across `hermes update` re-seeds. Hub skills never."""
|
||||
"""Bundled built-ins are curation candidates (ON by default); a suppression list
|
||||
keeps them archived across `hermes update` re-seeds. Hub skills are never pruned."""
|
||||
return bool(_load_config().get("prune_builtins", True))
|
||||
|
||||
|
||||
@@ -163,11 +159,9 @@ def should_run_now(now: Optional[datetime] = None) -> bool:
|
||||
tick. ``hermes curator run`` bypasses this; the idle check is the caller's."""
|
||||
if not is_enabled() or is_paused():
|
||||
return False
|
||||
|
||||
state = load_state()
|
||||
last = _parse_iso(state.get("last_run_at"))
|
||||
if now is None:
|
||||
now = datetime.now(timezone.utc)
|
||||
now = now or datetime.now(timezone.utc)
|
||||
if last is None:
|
||||
try:
|
||||
state["last_run_at"] = now.isoformat()
|
||||
@@ -203,18 +197,17 @@ def _archive_as_curator(_u, name: str) -> bool:
|
||||
ledger entry reads as an autonomous transition, not a foreground call."""
|
||||
try:
|
||||
from tools.skill_ledger import reset_ledger_actor, set_ledger_actor
|
||||
_tok = set_ledger_actor("curator")
|
||||
tok = set_ledger_actor("curator")
|
||||
except Exception:
|
||||
_tok = reset_ledger_actor = None # type: ignore[assignment]
|
||||
tok = reset_ledger_actor = None # type: ignore[assignment]
|
||||
try:
|
||||
ok, _msg = _u.archive_skill(name)
|
||||
return _u.archive_skill(name)[0]
|
||||
finally:
|
||||
if _tok is not None:
|
||||
if tok is not None:
|
||||
try:
|
||||
reset_ledger_actor(_tok)
|
||||
reset_ledger_actor(tok)
|
||||
except Exception:
|
||||
pass
|
||||
return ok
|
||||
|
||||
|
||||
def apply_automatic_transitions(now: Optional[datetime] = None) -> Dict[str, int]:
|
||||
@@ -224,8 +217,7 @@ def apply_automatic_transitions(now: Optional[datetime] = None) -> Dict[str, int
|
||||
starts NOW, not at epoch. Returns a counter dict."""
|
||||
from tools import skill_usage as _u
|
||||
|
||||
if now is None:
|
||||
now = datetime.now(timezone.utc)
|
||||
now = now or datetime.now(timezone.utc)
|
||||
stale_cutoff = now - timedelta(days=get_stale_after_days())
|
||||
archive_cutoff = now - timedelta(days=get_archive_after_days())
|
||||
# Cron-referenced skills are in use by definition (usage only bumps when a
|
||||
@@ -542,10 +534,7 @@ def _find_reference(args: Dict[str, Any], needles: Set[str]) -> Optional[str]:
|
||||
|
||||
|
||||
def _classify_removed_skills(
|
||||
removed: List[str],
|
||||
added: List[str],
|
||||
after_names: Set[str],
|
||||
tool_calls: List[Dict[str, Any]],
|
||||
removed: List[str], added: List[str], after_names: Set[str], tool_calls: List[Dict[str, Any]],
|
||||
) -> Dict[str, List[Dict[str, Any]]]:
|
||||
"""Split ``removed`` into consolidated vs pruned. Heuristic: a ``skill_manage``
|
||||
call on a DIFFERENT, surviving-or-new skill whose file_path/content arguments
|
||||
@@ -600,18 +589,13 @@ def _parse_structured_summary(llm_final: str) -> Dict[str, List[Dict[str, str]]]
|
||||
if not isinstance(data, dict):
|
||||
return out
|
||||
|
||||
def _entries(key: str) -> List[Dict[str, Any]]:
|
||||
def _entries(key: str, *fields: str) -> List[Dict[str, str]]:
|
||||
raw = data.get(key) or []
|
||||
return [e for e in raw if isinstance(e, dict)] if isinstance(raw, list) else []
|
||||
cleaned = ({f: _clean_str(e.get(f)) for f in (*fields, "reason")} for e in raw if isinstance(e, dict)) if isinstance(raw, list) else ()
|
||||
return [e for e in cleaned if all(e[f] for f in fields)]
|
||||
|
||||
for entry in _entries("consolidations"):
|
||||
frm, into = _clean_str(entry.get("from")), _clean_str(entry.get("into"))
|
||||
if frm and into:
|
||||
out["consolidations"].append({"from": frm, "into": into, "reason": _clean_str(entry.get("reason"))})
|
||||
for entry in _entries("prunings"):
|
||||
name = _clean_str(entry.get("name"))
|
||||
if name:
|
||||
out["prunings"].append({"name": name, "reason": _clean_str(entry.get("reason"))})
|
||||
out["consolidations"] = _entries("consolidations", "from", "into")
|
||||
out["prunings"] = _entries("prunings", "name")
|
||||
return out
|
||||
|
||||
|
||||
@@ -633,11 +617,8 @@ def _extract_absorbed_into_declarations(tool_calls: List[Dict[str, Any]]) -> Dic
|
||||
|
||||
|
||||
def _reconcile_classification(
|
||||
removed: List[str],
|
||||
heuristic: Dict[str, List[Dict[str, Any]]],
|
||||
model_block: Dict[str, List[Dict[str, str]]],
|
||||
destinations: Set[str],
|
||||
absorbed_declarations: Optional[Dict[str, Dict[str, Any]]] = None,
|
||||
removed: List[str], heuristic: Dict[str, List[Dict[str, Any]]], model_block: Dict[str, List[Dict[str, str]]],
|
||||
destinations: Set[str], absorbed_declarations: Optional[Dict[str, Dict[str, Any]]] = None,
|
||||
) -> Dict[str, List[Dict[str, Any]]]:
|
||||
"""Merge heuristic (tool-call evidence) with the model's structured block.
|
||||
First match wins; every removed skill lands in exactly one bucket:
|
||||
@@ -792,15 +773,8 @@ def _new_run_dir(started_at: datetime) -> Optional[Path]:
|
||||
|
||||
|
||||
def _write_run_report(
|
||||
*,
|
||||
started_at: datetime,
|
||||
elapsed_seconds: float,
|
||||
auto_counts: Dict[str, int],
|
||||
auto_summary: str,
|
||||
before_report: List[Dict[str, Any]],
|
||||
before_names: Set[str],
|
||||
after_report: List[Dict[str, Any]],
|
||||
llm_meta: Dict[str, Any],
|
||||
*, started_at: datetime, elapsed_seconds: float, auto_counts: Dict[str, int], auto_summary: str,
|
||||
before_report: List[Dict[str, Any]], before_names: Set[str], after_report: List[Dict[str, Any]], llm_meta: Dict[str, Any],
|
||||
) -> Optional[Path]:
|
||||
"""Write run.json + REPORT.md under logs/curator/{YYYYMMDD-HHMMSS}/. Returns
|
||||
the report dir, or None if it couldn't be created (reporting is best-effort)."""
|
||||
@@ -812,47 +786,31 @@ def _write_run_report(
|
||||
after_by_name, before_by_name = _by_name(after_report), _by_name(before_report)
|
||||
diff = _diff_and_classify(before_names, set(after_by_name), tool_calls, llm_meta.get("final", "") or "")
|
||||
|
||||
transitions: List[Dict[str, str]] = []
|
||||
for name in sorted(diff.after_names & before_names):
|
||||
s_before = (before_by_name.get(name) or {}).get("state")
|
||||
s_after = (after_by_name.get(name) or {}).get("state")
|
||||
if s_before and s_after and s_before != s_after:
|
||||
transitions.append({"name": name, "from": s_before, "to": s_after})
|
||||
states = ((n, (before_by_name.get(n) or {}).get("state"), (after_by_name.get(n) or {}).get("state")) for n in sorted(diff.after_names & before_names))
|
||||
transitions = [{"name": n, "from": b, "to": a} for n, b, a in states if b and a and b != a]
|
||||
|
||||
tc_counts: Dict[str, int] = dict(Counter(tc.get("name", "unknown") for tc in tool_calls))
|
||||
cron_rewrites = _rewrite_cron_refs(diff.consolidated, diff.pruned)
|
||||
jobs_updated = int(cron_rewrites.get("jobs_updated", 0))
|
||||
|
||||
payload = {
|
||||
"started_at": started_at.isoformat(),
|
||||
"duration_seconds": round(elapsed_seconds, 2),
|
||||
"model": llm_meta.get("model", ""),
|
||||
"provider": llm_meta.get("provider", ""),
|
||||
"started_at": started_at.isoformat(), "duration_seconds": round(elapsed_seconds, 2),
|
||||
"model": llm_meta.get("model", ""), "provider": llm_meta.get("provider", ""),
|
||||
"auto_transitions": auto_counts,
|
||||
"counts": {
|
||||
"before": len(before_names),
|
||||
"after": len(diff.after_names),
|
||||
"before": len(before_names), "after": len(diff.after_names),
|
||||
"delta": len(diff.after_names) - len(before_names),
|
||||
"archived_this_run": len(diff.removed),
|
||||
"added_this_run": len(diff.added),
|
||||
"consolidated_this_run": len(diff.consolidated),
|
||||
"pruned_this_run": len(diff.pruned),
|
||||
"state_transitions": len(transitions),
|
||||
"cron_jobs_rewritten": jobs_updated,
|
||||
"archived_this_run": len(diff.removed), "added_this_run": len(diff.added),
|
||||
"consolidated_this_run": len(diff.consolidated), "pruned_this_run": len(diff.pruned),
|
||||
"state_transitions": len(transitions), "cron_jobs_rewritten": jobs_updated,
|
||||
"tool_calls_total": sum(tc_counts.values()),
|
||||
},
|
||||
"tool_call_counts": tc_counts,
|
||||
"archived": diff.removed,
|
||||
"consolidated": diff.consolidated,
|
||||
"pruned": diff.pruned,
|
||||
"pruned_names": [p["name"] for p in diff.pruned],
|
||||
"added": diff.added,
|
||||
"state_transitions": transitions,
|
||||
"cron_rewrites": cron_rewrites,
|
||||
"llm_final": llm_meta.get("final", ""),
|
||||
"llm_summary": llm_meta.get("summary", ""),
|
||||
"llm_error": llm_meta.get("error"),
|
||||
"tool_calls": llm_meta.get("tool_calls", []),
|
||||
"archived": diff.removed, "consolidated": diff.consolidated, "pruned": diff.pruned,
|
||||
"pruned_names": [p["name"] for p in diff.pruned], "added": diff.added,
|
||||
"state_transitions": transitions, "cron_rewrites": cron_rewrites,
|
||||
"llm_final": llm_meta.get("final", ""), "llm_summary": llm_meta.get("summary", ""),
|
||||
"llm_error": llm_meta.get("error"), "tool_calls": llm_meta.get("tool_calls", []),
|
||||
}
|
||||
|
||||
_write_json(run_dir / "run.json", payload, "run.json")
|
||||
@@ -1029,6 +987,9 @@ def _safe_curated_report() -> List[Dict[str, Any]]:
|
||||
return []
|
||||
|
||||
|
||||
_AUTO_LABELS = (("marked_stale", "marked stale"), ("archived", "archived"), ("reactivated", "reactivated"))
|
||||
|
||||
|
||||
def _consolidation_pass(prefix: str, auto_summary: str, dry_run: bool, before_names: Set[str]) -> tuple:
|
||||
"""The LLM half of a run: fork (unless no candidates), then append the rename
|
||||
map (`old-name → umbrella`) so users needn't dig into REPORT.md.
|
||||
@@ -1067,10 +1028,8 @@ def _consolidation_pass(prefix: str, auto_summary: str, dry_run: bool, before_na
|
||||
|
||||
|
||||
def run_curator_review(
|
||||
on_summary: Optional[Callable[[str], None]] = None,
|
||||
synchronous: bool = False,
|
||||
dry_run: bool = False,
|
||||
consolidate: Optional[bool] = None,
|
||||
on_summary: Optional[Callable[[str], None]] = None, synchronous: bool = False,
|
||||
dry_run: bool = False, consolidate: Optional[bool] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Execute a single curator review pass: (1) automatic state transitions (no
|
||||
LLM); (2) if *consolidate* and there are candidates, fork an AIAgent on the
|
||||
@@ -1086,8 +1045,7 @@ def run_curator_review(
|
||||
if consolidate is None:
|
||||
consolidate = get_consolidate()
|
||||
start = datetime.now(timezone.utc)
|
||||
if dry_run:
|
||||
# Count candidates without mutating state.
|
||||
if dry_run: # count candidates without mutating state
|
||||
counts = {"checked": len(_safe_curated_report()), "marked_stale": 0, "archived": 0, "reactivated": 0}
|
||||
else:
|
||||
# Pre-mutation snapshot — best-effort, never blocks the run: a transient
|
||||
@@ -1102,9 +1060,7 @@ def run_curator_review(
|
||||
counts = apply_automatic_transitions(now=start)
|
||||
|
||||
auto_summary = ", ".join(
|
||||
f"{counts[key]} {label}"
|
||||
for key, label in (("marked_stale", "marked stale"), ("archived", "archived"), ("reactivated", "reactivated"))
|
||||
if counts[key]
|
||||
f"{counts[key]} {label}" for key, label in _AUTO_LABELS if counts[key]
|
||||
) or "no changes"
|
||||
|
||||
# Persist before the LLM pass so a crash mid-review still records the run.
|
||||
@@ -1139,14 +1095,12 @@ def run_curator_review(
|
||||
try:
|
||||
report_path = _write_run_report(
|
||||
started_at=start, elapsed_seconds=elapsed, auto_counts=counts, auto_summary=auto_summary,
|
||||
before_report=before_report, before_names=before_names,
|
||||
after_report=_safe_curated_report(), llm_meta=llm_meta,
|
||||
before_report=before_report, before_names=before_names, after_report=_safe_curated_report(), llm_meta=llm_meta,
|
||||
)
|
||||
if report_path is not None:
|
||||
state2["last_report_path"] = str(report_path)
|
||||
except Exception as e:
|
||||
logger.debug("Curator report write failed: %s", e, exc_info=True)
|
||||
|
||||
save_state(state2)
|
||||
_notify(on_summary, f"curator: {final_summary}")
|
||||
|
||||
@@ -1284,14 +1238,9 @@ def _run_llm_review(prompt: str) -> Dict[str, Any]:
|
||||
agent_kwargs["acp_command"] = acp_command
|
||||
agent_kwargs["acp_args"] = list(rp.get("args") or [])
|
||||
review_agent = AIAgent(
|
||||
model=model_name,
|
||||
provider=provider,
|
||||
api_key=rp.get("api_key"),
|
||||
base_url=rp.get("base_url"),
|
||||
api_mode=rp.get("api_mode"),
|
||||
credential_pool=rp.get("credential_pool"),
|
||||
request_overrides=request_overrides,
|
||||
**agent_kwargs,
|
||||
model=model_name, provider=provider, api_key=rp.get("api_key"), base_url=rp.get("base_url"),
|
||||
api_mode=rp.get("api_mode"), credential_pool=rp.get("credential_pool"),
|
||||
request_overrides=request_overrides, **agent_kwargs,
|
||||
# No ``terminal``: a shell mv/cp/rm under the skills tree writes bytes
|
||||
# with NO ledger entry, so rollback would restore a hollow skill. Every
|
||||
# mutation goes through ledgered skill_manage; dropping the toolset
|
||||
@@ -1299,10 +1248,7 @@ def _run_llm_review(prompt: str) -> Dict[str, Any]:
|
||||
enabled_toolsets=["skills"],
|
||||
# Umbrella-building over hundreds of skills takes 50-100 API calls.
|
||||
max_iterations=9999,
|
||||
quiet_mode=True,
|
||||
platform="curator",
|
||||
skip_context_files=True,
|
||||
skip_memory=True,
|
||||
quiet_mode=True, platform="curator", skip_context_files=True, skip_memory=True,
|
||||
)
|
||||
# Disable recursive nudges — the curator must never spawn its own review.
|
||||
review_agent._memory_nudge_interval = 0
|
||||
@@ -1335,17 +1281,13 @@ def _run_llm_review(prompt: str) -> Dict[str, Any]:
|
||||
# --- Public entrypoint for the session-start hook ---
|
||||
|
||||
def maybe_run_curator(
|
||||
*,
|
||||
idle_for_seconds: Optional[float] = None,
|
||||
on_summary: Optional[Callable[[str], None]] = None,
|
||||
*, idle_for_seconds: Optional[float] = None, on_summary: Optional[Callable[[str], None]] = None,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Best-effort: run a curator pass if all gates pass. Returns the result
|
||||
dict if a pass was started, else None. Never raises."""
|
||||
try:
|
||||
# Idle gating: only enforce when the caller provided a measurement.
|
||||
if not should_run_now() or (
|
||||
idle_for_seconds is not None and idle_for_seconds < get_min_idle_hours() * 3600.0
|
||||
):
|
||||
if not should_run_now() or (idle_for_seconds is not None and idle_for_seconds < get_min_idle_hours() * 3600.0):
|
||||
return None
|
||||
return run_curator_review(on_summary=on_summary)
|
||||
except Exception as e:
|
||||
|
||||
+92
-267
@@ -13,13 +13,7 @@ from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from agent.usage_pricing import (
|
||||
CanonicalUsage,
|
||||
estimate_usage_cost,
|
||||
format_cost_label,
|
||||
format_duration_compact,
|
||||
has_known_pricing,
|
||||
)
|
||||
from agent.usage_pricing import CanonicalUsage, estimate_usage_cost, format_cost_label, format_duration_compact, has_known_pricing
|
||||
|
||||
_TOKEN_KEYS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens")
|
||||
_SKILL_TOOLS = {"skill_view", "skill_manage"}
|
||||
@@ -30,16 +24,8 @@ def _fmt_est_cost(est_cost: float) -> str:
|
||||
return format_cost_label(Decimal(str(est_cost)))
|
||||
|
||||
|
||||
def _estimate_cost(
|
||||
session_or_model: Dict[str, Any] | str,
|
||||
input_tokens: int = 0,
|
||||
output_tokens: int = 0,
|
||||
*,
|
||||
cache_read_tokens: int = 0,
|
||||
cache_write_tokens: int = 0,
|
||||
provider: Optional[str] = None,
|
||||
base_url: Optional[str] = None,
|
||||
) -> tuple[float, str]:
|
||||
def _estimate_cost(session_or_model: Dict[str, Any] | str, input_tokens: int = 0, output_tokens: int = 0, *, cache_read_tokens: int = 0,
|
||||
cache_write_tokens: int = 0, provider: Optional[str] = None, base_url: Optional[str] = None) -> tuple[float, str]:
|
||||
"""Estimate the USD cost for a session row or a model/token tuple."""
|
||||
if isinstance(session_or_model, dict):
|
||||
s = session_or_model
|
||||
@@ -55,9 +41,7 @@ def _estimate_cost(
|
||||
|
||||
def _bar_chart(values: List[int], max_width: int = 20) -> List[str]:
|
||||
peak = max(values) if values else 1
|
||||
if peak == 0:
|
||||
return ["" for _ in values]
|
||||
return ["█" * max(1, int(v / peak * max_width)) if v > 0 else "" for v in values]
|
||||
return ["" for _ in values] if peak == 0 else ["█" * max(1, int(v / peak * max_width)) if v > 0 else "" for v in values]
|
||||
|
||||
|
||||
def _short_model(model: Optional[str]) -> str:
|
||||
@@ -91,10 +75,8 @@ def _day(ts: Any) -> str:
|
||||
|
||||
|
||||
def _scoped(before: str, after: str = "", *, src: str = " AND s.source = ?") -> tuple[str, str]:
|
||||
"""(unfiltered, source-filtered) query pair sharing one body.
|
||||
|
||||
Built once at class definition, so no runtime value can alter query structure.
|
||||
"""
|
||||
"""(unfiltered, source-filtered) query pair sharing one body. Built once at class definition,
|
||||
so no runtime value can alter query structure."""
|
||||
return before + after, before + src + after
|
||||
|
||||
|
||||
@@ -173,9 +155,7 @@ class InsightsEngine:
|
||||
self._conn = db._conn
|
||||
try:
|
||||
self._has_assistant_calls_index = bool(self._conn.execute(
|
||||
"SELECT 1 FROM sqlite_master WHERE type='index' AND name=?",
|
||||
(self._MESSAGES_ASSISTANT_CALLS_INDEX,),
|
||||
).fetchone())
|
||||
"SELECT 1 FROM sqlite_master WHERE type='index' AND name=?", (self._MESSAGES_ASSISTANT_CALLS_INDEX,)).fetchone())
|
||||
except sqlite3.Error:
|
||||
self._has_assistant_calls_index = False
|
||||
if not self._has_assistant_calls_index:
|
||||
@@ -185,33 +165,25 @@ class InsightsEngine:
|
||||
setattr(self, base + suffix, getattr(self, base + suffix).replace(strip, ""))
|
||||
|
||||
def _query(self, base: str, cutoff: float, source: Optional[str]) -> list:
|
||||
"""Rows of ``<base>_WITH_SOURCE`` or ``<base>_ALL`` (instance attrs, so the
|
||||
unpinned fallback applies)."""
|
||||
if source:
|
||||
return self._conn.execute(getattr(self, base + "_WITH_SOURCE"), (cutoff, source)).fetchall()
|
||||
return self._conn.execute(getattr(self, base + "_ALL"), (cutoff,)).fetchall()
|
||||
"""Rows of ``<base>_WITH_SOURCE`` or ``<base>_ALL`` (instance attrs, so the unpinned fallback applies)."""
|
||||
sql, params = (getattr(self, base + "_WITH_SOURCE"), (cutoff, source)) if source else (getattr(self, base + "_ALL"), (cutoff,))
|
||||
return self._conn.execute(sql, params).fetchall()
|
||||
|
||||
def generate(self, days: int = 30, source: str = None) -> Dict[str, Any]:
|
||||
"""Generate a complete insights report for the last ``days`` days,
|
||||
optionally filtered by source platform."""
|
||||
"""Generate a complete insights report for the last ``days`` days, optionally filtered by source platform."""
|
||||
cutoff = time.time() - (days * 86400)
|
||||
# Drain the SessionDB's async accounting queue so counters are exact
|
||||
# (self.db may be a raw sqlite3 connection in tests — guard).
|
||||
flush = getattr(self.db, "flush_token_counts", None)
|
||||
if callable(flush):
|
||||
flush()
|
||||
|
||||
sessions = self._get_sessions(cutoff, source)
|
||||
tool_usage = self._get_tool_usage(cutoff, source)
|
||||
skill_usage = self._get_skill_usage(cutoff, source)
|
||||
message_stats = self._get_message_stats(cutoff, source)
|
||||
if not sessions:
|
||||
return {
|
||||
"days": days, "source_filter": source, "empty": True, "overview": {}, "models": [],
|
||||
"platforms": [], "tools": [], "skills": self._compute_skill_breakdown([]),
|
||||
"activity": {}, "top_sessions": [],
|
||||
}
|
||||
|
||||
return {"days": days, "source_filter": source, "empty": True, "overview": {}, "models": [], "platforms": [], "tools": [],
|
||||
"skills": self._compute_skill_breakdown([]), "activity": {}, "top_sessions": []}
|
||||
models = self._compute_model_breakdown(sessions, cutoff, source)
|
||||
return {
|
||||
"days": days, "source_filter": source, "empty": False, "generated_at": time.time(),
|
||||
@@ -228,10 +200,8 @@ class InsightsEngine:
|
||||
"""Analytics-usage payload (tools + skills) without a full generate(); the
|
||||
instr()-prefiltered skill query loads only skill_view/skill_manage messages."""
|
||||
cutoff = time.time() - (days * 86400)
|
||||
return {
|
||||
"tools": self._compute_tool_breakdown(self._get_tool_usage(cutoff, source)),
|
||||
"skills": self._compute_skill_breakdown(self._get_skill_usage(cutoff, source)),
|
||||
}
|
||||
return {"tools": self._compute_tool_breakdown(self._get_tool_usage(cutoff, source)),
|
||||
"skills": self._compute_skill_breakdown(self._get_skill_usage(cutoff, source))}
|
||||
|
||||
# ------------------------------------------------------------------ SQL
|
||||
|
||||
@@ -245,19 +215,15 @@ class InsightsEngine:
|
||||
tool_counts = Counter()
|
||||
for row in self._query("_GET_TOOL_NAMES", cutoff, source):
|
||||
tool_counts[row["tool_name"]] += row["count"]
|
||||
|
||||
tool_calls_counts = Counter()
|
||||
for row in self._query("_GET_TOOL_CALLS", cutoff, source):
|
||||
try:
|
||||
tool_calls_counts.update(filter(None, (fn.get("name") for fn in _iter_functions(row["tool_calls"]))))
|
||||
except (TypeError, AttributeError):
|
||||
continue
|
||||
|
||||
if tool_calls_counts and tool_counts:
|
||||
tool_counts = Counter({
|
||||
tool: max(tool_counts.get(tool, 0), tool_calls_counts.get(tool, 0))
|
||||
for tool in set(tool_counts) | set(tool_calls_counts)
|
||||
})
|
||||
tool_counts = Counter({tool: max(tool_counts.get(tool, 0), tool_calls_counts.get(tool, 0))
|
||||
for tool in set(tool_counts) | set(tool_calls_counts)})
|
||||
elif tool_calls_counts:
|
||||
tool_counts = tool_calls_counts
|
||||
return [{"tool_name": name, "count": count} for name, count in tool_counts.most_common()]
|
||||
@@ -274,9 +240,7 @@ class InsightsEngine:
|
||||
skill_name = (_parse_json(func.get("arguments"), dict) or {}).get("name")
|
||||
if not isinstance(skill_name, str) or not skill_name.strip():
|
||||
continue
|
||||
entry = skill_counts.setdefault(
|
||||
skill_name, {"skill": skill_name, "view_count": 0, "manage_count": 0, "last_used_at": None},
|
||||
)
|
||||
entry = skill_counts.setdefault(skill_name, {"skill": skill_name, "view_count": 0, "manage_count": 0, "last_used_at": None})
|
||||
entry["view_count" if tool_name == "skill_view" else "manage_count"] += 1
|
||||
if timestamp is not None and (entry["last_used_at"] is None or timestamp > entry["last_used_at"]):
|
||||
entry["last_used_at"] = timestamp
|
||||
@@ -284,13 +248,10 @@ class InsightsEngine:
|
||||
|
||||
def _get_message_stats(self, cutoff: float, source: str = None) -> Dict:
|
||||
rows = self._query("_GET_MESSAGE_STATS", cutoff, source)
|
||||
return dict(rows[0]) if rows else {
|
||||
"total_messages": 0, "user_messages": 0, "assistant_messages": 0, "tool_messages": 0,
|
||||
}
|
||||
return dict(rows[0]) if rows else {"total_messages": 0, "user_messages": 0, "assistant_messages": 0, "tool_messages": 0}
|
||||
|
||||
def _get_model_usage(self, cutoff: float, source: str = None) -> List[Dict]:
|
||||
"""Per-model usage rows; [] when the table is missing (older DB) so the
|
||||
caller falls back to the per-session aggregate."""
|
||||
"""Per-model usage rows; [] when the table is missing (older DB) so the caller falls back to the per-session aggregate."""
|
||||
try:
|
||||
return [dict(row) for row in self._query("_GET_MODEL_USAGE", cutoff, source)]
|
||||
except sqlite3.OperationalError:
|
||||
@@ -304,16 +265,12 @@ class InsightsEngine:
|
||||
# main-loop usage only — sum the breakdown when available so overview
|
||||
# totals match the per-model table and aux spend isn't undercounted.
|
||||
rows = models or sessions
|
||||
total_input, total_output, total_cache_read, total_cache_write = (
|
||||
sum(int(r.get(k) or 0) for r in rows) for k in _TOKEN_KEYS
|
||||
)
|
||||
total_input, total_output, total_cache_read, total_cache_write = (sum(int(r.get(k) or 0) for r in rows) for k in _TOKEN_KEYS)
|
||||
total_tokens = total_input + total_output + total_cache_read + total_cache_write
|
||||
total_tool_calls = sum(s.get("tool_call_count") or 0 for s in sessions)
|
||||
total_messages = sum(s.get("message_count") or 0 for s in sessions)
|
||||
|
||||
total_cost = actual_cost = 0.0
|
||||
models_with_pricing, models_without_pricing = set(), set()
|
||||
status_counts = Counter()
|
||||
models_with_pricing, models_without_pricing, status_counts = set(), set(), Counter()
|
||||
for s in sessions:
|
||||
model = s.get("model") or ""
|
||||
estimated, status = _estimate_cost(s)
|
||||
@@ -324,26 +281,16 @@ class InsightsEngine:
|
||||
(models_with_pricing if known else models_without_pricing).add(_short_model(model))
|
||||
if models:
|
||||
total_cost = sum(float(m.get("cost") or 0.0) for m in models)
|
||||
|
||||
# Guard against negative durations from clock drift.
|
||||
durations = [
|
||||
s["ended_at"] - s["started_at"]
|
||||
for s in sessions
|
||||
if s.get("started_at") and s.get("ended_at") and s["ended_at"] > s["started_at"]
|
||||
]
|
||||
durations = [s["ended_at"] - s["started_at"] for s in sessions
|
||||
if s.get("started_at") and s.get("ended_at") and s["ended_at"] > s["started_at"]]
|
||||
started = [s["started_at"] for s in sessions if s.get("started_at")]
|
||||
n = len(sessions)
|
||||
return {
|
||||
"total_sessions": n,
|
||||
"total_messages": total_messages,
|
||||
"total_tool_calls": total_tool_calls,
|
||||
"total_input_tokens": total_input,
|
||||
"total_output_tokens": total_output,
|
||||
"total_cache_read_tokens": total_cache_read,
|
||||
"total_cache_write_tokens": total_cache_write,
|
||||
"total_tokens": total_tokens,
|
||||
"estimated_cost": total_cost,
|
||||
"actual_cost": actual_cost,
|
||||
"total_sessions": n, "total_messages": total_messages, "total_tool_calls": total_tool_calls,
|
||||
"total_input_tokens": total_input, "total_output_tokens": total_output,
|
||||
"total_cache_read_tokens": total_cache_read, "total_cache_write_tokens": total_cache_write,
|
||||
"total_tokens": total_tokens, "estimated_cost": total_cost, "actual_cost": actual_cost,
|
||||
"total_hours": sum(durations) / 3600 if durations else 0,
|
||||
"avg_session_duration": sum(durations) / len(durations) if durations else 0,
|
||||
"avg_messages_per_session": total_messages / n if sessions else 0,
|
||||
@@ -359,19 +306,15 @@ class InsightsEngine:
|
||||
"included_cost_sessions": status_counts["included"],
|
||||
}
|
||||
|
||||
def _compute_model_breakdown(
|
||||
self, sessions: List[Dict], cutoff: float, source: str = None
|
||||
) -> List[Dict]:
|
||||
def _compute_model_breakdown(self, sessions: List[Dict], cutoff: float, source: str = None) -> List[Dict]:
|
||||
"""Tokens/cost per model from session_model_usage, so a session that
|
||||
switched models via ``/model`` splits across every model it used.
|
||||
Sessions without per-model rows (pre-table data) fall back to their
|
||||
single recorded aggregate. Tool calls aren't tied to an API call, so
|
||||
they stay attributed to the session's recorded model."""
|
||||
count_keys = _TOKEN_KEYS + ("reasoning_tokens", "api_call_count")
|
||||
model_data = defaultdict(lambda: {
|
||||
"sessions": set(), **dict.fromkeys(_TOKEN_KEYS, 0), "reasoning_tokens": 0, "total_tokens": 0,
|
||||
"api_calls": 0, "tool_calls": 0, "cost": 0.0, "actual_cost": 0.0,
|
||||
})
|
||||
model_data = defaultdict(lambda: {"sessions": set(), **dict.fromkeys(_TOKEN_KEYS, 0), "reasoning_tokens": 0, "total_tokens": 0,
|
||||
"api_calls": 0, "tool_calls": 0, "cost": 0.0, "actual_cost": 0.0})
|
||||
|
||||
def _accumulate(model, provider, base_url, session_id, counts: Dict[str, int], *,
|
||||
stored_cost=None, actual_cost=None, cost_status=None):
|
||||
@@ -383,24 +326,15 @@ class InsightsEngine:
|
||||
d["total_tokens"] += sum(counts[k] for k in _TOKEN_KEYS)
|
||||
d["api_calls"] += counts["api_call_count"]
|
||||
if stored_cost is None:
|
||||
estimate, status = _estimate_cost(
|
||||
model, counts["input_tokens"], counts["output_tokens"],
|
||||
cache_read_tokens=counts["cache_read_tokens"], cache_write_tokens=counts["cache_write_tokens"],
|
||||
provider=provider or None, base_url=base_url,
|
||||
)
|
||||
estimate, status = _estimate_cost(model, counts["input_tokens"], counts["output_tokens"], cache_read_tokens=counts["cache_read_tokens"],
|
||||
cache_write_tokens=counts["cache_write_tokens"], provider=provider or None, base_url=base_url)
|
||||
else:
|
||||
estimate, status = float(stored_cost or 0.0), cost_status or "unknown"
|
||||
d["cost"] += estimate
|
||||
d["actual_cost"] += float(actual_cost or 0.0)
|
||||
d["cost_status"] = status
|
||||
if has_known_pricing(model, provider or None, base_url):
|
||||
d["has_pricing"] = True
|
||||
else:
|
||||
d.setdefault("has_pricing", False)
|
||||
|
||||
usage_totals = defaultdict(lambda: dict.fromkeys(count_keys, 0) | {
|
||||
"estimated_cost_usd": 0.0, "actual_cost_usd": 0.0,
|
||||
})
|
||||
d["has_pricing"] = has_known_pricing(model, provider or None, base_url) or d.get("has_pricing", False)
|
||||
usage_totals = defaultdict(lambda: dict.fromkeys(count_keys, 0) | {"estimated_cost_usd": 0.0, "actual_cost_usd": 0.0})
|
||||
for r in self._get_model_usage(cutoff, source):
|
||||
totals: Dict[str, Any] = usage_totals[r["session_id"]]
|
||||
counts = {key: r[key] or 0 for key in count_keys}
|
||||
@@ -408,12 +342,9 @@ class InsightsEngine:
|
||||
totals[key] += counts[key]
|
||||
totals["estimated_cost_usd"] += r["estimated_cost_usd"] or 0.0
|
||||
totals["actual_cost_usd"] += r["actual_cost_usd"] or 0.0
|
||||
_accumulate(
|
||||
r["model"], r["billing_provider"], r.get("billing_base_url"), r["session_id"], counts,
|
||||
stored_cost=r["estimated_cost_usd"] if r.get("cost_status") or r.get("cost_source") else None,
|
||||
actual_cost=r["actual_cost_usd"], cost_status=r.get("cost_status"),
|
||||
)
|
||||
|
||||
_accumulate(r["model"], r["billing_provider"], r.get("billing_base_url"), r["session_id"], counts,
|
||||
stored_cost=r["estimated_cost_usd"] if r.get("cost_status") or r.get("cost_source") else None,
|
||||
actual_cost=r["actual_cost_usd"], cost_status=r.get("cost_status"))
|
||||
# Reconcile against the aggregate row: covers legacy sessions,
|
||||
# interrupted migrations, and absolute cumulative updates without
|
||||
# double-counting already-attributed route deltas.
|
||||
@@ -424,31 +355,20 @@ class InsightsEngine:
|
||||
residual_cost = max(0.0, float(s.get("estimated_cost_usd") or 0.0) - totals["estimated_cost_usd"])
|
||||
residual_actual = max(0.0, float(s.get("actual_cost_usd") or 0.0) - totals["actual_cost_usd"])
|
||||
if any(residual.values()) or residual_cost or residual_actual:
|
||||
_accumulate(
|
||||
s.get("model"), s.get("billing_provider"), s.get("billing_base_url"), s["id"], residual,
|
||||
stored_cost=residual_cost, actual_cost=residual_actual, cost_status=s.get("cost_status"),
|
||||
)
|
||||
|
||||
_accumulate(s.get("model"), s.get("billing_provider"), s.get("billing_base_url"), s["id"], residual,
|
||||
stored_cost=residual_cost, actual_cost=residual_actual, cost_status=s.get("cost_status"))
|
||||
for s in sessions:
|
||||
tool_calls = s.get("tool_call_count") or 0
|
||||
if tool_calls:
|
||||
model_data[_short_model(s.get("model"))]["tool_calls"] += tool_calls
|
||||
|
||||
result = []
|
||||
for model, data in model_data.items():
|
||||
entry = {"model": model, **data, "sessions": len(data["sessions"])}
|
||||
# Models seen only via tool-call attribution never hit _accumulate —
|
||||
# default these so the output shape is uniform for JSON consumers.
|
||||
entry.setdefault("has_pricing", False)
|
||||
entry.setdefault("cost_status", "unknown")
|
||||
result.append(entry)
|
||||
result.sort(key=lambda x: (x["total_tokens"], x["sessions"]), reverse=True)
|
||||
return result
|
||||
if s.get("tool_call_count"):
|
||||
model_data[_short_model(s.get("model"))]["tool_calls"] += s["tool_call_count"]
|
||||
# Models seen only via tool-call attribution never hit _accumulate —
|
||||
# default has_pricing/cost_status so the output shape is uniform for JSON consumers.
|
||||
defaults = (("has_pricing", False), ("cost_status", "unknown"))
|
||||
result = [{"model": model, **data, "sessions": len(data["sessions"]), **{k: v for k, v in defaults if k not in data}}
|
||||
for model, data in model_data.items()]
|
||||
return sorted(result, key=lambda x: (x["total_tokens"], x["sessions"]), reverse=True)
|
||||
|
||||
def _compute_platform_breakdown(self, sessions: List[Dict]) -> List[Dict]:
|
||||
platform_data = defaultdict(lambda: {
|
||||
"sessions": 0, "messages": 0, **dict.fromkeys(_TOKEN_KEYS, 0), "total_tokens": 0, "tool_calls": 0,
|
||||
})
|
||||
platform_data = defaultdict(lambda: {"sessions": 0, "messages": 0, **dict.fromkeys(_TOKEN_KEYS, 0), "total_tokens": 0, "tool_calls": 0})
|
||||
for s in sessions:
|
||||
d = platform_data[s.get("source") or "unknown"]
|
||||
d["sessions"] += 1
|
||||
@@ -457,53 +377,32 @@ class InsightsEngine:
|
||||
d[k] += s.get(k) or 0
|
||||
d["total_tokens"] += s.get(k) or 0
|
||||
d["tool_calls"] += s.get("tool_call_count") or 0
|
||||
|
||||
result = [{"platform": platform, **data} for platform, data in platform_data.items()]
|
||||
result.sort(key=lambda x: x["sessions"], reverse=True)
|
||||
return result
|
||||
return sorted(({"platform": platform, **data} for platform, data in platform_data.items()), key=lambda x: x["sessions"], reverse=True)
|
||||
|
||||
def _compute_tool_breakdown(self, tool_usage: List[Dict]) -> List[Dict]:
|
||||
"""Ranked tool list with percentages."""
|
||||
total_calls = sum(t["count"] for t in tool_usage)
|
||||
return [
|
||||
{"tool": t["tool_name"], "count": t["count"], "percentage": (t["count"] / total_calls * 100) if total_calls else 0}
|
||||
for t in tool_usage
|
||||
]
|
||||
return [{"tool": t["tool_name"], "count": t["count"], "percentage": (t["count"] / total_calls * 100) if total_calls else 0} for t in tool_usage]
|
||||
|
||||
def _compute_skill_breakdown(self, skill_usage: List[Dict]) -> Dict[str, Any]:
|
||||
"""Per-skill usage → summary + ranked list."""
|
||||
total_skill_loads = sum(s["view_count"] for s in skill_usage)
|
||||
total_skill_edits = sum(s["manage_count"] for s in skill_usage)
|
||||
total_skill_actions = total_skill_loads + total_skill_edits
|
||||
|
||||
top_skills = []
|
||||
for skill in skill_usage:
|
||||
total_count = skill["view_count"] + skill["manage_count"]
|
||||
top_skills.append({
|
||||
"skill": skill["skill"], "view_count": skill["view_count"], "manage_count": skill["manage_count"],
|
||||
"total_count": total_count,
|
||||
"percentage": (total_count / total_skill_actions * 100) if total_skill_actions else 0,
|
||||
"last_used_at": skill.get("last_used_at"),
|
||||
})
|
||||
top_skills.sort(
|
||||
key=lambda s: (s["total_count"], s["view_count"], s["manage_count"], s["last_used_at"] or 0, s["skill"]),
|
||||
reverse=True,
|
||||
)
|
||||
top_skills = [{
|
||||
"skill": skill["skill"], "view_count": skill["view_count"], "manage_count": skill["manage_count"], "total_count": total_count,
|
||||
"percentage": (total_count / total_skill_actions * 100) if total_skill_actions else 0, "last_used_at": skill.get("last_used_at"),
|
||||
} for skill in skill_usage for total_count in (skill["view_count"] + skill["manage_count"],)]
|
||||
top_skills.sort(key=lambda s: (s["total_count"], s["view_count"], s["manage_count"], s["last_used_at"] or 0, s["skill"]), reverse=True)
|
||||
return {
|
||||
"summary": {
|
||||
"total_skill_loads": total_skill_loads,
|
||||
"total_skill_edits": total_skill_edits,
|
||||
"total_skill_actions": total_skill_actions,
|
||||
"distinct_skills_used": len(skill_usage),
|
||||
},
|
||||
"summary": {"total_skill_loads": total_skill_loads, "total_skill_edits": total_skill_edits,
|
||||
"total_skill_actions": total_skill_actions, "distinct_skills_used": len(skill_usage)},
|
||||
"top_skills": top_skills,
|
||||
}
|
||||
|
||||
def _compute_activity_patterns(self, sessions: List[Dict]) -> Dict:
|
||||
"""Activity by day of week, hour, and active-day streak."""
|
||||
day_counts = Counter() # 0=Monday ... 6=Sunday
|
||||
hour_counts = Counter()
|
||||
daily_counts = Counter() # "YYYY-MM-DD" -> count
|
||||
day_counts, hour_counts, daily_counts = Counter(), Counter(), Counter() # weekday (0=Monday), hour, "YYYY-MM-DD"
|
||||
for s in sessions:
|
||||
ts = s.get("started_at")
|
||||
if not ts:
|
||||
@@ -512,11 +411,9 @@ class InsightsEngine:
|
||||
day_counts[dt.weekday()] += 1
|
||||
hour_counts[dt.hour] += 1
|
||||
daily_counts[dt.strftime("%Y-%m-%d")] += 1
|
||||
|
||||
day_names = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
|
||||
day_breakdown = [{"day": day_names[i], "count": day_counts.get(i, 0)} for i in range(7)]
|
||||
hour_breakdown = [{"hour": i, "count": hour_counts.get(i, 0)} for i in range(24)]
|
||||
|
||||
max_streak = 0
|
||||
if daily_counts:
|
||||
dates = [datetime.strptime(d, "%Y-%m-%d") for d in sorted(daily_counts)]
|
||||
@@ -524,15 +421,8 @@ class InsightsEngine:
|
||||
for prev, cur in zip(dates, dates[1:]):
|
||||
current_streak = current_streak + 1 if (cur - prev).days == 1 else 1
|
||||
max_streak = max(max_streak, current_streak)
|
||||
|
||||
return {
|
||||
"by_day": day_breakdown,
|
||||
"by_hour": hour_breakdown,
|
||||
"busiest_day": max(day_breakdown, key=lambda x: x["count"]),
|
||||
"busiest_hour": max(hour_breakdown, key=lambda x: x["count"]),
|
||||
"active_days": len(daily_counts),
|
||||
"max_streak": max_streak,
|
||||
}
|
||||
return {"by_day": day_breakdown, "by_hour": hour_breakdown, "busiest_day": max(day_breakdown, key=lambda x: x["count"]),
|
||||
"busiest_hour": max(hour_breakdown, key=lambda x: x["count"]), "active_days": len(daily_counts), "max_streak": max_streak}
|
||||
|
||||
_TOP_METRICS = (
|
||||
("Most messages", lambda s: s.get("message_count") or 0, "{} msgs"),
|
||||
@@ -546,11 +436,8 @@ class InsightsEngine:
|
||||
timed = [s for s in sessions if s.get("started_at") and s.get("ended_at")]
|
||||
if timed:
|
||||
longest = max(timed, key=lambda s: s["ended_at"] - s["started_at"])
|
||||
top.append({
|
||||
"label": "Longest session", "session_id": longest["id"][:16],
|
||||
"value": format_duration_compact(longest["ended_at"] - longest["started_at"]),
|
||||
"date": _day(longest["started_at"]),
|
||||
})
|
||||
top.append({"label": "Longest session", "session_id": longest["id"][:16],
|
||||
"value": format_duration_compact(longest["ended_at"] - longest["started_at"]), "date": _day(longest["started_at"])})
|
||||
for label, metric, fmt in self._TOP_METRICS:
|
||||
best = max(sessions, key=metric)
|
||||
value = metric(best)
|
||||
@@ -568,108 +455,69 @@ class InsightsEngine:
|
||||
def _cost_lines(o: Dict, templates: tuple) -> List[str]:
|
||||
"""One formatted line per non-zero cost bucket (estimated, included, unknown)."""
|
||||
est_cost = o.get("estimated_cost", 0.0)
|
||||
values = (
|
||||
_fmt_est_cost(est_cost) if est_cost > 0 else "",
|
||||
o.get("included_cost_sessions", 0), o.get("unknown_cost_sessions", 0),
|
||||
)
|
||||
values = (_fmt_est_cost(est_cost) if est_cost > 0 else "", o.get("included_cost_sessions", 0), o.get("unknown_cost_sessions", 0))
|
||||
return [tpl.format(v) for tpl, v in zip(templates, values) if v]
|
||||
|
||||
def format_terminal(self, report: Dict) -> str:
|
||||
"""Format the insights report for terminal display (CLI)."""
|
||||
if report.get("empty"):
|
||||
days = report.get("days", 30)
|
||||
src = f" (source: {report['source_filter']})" if report.get("source_filter") else ""
|
||||
return f" No sessions found in the last {days} days{src}."
|
||||
|
||||
return f" No sessions found in the last {report.get('days', 30)} days{src}."
|
||||
o = report["overview"]
|
||||
period_label = f"Last {report['days']} days"
|
||||
if report.get("source_filter"):
|
||||
period_label += f" ({report['source_filter']})"
|
||||
padding = 58 - len(period_label) - 2
|
||||
left_pad = padding // 2
|
||||
lines = [
|
||||
"",
|
||||
" ╔══════════════════════════════════════════════════════════╗",
|
||||
" ║ 📊 Hermes Insights ║",
|
||||
f" ║{' ' * left_pad} {period_label} {' ' * (padding - left_pad)}║",
|
||||
" ╚══════════════════════════════════════════════════════════╝",
|
||||
"",
|
||||
]
|
||||
|
||||
lines = ["", " ╔══════════════════════════════════════════════════════════╗", " ║ 📊 Hermes Insights ║",
|
||||
f" ║{' ' * left_pad} {period_label} {' ' * (padding - left_pad)}║", " ╚══════════════════════════════════════════════════════════╝", ""]
|
||||
if o.get("date_range_start") and o.get("date_range_end"):
|
||||
start_str = datetime.fromtimestamp(o["date_range_start"]).strftime("%b %d, %Y")
|
||||
end_str = datetime.fromtimestamp(o["date_range_end"]).strftime("%b %d, %Y")
|
||||
lines += [f" Period: {start_str} — {end_str}", ""]
|
||||
|
||||
lines += self._section("📋 Overview")
|
||||
lines.append(f" Sessions: {o['total_sessions']:<12} Messages: {o['total_messages']:,}")
|
||||
lines.append(f" Tool calls: {o['total_tool_calls']:<12,} User messages: {o['user_messages']:,}")
|
||||
lines.append(f" Input tokens: {o['total_input_tokens']:<12,} Output tokens: {o['total_output_tokens']:,}")
|
||||
lines.append(f" Total tokens: {o['total_tokens']:,}")
|
||||
lines += self._section("📋 Overview") + [
|
||||
f" Sessions: {o['total_sessions']:<12} Messages: {o['total_messages']:,}",
|
||||
f" Tool calls: {o['total_tool_calls']:<12,} User messages: {o['user_messages']:,}",
|
||||
f" Input tokens: {o['total_input_tokens']:<12,} Output tokens: {o['total_output_tokens']:,}",
|
||||
f" Total tokens: {o['total_tokens']:,}",
|
||||
]
|
||||
if o["total_hours"] > 0:
|
||||
lines.append(f" Active time: ~{format_duration_compact(o['total_hours'] * 3600):<11} Avg session: ~{format_duration_compact(o['avg_session_duration'])}")
|
||||
lines += [f" Avg msgs/session: {o['avg_messages_per_session']:.1f}", ""]
|
||||
|
||||
# Cost buckets: show included/unknown sessions instead of collapsing to $0.
|
||||
cost_lines = self._cost_lines(o, (
|
||||
" Estimated: {}", " Included: {} session(s) (subscription — no provider invoice)",
|
||||
" Unknown: {} session(s) (no pricing data)",
|
||||
))
|
||||
cost_lines = self._cost_lines(o, (" Estimated: {}", " Included: {} session(s) (subscription — no provider invoice)",
|
||||
" Unknown: {} session(s) (no pricing data)"))
|
||||
if cost_lines:
|
||||
lines += self._section("💰 Cost") + cost_lines + [""]
|
||||
|
||||
if report["models"]:
|
||||
lines += self._section("🤖 Models Used")
|
||||
lines.append(f" {'Model':<30} {'Sessions':>8} {'Tokens':>12}")
|
||||
for m in report["models"]:
|
||||
lines.append(f" {m['model'][:28]:<30} {m['sessions']:>8} {m['total_tokens']:>12,}")
|
||||
lines.append("")
|
||||
|
||||
lines += self._section("🤖 Models Used") + [f" {'Model':<30} {'Sessions':>8} {'Tokens':>12}"]
|
||||
lines += [f" {m['model'][:28]:<30} {m['sessions']:>8} {m['total_tokens']:>12,}" for m in report["models"]] + [""]
|
||||
platforms = report["platforms"]
|
||||
if len(platforms) > 1 or (platforms and platforms[0]["platform"] != "cli"):
|
||||
lines += self._section("📱 Platforms")
|
||||
lines.append(f" {'Platform':<14} {'Sessions':>8} {'Messages':>10} {'Tokens':>14}")
|
||||
for p in platforms:
|
||||
lines.append(f" {p['platform']:<14} {p['sessions']:>8} {p['messages']:>10,} {p['total_tokens']:>14,}")
|
||||
lines.append("")
|
||||
|
||||
lines += self._section("📱 Platforms") + [f" {'Platform':<14} {'Sessions':>8} {'Messages':>10} {'Tokens':>14}"]
|
||||
lines += [f" {p['platform']:<14} {p['sessions']:>8} {p['messages']:>10,} {p['total_tokens']:>14,}" for p in platforms] + [""]
|
||||
if report["tools"]:
|
||||
lines += self._section("🔧 Top Tools")
|
||||
lines.append(f" {'Tool':<28} {'Calls':>8} {'%':>8}")
|
||||
for t in report["tools"][:15]:
|
||||
lines.append(f" {t['tool']:<28} {t['count']:>8,} {t['percentage']:>7.1f}%")
|
||||
lines += self._section("🔧 Top Tools") + [f" {'Tool':<28} {'Calls':>8} {'%':>8}"]
|
||||
lines += [f" {t['tool']:<28} {t['count']:>8,} {t['percentage']:>7.1f}%" for t in report["tools"][:15]]
|
||||
if len(report["tools"]) > 15:
|
||||
lines.append(f" ... and {len(report['tools']) - 15} more tools")
|
||||
lines.append("")
|
||||
|
||||
skills = report.get("skills", {})
|
||||
top_skills = skills.get("top_skills", [])
|
||||
if top_skills:
|
||||
lines += self._section("🧠 Top Skills")
|
||||
lines.append(f" {'Skill':<28} {'Loads':>7} {'Edits':>7} {'Last used':>11}")
|
||||
lines += self._section("🧠 Top Skills") + [f" {'Skill':<28} {'Loads':>7} {'Edits':>7} {'Last used':>11}"]
|
||||
for skill in top_skills[:10]:
|
||||
last_used = _day(skill.get("last_used_at")) if skill.get("last_used_at") else "—"
|
||||
lines.append(
|
||||
f" {skill['skill'][:28]:<28} {skill['view_count']:>7,} {skill['manage_count']:>7,} {last_used:>11}"
|
||||
)
|
||||
lines.append(f" {skill['skill'][:28]:<28} {skill['view_count']:>7,} {skill['manage_count']:>7,} {last_used:>11}")
|
||||
summary = skills.get("summary", {})
|
||||
lines.append(
|
||||
f" Distinct skills: {summary.get('distinct_skills_used', 0)} "
|
||||
f"Loads: {summary.get('total_skill_loads', 0):,} "
|
||||
f"Edits: {summary.get('total_skill_edits', 0):,}"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
lines += [f" Distinct skills: {summary.get('distinct_skills_used', 0)} Loads: {summary.get('total_skill_loads', 0):,} "
|
||||
f"Edits: {summary.get('total_skill_edits', 0):,}", ""]
|
||||
act = report.get("activity", {})
|
||||
if act.get("by_day"):
|
||||
lines += self._section("📅 Activity Patterns")
|
||||
bars = _bar_chart([d["count"] for d in act["by_day"]], max_width=15)
|
||||
for bar, d in zip(bars, act["by_day"]):
|
||||
lines.append(f" {d['day']} {bar:<15} {d['count']}")
|
||||
lines.append("")
|
||||
|
||||
busy_hours = sorted(act["by_hour"], key=lambda x: x["count"], reverse=True)
|
||||
busy_hours = [h for h in busy_hours if h["count"] > 0][:5]
|
||||
lines += [f" {d['day']} {bar:<15} {d['count']}" for bar, d in zip(bars, act["by_day"])] + [""]
|
||||
busy_hours = [h for h in sorted(act["by_hour"], key=lambda x: x["count"], reverse=True) if h["count"] > 0][:5]
|
||||
if busy_hours:
|
||||
hour_strs = [f"{_hour12(h['hour'])} ({h['count']})" for h in busy_hours]
|
||||
lines.append(f" Peak hours: {', '.join(hour_strs)}")
|
||||
@@ -678,20 +526,15 @@ class InsightsEngine:
|
||||
if act.get("max_streak") and act["max_streak"] > 1:
|
||||
lines.append(f" Best streak: {act['max_streak']} consecutive days")
|
||||
lines.append("")
|
||||
|
||||
if report.get("top_sessions"):
|
||||
lines += self._section("🏆 Notable Sessions")
|
||||
for ts in report["top_sessions"]:
|
||||
lines.append(f" {ts['label']:<20} {ts['value']:<18} ({ts['date']}, {ts['session_id']})")
|
||||
lines.append("")
|
||||
|
||||
lines += [f" {ts['label']:<20} {ts['value']:<18} ({ts['date']}, {ts['session_id']})" for ts in report["top_sessions"]] + [""]
|
||||
return "\n".join(lines)
|
||||
|
||||
def format_gateway(self, report: Dict) -> str:
|
||||
"""Format the insights report for gateway/messaging (shorter)."""
|
||||
if report.get("empty"):
|
||||
return f"No sessions found in the last {report.get('days', 30)} days."
|
||||
|
||||
o = report["overview"]
|
||||
lines = [
|
||||
f"📊 **Hermes Insights** — Last {report['days']} days\n",
|
||||
@@ -701,39 +544,22 @@ class InsightsEngine:
|
||||
if o["total_hours"] > 0:
|
||||
lines.append(f"**Active time:** ~{format_duration_compact(o['total_hours'] * 3600)} | **Avg session:** ~{format_duration_compact(o['avg_session_duration'])}")
|
||||
lines.append("")
|
||||
|
||||
cost_parts = self._cost_lines(o, ("{} estimated", "{} included (subscription)", "{} unknown"))
|
||||
if cost_parts:
|
||||
lines += [f"**Cost:** {' | '.join(cost_parts)}", ""]
|
||||
|
||||
if report["models"]:
|
||||
lines.append("**🤖 Models:**")
|
||||
for m in report["models"][:5]:
|
||||
lines.append(f" {m['model'][:25]} — {m['sessions']} sessions, {m['total_tokens']:,} tokens")
|
||||
lines.append("")
|
||||
|
||||
lines += ["**🤖 Models:**"] + [f" {m['model'][:25]} — {m['sessions']} sessions, {m['total_tokens']:,} tokens" for m in report["models"][:5]] + [""]
|
||||
if len(report["platforms"]) > 1:
|
||||
lines.append("**📱 Platforms:**")
|
||||
for p in report["platforms"]:
|
||||
lines.append(f" {p['platform']} — {p['sessions']} sessions, {p['messages']:,} msgs")
|
||||
lines.append("")
|
||||
|
||||
lines += ["**📱 Platforms:**"] + [f" {p['platform']} — {p['sessions']} sessions, {p['messages']:,} msgs" for p in report["platforms"]] + [""]
|
||||
if report["tools"]:
|
||||
lines.append("**🔧 Top Tools:**")
|
||||
for t in report["tools"][:8]:
|
||||
lines.append(f" {t['tool']} — {t['count']:,} calls ({t['percentage']:.1f}%)")
|
||||
lines.append("")
|
||||
|
||||
lines += ["**🔧 Top Tools:**"] + [f" {t['tool']} — {t['count']:,} calls ({t['percentage']:.1f}%)" for t in report["tools"][:8]] + [""]
|
||||
skills = report.get("skills", {})
|
||||
if skills.get("top_skills"):
|
||||
lines.append("**🧠 Top Skills:**")
|
||||
for skill in skills["top_skills"][:5]:
|
||||
suffix = f", last used {_day(skill['last_used_at'])}" if skill.get("last_used_at") else ""
|
||||
lines.append(
|
||||
f" {skill['skill']} — {skill['view_count']:,} loads, {skill['manage_count']:,} edits{suffix}"
|
||||
)
|
||||
lines.append(f" {skill['skill']} — {skill['view_count']:,} loads, {skill['manage_count']:,} edits{suffix}")
|
||||
lines.append("")
|
||||
|
||||
act = report.get("activity", {})
|
||||
if act.get("busiest_day") and act.get("busiest_hour"):
|
||||
lines.append(f"**📅 Busiest:** {act['busiest_day']['day']}s ({act['busiest_day']['count']} sessions), {_hour12(act['busiest_hour']['hour'])} ({act['busiest_hour']['count']} sessions)")
|
||||
@@ -741,5 +567,4 @@ class InsightsEngine:
|
||||
lines.append(f"**Active days:** {act['active_days']}")
|
||||
if act.get("max_streak", 0) > 1:
|
||||
lines.append(f"**Best streak:** {act['max_streak']} consecutive days")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
Reference in New Issue
Block a user