David/HERMES-GIELIE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add session discoverability audit

Browse source
This commit is contained in:
ai-ag2026 2026-05-28 00:13:20 +02:00
parent 329debcd33
commit 0af58f2b82
2 changed files with 646 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

433
api/session_discoverability.py Normal file
View file

@ -0,0 +1,433 @@
"""Read-only sidebar discoverability audit for Hermes WebUI sessions.
This module does not repair or mutate session state. It cross-checks the four
places that decide whether a session can be found from the WebUI sidebar:
- JSON sidecars under the WebUI session directory
- ``_index.json`` sidebar metadata
- canonical ``state.db`` rows/messages
- the live ``api.models.all_sessions()`` sidebar response, when available
"""
from __future__ import annotations
import argparse
import json
import sqlite3
from collections import Counter
from pathlib import Path
from typing import Iterable
def _safe_int(value, default: int = 0) -> int:
try:
if value is None:
return default
return int(value)
except (TypeError, ValueError):
return default
def _read_json(path: Path):
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError, ValueError):
return None
def _message_count_from_payload(payload: dict) -> int:
messages = payload.get("messages")
if isinstance(messages, list):
return len(messages)
return _safe_int(payload.get("message_count"), 0)
def _record_from_mapping(mapping: dict, source_name: str) -> dict:
sid = str(mapping.get("session_id") or mapping.get("id") or "").strip()
if not sid:
return {}
return {
"session_id": sid,
"title": mapping.get("title"),
"message_count": _message_count_from_payload(mapping),
"source_tag": mapping.get("source_tag"),
"session_source": mapping.get("session_source"),
"source": mapping.get("source"),
"is_cli_session": mapping.get("is_cli_session"),
"parent_session_id": mapping.get("parent_session_id"),
"pre_compression_snapshot": bool(mapping.get("pre_compression_snapshot")),
"_lineage_root_id": mapping.get("_lineage_root_id"),
"archived": bool(mapping.get("archived")),
"project_id": mapping.get("project_id"),
"workspace": mapping.get("workspace"),
"_source_name": source_name,
}
def _read_sidecars(session_dir: Path) -> dict[str, dict]:
records: dict[str, dict] = {}
if not session_dir.exists():
return records
for path in sorted(p for p in session_dir.glob("*.json") if not p.name.startswith("_")):
payload = _read_json(path)
if not isinstance(payload, dict):
continue
record = _record_from_mapping(payload, "sidecar")
if record:
records[record["session_id"]] = record
return records
def _read_index(session_dir: Path) -> dict[str, dict]:
payload = _read_json(session_dir / "_index.json")
records: dict[str, dict] = {}
if not isinstance(payload, list):
return records
for entry in payload:
if not isinstance(entry, dict):
continue
record = _record_from_mapping(entry, "index")
if record:
records[record["session_id"]] = record
return records
def _optional_expr(name: str, columns: set[str], fallback: str = "NULL") -> str:
return name if name in columns else f"{fallback} AS {name}"
def _read_state_db(state_db_path: Path | None) -> dict[str, dict]:
if state_db_path is None or not state_db_path.exists():
return {}
try:
with sqlite3.connect(f"file:{state_db_path}?mode=ro", uri=True) as conn:
conn.row_factory = sqlite3.Row
tables = {row[0] for row in conn.execute("select name from sqlite_master where type='table'")}
if "sessions" not in tables:
return {}
session_cols = {row[1] for row in conn.execute("pragma table_info(sessions)")}
if "id" not in session_cols:
return {}
message_cols: set[str] = set()
if "messages" in tables:
message_cols = {row[1] for row in conn.execute("pragma table_info(messages)")}
title_expr = _optional_expr("title", session_cols)
source_expr = _optional_expr("source", session_cols)
parent_expr = _optional_expr("parent_session_id", session_cols)
msg_expr = _optional_expr("message_count", session_cols, "0")
workspace_expr = _optional_expr("workspace", session_cols)
rows = conn.execute(
f"""
SELECT id, {title_expr}, {source_expr}, {parent_expr}, {msg_expr}, {workspace_expr}
FROM sessions
"""
).fetchall()
message_counts: dict[str, int] = {}
if {"session_id"}.issubset(message_cols):
for row in conn.execute("SELECT session_id, COUNT(*) AS count FROM messages GROUP BY session_id"):
message_counts[str(row["session_id"])] = _safe_int(row["count"], 0)
records: dict[str, dict] = {}
for row in rows:
sid = str(row["id"] or "").strip()
if not sid:
continue
count = message_counts.get(sid, _safe_int(row["message_count"], 0))
records[sid] = {
"session_id": sid,
"title": row["title"],
"message_count": count,
"source": row["source"],
"source_tag": row["source"],
"session_source": row["source"],
"parent_session_id": row["parent_session_id"],
"workspace": row["workspace"],
"_source_name": "state_db",
}
return records
except Exception:
return {}
def _normalize_api_sessions(api_sessions: Iterable[dict] | None) -> dict[str, dict]:
records: dict[str, dict] = {}
if api_sessions is None:
try:
from api.models import all_sessions
api_sessions = all_sessions()
except Exception:
api_sessions = []
for entry in api_sessions or []:
if not isinstance(entry, dict):
continue
record = _record_from_mapping(entry, "api")
if record:
records[record["session_id"]] = record
return records
def _merged_field(sid: str, stores: list[dict[str, dict]], field: str):
for store in stores:
value = store.get(sid, {}).get(field)
if value not in (None, ""):
return value
return None
def _max_message_count(sid: str, stores: list[dict[str, dict]]) -> int:
return max((_safe_int(store.get(sid, {}).get("message_count"), 0) for store in stores), default=0)
def _lineage_root(sid: str, parent_by_id: dict[str, str | None]) -> str:
seen: set[str] = set()
current = sid
while current and current not in seen:
seen.add(current)
parent = parent_by_id.get(current)
if not parent:
return current
current = parent
return sid
def _webui_origin(*records: dict) -> bool:
values: list[str] = []
for record in records:
for key in ("source", "source_tag", "session_source"):
value = record.get(key)
if value is not None:
values.append(str(value).strip().lower())
return "webui" in values
def _computed_is_cli_session(row: dict) -> bool:
sources = {
str(row.get(key) or "").strip().lower()
for key in ("session_source", "source_tag", "raw_source", "source", "source_label")
}
if "webui" in sources:
return False
try:
from api.agent_sessions import is_cli_session_row
return is_cli_session_row(row)
except Exception:
source = str(row.get("session_source") or row.get("source_tag") or row.get("raw_source") or row.get("source") or "").strip().lower()
return source == "cli"
def _new_item(session_id: str, kind: str, category: str, recommendation: str, **extra) -> dict:
item = {
"session_id": session_id,
"kind": kind,
"category": category,
"recommendation": recommendation,
}
item.update(extra)
return item
def audit_session_discoverability(
session_dir: Path,
state_db_path: Path | None = None,
*,
api_sessions: Iterable[dict] | None = None,
) -> dict:
"""Return a read-only cross-store discoverability report.
The audit is intentionally diagnostic only. It reports cases where
messageful sessions have no visible API/sidebar representative, stale source
flags can put WebUI sessions into the CLI tab, and index/sidecar/state-db
drift can make a session harder to resolve.
"""
session_dir = Path(session_dir)
sidecars = _read_sidecars(session_dir)
index = _read_index(session_dir)
state = _read_state_db(state_db_path)
api = _normalize_api_sessions(api_sessions)
stores = [sidecars, index, state, api]
all_ids = set().union(*(store.keys() for store in stores))
parent_by_id: dict[str, str | None] = {}
for sid in all_ids:
parent = _merged_field(sid, stores, "parent_session_id")
parent_by_id[sid] = str(parent) if parent else None
api_lineage_ids: set[str] = set()
for sid, row in api.items():
explicit_root = row.get("_lineage_root_id")
if explicit_root:
api_lineage_ids.add(str(explicit_root))
current = sid
seen: set[str] = set()
while current and current not in seen:
seen.add(current)
api_lineage_ids.add(current)
current = parent_by_id.get(current) or ""
items: list[dict] = []
for sid in sorted(all_ids):
message_count = _max_message_count(sid, stores)
present_in = {
"sidecar": sid in sidecars,
"index": sid in index,
"state_db": sid in state,
"api": sid in api,
}
sidecar = sidecars.get(sid, {})
index_row = index.get(sid, {})
state_row = state.get(sid, {})
api_row = api.get(sid, {})
webui_origin = _webui_origin(sidecar, index_row, state_row, api_row)
api_is_cli = api_row.get("is_cli_session") is True
api_computed_is_cli = _computed_is_cli_session(api_row) if api_row else False
index_is_cli = index_row.get("is_cli_session") is True
sidecar_is_cli = sidecar.get("is_cli_session") is True
if webui_origin and api_is_cli and api_computed_is_cli:
items.append(_new_item(
sid,
"source_misclassified",
"warning",
"normalize_api_source_flags",
message_count=message_count,
state_source=state_row.get("source"),
api_is_cli_session=api_row.get("is_cli_session"),
api_computed_is_cli_session=api_computed_is_cli,
index_is_cli_session=index_row.get("is_cli_session"),
sidecar_is_cli_session=sidecar.get("is_cli_session"),
present_in=present_in,
))
elif webui_origin and (api_is_cli or index_is_cli or sidecar_is_cli):
items.append(_new_item(
sid,
"persisted_source_flag_stale",
"warning",
"rewrite_persisted_sidebar_source_flags_or_ignore_route_normalizes",
message_count=message_count,
state_source=state_row.get("source"),
api_is_cli_session=api_row.get("is_cli_session"),
api_computed_is_cli_session=api_computed_is_cli,
index_is_cli_session=index_row.get("is_cli_session"),
sidecar_is_cli_session=sidecar.get("is_cli_session"),
present_in=present_in,
))
if message_count <= 0 or sid in api:
continue
lineage_root = _lineage_root(sid, parent_by_id)
is_hidden_snapshot = bool(sidecar.get("pre_compression_snapshot") or index_row.get("pre_compression_snapshot"))
if sid in api_lineage_ids or lineage_root in api_lineage_ids:
continue
if is_hidden_snapshot:
items.append(_new_item(
sid,
"lineage_missing_visible_representative",
"warning",
"repair_lineage_or_expose_tip",
message_count=message_count,
lineage_root=lineage_root,
present_in=present_in,
))
continue
if not present_in["sidecar"] and not present_in["index"] and present_in["state_db"]:
items.append(_new_item(
sid,
"state_db_messageful_missing_sidecar",
"warning",
"materialize_sidecar_or_archive_state_row",
message_count=message_count,
lineage_root=lineage_root,
present_in=present_in,
))
continue
items.append(_new_item(
sid,
"api_missing_messageful",
"warning",
"investigate_sidebar_filters_or_api_merge",
message_count=message_count,
lineage_root=lineage_root,
present_in=present_in,
))
summary = {
"sessions_seen": len(all_ids),
"messageful": sum(1 for sid in all_ids if _max_message_count(sid, stores) > 0),
"visible_api": len(api),
"warnings": sum(1 for item in items if item.get("category") == "warning"),
}
status = "warn" if summary["warnings"] else "ok"
return {
"status": status,
"summary": summary,
"stores": {
"sidecar": len(sidecars),
"index": len(index),
"state_db": len(state),
"api": len(api),
},
"items": items,
}
def render_discoverability_markdown(report: dict) -> str:
lines = [
"# WebUI Session Discoverability Audit",
"",
f"Status: `{report.get('status')}`",
"",
"## Summary",
"",
]
for key, value in (report.get("summary") or {}).items():
lines.append(f"- `{key}`: {value}")
lines.extend(["", "## Stores", ""])
for key, value in (report.get("stores") or {}).items():
lines.append(f"- `{key}`: {value}")
lines.extend(["", "## Findings", ""])
items = report.get("items") or []
if items:
lines.extend(["### By kind", ""])
for kind, count in sorted(Counter(str(item.get("kind")) for item in items).items()):
lines.append(f"- `{kind}`: {count}")
lines.extend(["", "### Details", ""])
if not items:
lines.append("No discoverability findings.")
else:
for item in items:
lines.append(
f"- `{item.get('kind')}` `{item.get('session_id')}` "
f"messages={item.get('message_count', 'n/a')} recommendation=`{item.get('recommendation')}`"
)
present = item.get("present_in")
if isinstance(present, dict):
lines.append(
" - present_in: " + ", ".join(f"{k}={v}" for k, v in sorted(present.items()))
)
lines.append("")
return "\n".join(lines)
def _main() -> int:
parser = argparse.ArgumentParser(description="Read-only Hermes WebUI session discoverability audit")
parser.add_argument("--session-dir", type=Path, required=True)
parser.add_argument("--state-db", type=Path, default=None)
parser.add_argument("--format", choices=("json", "markdown"), default="json")
parser.add_argument("--out", type=Path, default=None)
args = parser.parse_args()
report = audit_session_discoverability(args.session_dir, state_db_path=args.state_db)
text = render_discoverability_markdown(report) if args.format == "markdown" else json.dumps(report, sort_keys=True)
if args.out:
args.out.parent.mkdir(parents=True, exist_ok=True)
args.out.write_text(text, encoding="utf-8")
else:
print(text)
return 0
if __name__ == "__main__":
raise SystemExit(_main())

213
tests/test_session_discoverability_audit.py Normal file
View file

@ -0,0 +1,213 @@
import json
import sqlite3
from pathlib import Path
from api.session_discoverability import audit_session_discoverability
def _write_sidecar(session_dir: Path, sid: str, *, messages=1, **metadata):
payload = {
"session_id": sid,
"id": sid,
"title": metadata.pop("title", sid),
"messages": [{"role": "user", "content": f"message {i}"} for i in range(messages)],
**metadata,
}
path = session_dir / f"{sid}.json"
path.write_text(json.dumps(payload), encoding="utf-8")
return path
def _write_index(session_dir: Path, *entries):
(session_dir / "_index.json").write_text(json.dumps(list(entries)), encoding="utf-8")
def _state_db(session_dir: Path, rows, message_counts=None):
db = session_dir / "state.db"
message_counts = message_counts or {}
with sqlite3.connect(db) as conn:
conn.execute(
"""
create table sessions (
id text primary key,
source text,
title text,
parent_session_id text,
message_count integer
)
"""
)
conn.execute("create table messages (session_id text, role text, content text)")
for row in rows:
conn.execute(
"insert into sessions (id, source, title, parent_session_id, message_count) values (?, ?, ?, ?, ?)",
(
row["id"],
row.get("source"),
row.get("title"),
row.get("parent_session_id"),
row.get("message_count", message_counts.get(row["id"], 0)),
),
)
for i in range(message_counts.get(row["id"], 0)):
conn.execute(
"insert into messages (session_id, role, content) values (?, 'user', ?)",
(row["id"], f"message {i}"),
)
return db
def test_audit_reports_messageful_sidecar_missing_from_api_visible_sessions(tmp_path):
sid = "20260526_234828_b8393a"
_write_sidecar(tmp_path, sid, messages=12, source_tag="webui", session_source="webui")
_write_index(tmp_path, {"session_id": sid, "message_count": 12, "source_tag": "webui"})
db = _state_db(tmp_path, [{"id": sid, "source": "webui", "message_count": 12}], {sid: 12})
report = audit_session_discoverability(tmp_path, state_db_path=db, api_sessions=[])
assert report["status"] == "warn"
item = report["items"][0]
assert item["kind"] == "api_missing_messageful"
assert item["category"] == "warning"
assert item["session_id"] == sid
assert item["message_count"] == 12
assert item["present_in"] == {"sidecar": True, "index": True, "state_db": True, "api": False}
def test_audit_reports_stale_cli_flag_on_webui_session(tmp_path):
sid = "webui-marked-cli"
_write_sidecar(tmp_path, sid, messages=3, source_tag="webui", session_source="webui", is_cli_session=True)
_write_index(tmp_path, {"session_id": sid, "message_count": 3, "source_tag": "webui", "is_cli_session": True})
db = _state_db(tmp_path, [{"id": sid, "source": "webui", "message_count": 3}], {sid: 3})
report = audit_session_discoverability(
tmp_path,
state_db_path=db,
api_sessions=[{"session_id": sid, "title": "WebUI Project Work", "message_count": 3, "source_tag": "webui", "session_source": "webui", "is_cli_session": True}],
)
kinds = {item["kind"] for item in report["items"]}
assert "persisted_source_flag_stale" in kinds
assert "source_misclassified" not in kinds
item = next(item for item in report["items"] if item["kind"] == "persisted_source_flag_stale")
assert item["session_id"] == sid
assert item["state_source"] == "webui"
assert item["api_is_cli_session"] is True
assert item["api_computed_is_cli_session"] is False
assert item["index_is_cli_session"] is True
assert item["sidecar_is_cli_session"] is True
def test_audit_classifies_state_db_only_messageful_rows_as_missing_sidecar(tmp_path):
sid = "state-only-session"
db = _state_db(tmp_path, [{"id": sid, "source": "webui", "message_count": 5}], {sid: 5})
report = audit_session_discoverability(tmp_path, state_db_path=db, api_sessions=[])
assert report["items"][0]["kind"] == "state_db_messageful_missing_sidecar"
assert report["items"][0]["recommendation"] == "materialize_sidecar_or_archive_state_row"
assert report["items"][0]["present_in"] == {"sidecar": False, "index": False, "state_db": True, "api": False}
def test_audit_treats_hidden_snapshot_as_represented_when_visible_lineage_tip_exists(tmp_path):
root = "root-session"
child = "child-session"
_write_sidecar(
tmp_path,
root,
messages=8,
source_tag="webui",
session_source="webui",
pre_compression_snapshot=True,
)
_write_sidecar(
tmp_path,
child,
messages=2,
source_tag="webui",
session_source="webui",
parent_session_id=root,
)
_write_index(tmp_path, {"session_id": root, "message_count": 8}, {"session_id": child, "message_count": 2})
db = _state_db(
tmp_path,
[
{"id": root, "source": "webui", "message_count": 8},
{"id": child, "source": "webui", "parent_session_id": root, "message_count": 2},
],
{root: 8, child: 2},
)
report = audit_session_discoverability(
tmp_path,
state_db_path=db,
api_sessions=[{"session_id": child, "message_count": 2, "parent_session_id": root}],
)
assert report["status"] == "ok"
assert report["items"] == []
def test_audit_treats_siblings_as_represented_when_visible_tip_points_at_lineage_root(tmp_path):
root = "root-session"
hidden_sibling = "hidden-sibling"
visible_tip = "visible-tip"
_write_sidecar(
tmp_path,
root,
messages=8,
source_tag="webui",
session_source="webui",
pre_compression_snapshot=True,
)
_write_sidecar(
tmp_path,
hidden_sibling,
messages=5,
source_tag="webui",
session_source="webui",
parent_session_id=root,
)
_write_index(
tmp_path,
{"session_id": root, "message_count": 8, "pre_compression_snapshot": True},
{"session_id": hidden_sibling, "message_count": 5, "parent_session_id": root},
)
db = _state_db(
tmp_path,
[
{"id": root, "source": "webui", "message_count": 8},
{"id": hidden_sibling, "source": "webui", "parent_session_id": root, "message_count": 5},
{"id": visible_tip, "source": "webui", "parent_session_id": root, "message_count": 9},
],
{root: 8, hidden_sibling: 5, visible_tip: 9},
)
report = audit_session_discoverability(
tmp_path,
state_db_path=db,
api_sessions=[{"session_id": visible_tip, "message_count": 9, "_lineage_root_id": root}],
)
assert report["status"] == "ok"
assert report["items"] == []
def test_audit_reports_lineage_without_visible_representative(tmp_path):
root = "root-session"
_write_sidecar(
tmp_path,
root,
messages=8,
source_tag="webui",
session_source="webui",
pre_compression_snapshot=True,
)
_write_index(tmp_path, {"session_id": root, "message_count": 8})
db = _state_db(tmp_path, [{"id": root, "source": "webui", "message_count": 8}], {root: 8})
report = audit_session_discoverability(tmp_path, state_db_path=db, api_sessions=[])
assert report["status"] == "warn"
assert [item["kind"] for item in report["items"]] == ["lineage_missing_visible_representative"]
assert report["items"][0]["session_id"] == root