"""Read-only sidebar discoverability audit for Hermes WebUI sessions. This module does not repair or mutate session state. It cross-checks the fourplaces 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 argparseimport jsonimport osimport threadingimport shutilimport sqlite3from collections import Counterfrom pathlib import Pathfrom 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()    api_lineage_representative_by_id: dict[str, str] = {}    for sid, row in api.items():        explicit_root = row.get("_lineage_root_id")        if explicit_root:            root_id = str(explicit_root)            api_lineage_ids.add(root_id)            api_lineage_representative_by_id.setdefault(root_id, sid)        current = sid        seen: set[str] = set()        while current and current not in seen:            seen.add(current)            api_lineage_ids.add(current)            api_lineage_representative_by_id.setdefault(current, sid)            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        lineage_root = _lineage_root(sid, parent_by_id)        api_representative = api_lineage_representative_by_id.get(sid) or api_lineage_representative_by_id.get(lineage_root)        api_lineage_extra = {            "represented_by_api_lineage": bool(api_representative),            "api_representative_session_id": api_representative,        }        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,                **api_lineage_extra,            ))        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,                **api_lineage_extra,            ))         if message_count <= 0 or sid in api:            continue         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 _atomic_write_json(path: Path, payload) -> None:    # Temp name includes the thread id, not just the pid: WebUI runs a    # ThreadingHTTPServer, so two request threads reconciling the same _index.json    # would otherwise collide on one `<name>.tmp.<pid>` path and truncate each    # other's temp before its os.replace. fsync before the rename so a power loss    # can't leave a zero-length/garbage file where the durable JSON should be; a    # finally drops the temp if anything fails. Mode matches the prior write_text    # (umask default) since the plain open() respects the umask.    text = json.dumps(payload, ensure_ascii=False, indent=2)    tmp = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp")    try:        with open(tmp, "w", encoding="utf-8") as f:            f.write(text)            f.flush()            os.fsync(f.fileno())        os.replace(tmp, path)    except BaseException:        try:            os.unlink(tmp)        except OSError:            pass        raise  def _backup_file(path: Path, backup_dir: Path, backed_up: dict[Path, str]) -> str | None:    if not path.exists():        return None    resolved = path.resolve()    if resolved in backed_up:        return backed_up[resolved]    backup_dir.mkdir(parents=True, exist_ok=True)    target = backup_dir / path.name    if target.exists():        stem = target.name        i = 1        while (backup_dir / f"{stem}.{i}").exists():            i += 1        target = backup_dir / f"{stem}.{i}"    shutil.copy2(path, target)    backed_up[resolved] = str(target)    return str(target)  def _plan_discoverability_repairs(report: dict) -> list[dict]:    actions: list[dict] = []    for item in report.get("items") or []:        sid = str(item.get("session_id") or "")        if not sid:            continue        if item.get("kind") == "persisted_source_flag_stale":            if item.get("sidecar_is_cli_session") is True:                actions.append({"session_id": sid, "action": "clear_sidecar_cli_flag"})            if item.get("index_is_cli_session") is True:                actions.append({"session_id": sid, "action": "clear_index_cli_flag"})        elif item.get("kind") == "state_db_messageful_missing_sidecar":            actions.append({"session_id": sid, "action": "materialize_sidecar_from_state_db"})    return actions  def _clear_sidecar_cli_flag(session_dir: Path, sid: str, backup_dir: Path, backed_up: dict[Path, str]) -> dict:    path = session_dir / f"{sid}.json"    payload = _read_json(path)    if not isinstance(payload, dict):        return {"session_id": sid, "action": "clear_sidecar_cli_flag", "applied": False, "error": "sidecar_unreadable"}    if not _webui_origin(payload):        return {"session_id": sid, "action": "clear_sidecar_cli_flag", "applied": False, "skipped": "not_webui_origin"}    if payload.get("is_cli_session") is not True:        return {"session_id": sid, "action": "clear_sidecar_cli_flag", "applied": False, "skipped": "already_clear"}    backup = _backup_file(path, backup_dir, backed_up)    payload["is_cli_session"] = False    _atomic_write_json(path, payload)    return {"session_id": sid, "action": "clear_sidecar_cli_flag", "applied": True, "backup": backup}  def _clear_index_cli_flag(session_dir: Path, sid: str, backup_dir: Path, backed_up: dict[Path, str]) -> dict:    path = session_dir / "_index.json"    payload = _read_json(path)    if not isinstance(payload, list):        return {"session_id": sid, "action": "clear_index_cli_flag", "applied": False, "error": "index_unreadable"}    changed = False    for entry in payload:        if not isinstance(entry, dict):            continue        if str(entry.get("session_id") or "") != sid:            continue        if not _webui_origin(entry):            continue        if entry.get("is_cli_session") is True:            entry["is_cli_session"] = False            changed = True    if not changed:        return {"session_id": sid, "action": "clear_index_cli_flag", "applied": False, "skipped": "already_clear_or_missing"}    backup = _backup_file(path, backup_dir, backed_up)    _atomic_write_json(path, payload)    return {"session_id": sid, "action": "clear_index_cli_flag", "applied": True, "backup": backup}  def _materialize_sidecar_from_state_db(session_dir: Path, state_db_path: Path | None, sid: str, backup_dir: Path, backed_up: dict[Path, str]) -> dict:    if state_db_path is None:        return {"session_id": sid, "action": "materialize_sidecar_from_state_db", "applied": False, "error": "state_db_required"}    target = session_dir / f"{sid}.json"    if target.exists():        return {"session_id": sid, "action": "materialize_sidecar_from_state_db", "applied": False, "skipped": "sidecar_exists"}    try:        from api.session_recovery import _read_state_db_missing_sidecar_rows, _state_db_row_to_sidecar    except Exception as exc:        return {"session_id": sid, "action": "materialize_sidecar_from_state_db", "applied": False, "error": f"recovery_import_failed:{exc}"}    rows = {str(row.get("id") or ""): row for row in _read_state_db_missing_sidecar_rows(session_dir, state_db_path)}    row = rows.get(sid)    if not row:        return {"session_id": sid, "action": "materialize_sidecar_from_state_db", "applied": False, "skipped": "state_row_not_repairable"}

Lines 1–500 of 659. Load more