fix: show lineage representative in discoverability audit

This commit is contained in:
ai-ag2026
2026-05-28 08:13:35 +02:00
parent 5528e2c579
commit bd8fd22d81
3 changed files with 65 additions and 2 deletions

View File

@@ -3,6 +3,9 @@
## [Unreleased]
### Changed
- Session discoverability audit findings for stale persisted WebUI-as-CLI flags now report whether an API-visible lineage representative already covers the hidden snapshot, including the representative session id in JSON and Markdown output.
## [v0.51.152] — 2026-05-28 — Release DX (stage-batch34 — single-PR optional gateway-backed browser chat)
### Added

View File

@@ -252,15 +252,19 @@ def audit_session_discoverability(
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:
api_lineage_ids.add(str(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] = []
@@ -282,6 +286,12 @@ def audit_session_discoverability(
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,
@@ -295,6 +305,7 @@ def audit_session_discoverability(
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(
@@ -309,12 +320,12 @@ def audit_session_discoverability(
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
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
@@ -407,6 +418,10 @@ def render_discoverability_markdown(report: dict) -> str:
lines.append(
" - present_in: " + ", ".join(f"{k}={v}" for k, v in sorted(present.items()))
)
if item.get("represented_by_api_lineage"):
lines.append(
f" - represented_by_api_lineage: true via `{item.get('api_representative_session_id')}`"
)
lines.append("")
return "\n".join(lines)

View File

@@ -98,6 +98,51 @@ def test_audit_reports_stale_cli_flag_on_webui_session(tmp_path):
assert item["sidecar_is_cli_session"] is True
def test_stale_flag_hidden_snapshot_reports_visible_api_representative(tmp_path):
root = "webui-hidden-stale-cli-root"
tip = "webui-visible-lineage-tip"
_write_sidecar(
tmp_path,
root,
messages=12,
source_tag="webui",
session_source="webui",
is_cli_session=True,
pre_compression_snapshot=True,
)
_write_index(
tmp_path,
{
"session_id": root,
"message_count": 12,
"source_tag": "webui",
"session_source": "webui",
"is_cli_session": True,
"pre_compression_snapshot": True,
},
)
db = _state_db(
tmp_path,
[
{"id": root, "source": "webui", "message_count": 12},
{"id": tip, "source": "webui", "parent_session_id": root, "message_count": 14},
],
{root: 12, tip: 14},
)
report = audit_session_discoverability(
tmp_path,
state_db_path=db,
api_sessions=[{"session_id": tip, "message_count": 14, "parent_session_id": root, "_lineage_root_id": root}],
)
item = next(item for item in report["items"] if item["session_id"] == root)
assert item["kind"] == "persisted_source_flag_stale"
assert item["present_in"] == {"sidecar": True, "index": True, "state_db": True, "api": False}
assert item["represented_by_api_lineage"] is True
assert item["api_representative_session_id"] == tip
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})