可疑实体找「最接近的名字」时重建的目录(_scope_catalog)不收分析结论:候选只看表和字段,用不到结论的内容,以前每点开一个可疑的名字都把这次运行的分析结论读盘、复验一遍哈希(build_catalog 给 A 条目装整份结论)。查询、表和字段的编号不受影响。测试:台账里有分析结论时重建目录不读任何工件,按报告节点和退回不分范围两条路都是

This commit is contained in:
ThearchyHelios
2026-10-21 15:39:43 +08:00
parent 496dc27419
commit 40918a7fb1
2 changed files with 25 additions and 2 deletions
+6 -2
View File
@@ -4048,18 +4048,22 @@ def _scope_catalog(sealed: _Sealed, report_node: str) -> dict[str, Any]:
doc.catalog 只留了正文用到的实体,给「是不是想写…」找候选不够;运行状态(checkpoint)不在封存
范围里,也不能用。图读不出来、找不到报告节点时退回全部封存台账。
台账里的分析结论不进:候选只看表和字段,用不到结论的内容,进了目录就要把每份结论读盘、复验一遍哈希(build_catalog
给 A 条目装整份结论)。查询、表和字段的编号不受影响(分析结论另编 A)。
"""
from app.engine.nodes.report import report_catalog
from app.engine.schema import GraphSpec
entries = [e for e in sealed.entries if e.get("kind") != LEDGER_KIND]
try:
spec = GraphSpec.model_validate(sealed.graph)
node = spec.node_map().get(report_node)
except Exception: # noqa: BLE001 - 老运行的图可能已经不合现在的格式,退回不分范围
node = None
if node is None:
return build_catalog(nodes={}, ledger=sealed.entries)
return report_catalog({"evidence": sealed.entries, "nodes": {}, "input": {}}, spec, node) # type: ignore[arg-type]
return build_catalog(nodes={}, ledger=entries)
return report_catalog({"evidence": entries, "nodes": {}, "input": {}}, spec, node) # type: ignore[arg-type]
def _suspicious_step(seg: dict[str, Any], cite: dict[str, Any], sealed: _Sealed, report_node: str) -> dict[str, Any]:
@@ -287,6 +287,25 @@ async def test_a_tampered_result_is_said_and_not_shown(client):
assert seal["hash_ok"] is False and seal["sealed"] is False and body["seal"]["covered"] is False
def test_suspicious_name_hints_do_not_read_the_analysis_results(monkeypatch):
"""可疑实体找「最接近的名字」时按封存的台账重建目录(_scope_catalog):分析结论不进,不读盘、不复验哈希(候选只看表和
字段,用不到结论的内容)。以前每点开一个可疑的名字,就把这次运行的分析结论全读一遍。"""
from app.api import evidence as api
from app.engine import evidence as engine_evidence
read: list[str] = []
monkeypatch.setattr(engine_evidence, "load_artifact", lambda artifact: read.append(artifact))
q, a = "1" * 64, "2" * 64
entries = [_query_entry(SNAPSHOT, q), {"kind": "analysis", "node_id": "an", "exec": 1, "artifact": a,
"op": "decompose", "mode": "additive", "item": "drop", "inputs": [q]}]
graph = {**GRAPH, "edges": [{"source": "fetch", "target": "an"}, {"source": "an", "target": "write"}]}
sealed = api._Sealed(run_id="r", graph=graph, seal={"sealed": True, "ok": True}, events=[], entries=entries)
for report in ("write", "nope"):
catalog = api._scope_catalog(sealed, report)
assert "Q1" in catalog and not [e for e in catalog.values() if e.get("kind") == "analysis"]
assert read == []
async def test_agent_analyses_are_marked_as_such(client):
made = await sealed_run(TEXT, item=None, call_id="call-9")
body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json()