mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 04:11:58 +00:00
ThearchyHelios
9d8b41430e
- 完善口径编辑、只读试运行、统一质量验证与执行语义指纹 - 增加可执行指标约束、有预算的主动调查和过程评测 - 提取发布检查、证据读取与起草服务，支持任务停止、重试和检查点恢复 - 增加版本传递影响分析，修复并发覆盖、敏感值遮罩、异常评分及页面状态问题 - 修复 SQLite 复合查询追溯和 microVM 可选依赖缺失处理 - 纳入当前设计文档及部署配置调整 验证：相关主回归 3822 项通过，4 项跳过、8 项预期失败；补充回归通过。 前端构建、文案检查及 git diff --check 通过。 全量检查仍有环境权限、PostgreSQL 可选依赖和既有部署配置、SQLite 文件基线差异导致的未通过项。
189 lines
11 KiB
Python
189 lines
11 KiB
Python
"""读取并验证封存证据的公共服务，供试运行、证据接口和评测使用。"""
|
|
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
from sqlalchemy import select
|
|
from app.core.artifact_store import load
|
|
from app.core.events import EventType
|
|
from app.db.base import SessionLocal
|
|
from app.db.models import Run, RunEvent
|
|
from app.engine.evidence import ledger_enabled
|
|
from app.engine.lineage import scoped_id as _scoped
|
|
from app.analysis.contract import LEDGER_KIND
|
|
|
|
|
|
def _fetch(artifact: str) -> tuple[Any, bool]:
|
|
try:
|
|
value = load(artifact)
|
|
return value, value is not None
|
|
except (OSError, ValueError):
|
|
return None, False
|
|
|
|
|
|
@dataclass
|
|
class Sealed:
|
|
run_id: str
|
|
graph: dict[str, Any]
|
|
seal: dict[str, Any]
|
|
events: list[tuple[int, str, str | None, dict[str, Any]]]
|
|
#: 封存范围内 node.finished.evidence 记过的 (节点, 工件)
|
|
ledger: set[tuple[str, str]] = field(default_factory=set)
|
|
#: 节点 id → 它最后一次 node.finished 的产出工件 id
|
|
outputs: dict[str, str] = field(default_factory=dict)
|
|
#: 封存范围内交回过的查询快照：工件 id → {node_id, tool, source, via, call_id, exec}
|
|
queries: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
#: 口径卡节点 id → 封存范围内它的 caliber.upgrade 事件（升版处置）
|
|
upgrades: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
#: 封存范围内 node.finished.evidence 的台账条目，按事件顺序（给可疑实体找「最接近的名字」重建目录用）
|
|
entries: list[dict[str, Any]] = field(default_factory=list)
|
|
#: 封存范围内交回过的表结构快照：工件 id → {node_id, source}
|
|
schemas: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
#: 封存范围内交回过的检索快照：工件 id → {node_id, source}
|
|
retrievals: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
#: 封存范围内台账里记过的分析结论：工件 id → {node_id, exec, call_id / item, inputs}
|
|
analyses: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
#: 封存时记下的清单哈希（没封存的是 None）：旧答案的猜测按 (run_id, 它) 缓存
|
|
manifest_hash: str | None = None
|
|
#: 运行记录上的 manifest_seq 原值（老运行是 None）：按需裁判追加判定前核对封存没换过
|
|
run_manifest_seq: int | None = None
|
|
#: verify_manifest 的完整结论（审计表照实附上）
|
|
verdict: dict[str, Any] = field(default_factory=dict)
|
|
#: 运行记录上的几样：按需裁判只给跑完、封存了的探索运行，升级前发起的不给
|
|
run_class: str | None = None
|
|
status: str | None = None
|
|
ledger_on: bool = False
|
|
#: 按需裁判的判定（evidence.judged 事件），按先后：(seq, 节点, 载荷)。不管落在封存前后都收——
|
|
#: 失败后接着跑的运行会把早先追加的判定封进新的清单里，它仍然是按需裁判的批注
|
|
judged: list[tuple[int, str | None, dict[str, Any]]] = field(default_factory=list)
|
|
#: 节点 id → node.started 上记的名字、类型。子工作流里的节点（「子工作流节点 id/…」）不在这次运行的图上，名字从这里取
|
|
labels: dict[str, str] = field(default_factory=dict)
|
|
types: dict[str, str] = field(default_factory=dict)
|
|
#: 这一次请求里复算过的分析结论：结论工件 id → 复算的结论（_analysis_recheck）。导出页里同一份结论被判定语、具名的数、
|
|
#: 整表的每一格、分析范围各引一处，各接一条链，复算只做一次。只活在这一次请求里：下一次请求重新取、重新验
|
|
rechecks: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
|
|
@property
|
|
def sealed(self) -> bool:
|
|
return bool(self.seal["sealed"])
|
|
|
|
@property
|
|
def trusted(self) -> bool:
|
|
"""封存了、而且封存核对通过。封存范围内有事件被改过的话，范围里的哪一条都不能再
|
|
当「已封存」展示——只看 sealed 的消费方会把一条断掉的封存链画成绿的。"""
|
|
return self.sealed and self.seal["ok"] is True
|
|
|
|
def nodes_of(self, node_type: str, *, nested: bool = False) -> list[str]:
|
|
"""这次运行的图上这一类的节点；nested 为真时连子工作流里跑过的一起给（「子工作流节点 id/…」，类型照 node.started 记的）。"""
|
|
out = [str(n.get("id")) for n in self.graph.get("nodes") or []
|
|
if isinstance(n, dict) and n.get("type") == node_type]
|
|
if nested:
|
|
out += [nid for nid, kind in self.types.items() if "/" in nid and kind == node_type]
|
|
return out
|
|
|
|
def payload_of(self, node_ids: list[str]) -> dict[str, Any] | None:
|
|
"""这几个节点里最后跑完的那个的产出，从封存的事件指向的工件里取。"""
|
|
for node_id in reversed([nid for _, t, nid, _ in self.events if t == "node.finished" and nid in node_ids]):
|
|
artifact = self.outputs.get(node_id)
|
|
try:
|
|
payload = load(artifact) if artifact else None
|
|
except ValueError:
|
|
payload = None
|
|
if isinstance(payload, dict):
|
|
return payload
|
|
return None
|
|
|
|
|
|
|
|
async def read_sealed(run_id: str) -> Sealed:
|
|
from app.engine.runner import verify_manifest
|
|
|
|
async with SessionLocal() as session:
|
|
run = await session.get(Run, run_id)
|
|
if run is None:
|
|
raise LookupError("运行记录不存在")
|
|
rows = [tuple(r) for r in await session.execute(
|
|
select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data)
|
|
.where(RunEvent.run_id == run_id).order_by(RunEvent.seq))]
|
|
graph = run.graph if isinstance(run.graph, dict) else {}
|
|
manifest, manifest_seq = run.manifest_hash, run.manifest_seq
|
|
run_class, status, ledger_on = run.run_class, run.status, ledger_enabled(run)
|
|
verdict = await verify_manifest(run_id)
|
|
bound = verdict.get("sealed_at") if verdict.get("sealed") else None
|
|
# 没封存（还在跑、停在审批）的运行没有「封存之后」：现有的事件都算，但每件证据都标未封存
|
|
events = [r for r in rows if bound is None or r[0] <= bound]
|
|
out = Sealed(run_id=run_id, graph=graph, events=events, seal={
|
|
"sealed": bool(verdict.get("sealed")), "ok": verdict.get("ok"), "manifest_seq": bound,
|
|
"legacy": bool(verdict.get("legacy", False))}, manifest_hash=manifest, run_manifest_seq=manifest_seq,
|
|
verdict=verdict, run_class=run_class, status=status, ledger_on=ledger_on,
|
|
judged=[(seq, nid, data) for seq, etype, nid, data in rows
|
|
if etype == EventType.EVIDENCE_JUDGED and isinstance(data, dict)])
|
|
for _, etype, node_id, data in events:
|
|
if not node_id or not isinstance(data, dict):
|
|
continue
|
|
if etype == "tool.end":
|
|
if data.get("query_artifact"):
|
|
out.queries.setdefault(str(data["query_artifact"]), {
|
|
"node_id": node_id, "tool": data.get("tool"), "via": data.get("artifact")})
|
|
if data.get("schema_artifact"):
|
|
out.schemas.setdefault(str(data["schema_artifact"]), {"node_id": node_id})
|
|
elif etype == "retrieve.end" and data.get("artifact"):
|
|
out.retrievals.setdefault(str(data["artifact"]), {"node_id": node_id, "source": data.get("collection")})
|
|
elif etype == "caliber.upgrade":
|
|
out.upgrades[node_id] = data
|
|
elif etype == "node.started":
|
|
if isinstance(data.get("label"), str) and data["label"]:
|
|
out.labels.setdefault(node_id, data["label"])
|
|
if isinstance(data.get("node_type"), str) and data["node_type"]:
|
|
out.types.setdefault(node_id, data["node_type"])
|
|
if etype != "node.finished":
|
|
continue
|
|
if data.get("artifact"):
|
|
out.outputs[node_id] = data["artifact"]
|
|
for entry in data.get("evidence") or []:
|
|
if isinstance(entry, dict) and entry.get("artifact"):
|
|
out.ledger.add((node_id, entry["artifact"]))
|
|
# 条目里另记的节点 id（双盲复算复算的口径卡）是那个节点所在那一层图里的 id：换成这次运行里的
|
|
out.entries.append({**entry, "node_id": node_id,
|
|
**({"card": _scoped(node_id, entry["card"])} if entry.get("card") else {})})
|
|
artifact = str(entry["artifact"])
|
|
if entry.get("kind") == "query":
|
|
out.queries[artifact] = {
|
|
"node_id": node_id, **{k: entry[k] for k in ("tool", "source", "via", "call_id", "exec")
|
|
if entry.get(k) is not None}}
|
|
elif entry.get("kind") == "schema":
|
|
out.schemas[artifact] = {"node_id": node_id, "source": entry.get("source")}
|
|
elif entry.get("kind") == "retrieval":
|
|
out.retrievals[artifact] = {"node_id": node_id, "source": entry.get("source")}
|
|
elif entry.get("kind") == LEDGER_KIND:
|
|
out.analyses.setdefault(artifact, {
|
|
"node_id": node_id, **{k: entry[k] for k in ("exec", "call_id", "item", "inputs")
|
|
if entry.get(k) is not None}})
|
|
return out
|
|
|
|
|
|
|
|
def metric_cards(sealed: Sealed) -> list[tuple[str, str, dict[str, Any]]]:
|
|
"""封存范围内的口径卡：(节点, 工件, 指标集)。取回时复验哈希，对不上的、取不回来的不要。
|
|
|
|
一期以后口径卡把指标集落成 metric_set 工件、记在台账里；一期以前没有，节点的产出（node.finished
|
|
的 node_output 工件）本身就是指标集——只对画布上确实是口径卡的节点这样认。
|
|
"""
|
|
out: list[tuple[str, str, dict[str, Any]]] = []
|
|
seen: set[tuple[str, str]] = set()
|
|
calibers = set(sealed.nodes_of("metrics"))
|
|
for _, etype, node_id, data in sealed.events:
|
|
if etype != "node.finished" or not node_id or not isinstance(data, dict):
|
|
continue
|
|
arts = [str(e["artifact"]) for e in data.get("evidence") or []
|
|
if isinstance(e, dict) and e.get("kind") == "metric_set" and e.get("artifact")]
|
|
if not arts and node_id in calibers and data.get("artifact"):
|
|
arts = [str(data["artifact"])]
|
|
for art in arts:
|
|
if (node_id, art) in seen:
|
|
continue
|
|
seen.add((node_id, art))
|
|
content, ok = _fetch(art)
|
|
if ok is True and isinstance(content, dict) and isinstance(content.get("metrics"), list):
|
|
out.append((node_id, art, content))
|
|
return out
|