171 lines
6.9 KiB
Python
171 lines
6.9 KiB
Python
"""Describe recorded provenance without turning a source into a verified assertion."""
|
|
|
|
import hashlib
|
|
from uuid import UUID
|
|
import json
|
|
|
|
|
|
SOURCE_KINDS = {"android", "computer", "terminal", "api", "file", "other"}
|
|
|
|
|
|
def android_action_source(device_id: object) -> dict | None:
|
|
"""Stamp a native action at dispatch, using its actual selected device."""
|
|
if not isinstance(device_id, str) or not device_id or device_id == "computer:local":
|
|
return None
|
|
return {"kind": "android", "target_id": device_id, "label": "安卓屏幕"}
|
|
|
|
|
|
def _android_screenshot_records(step: dict) -> list[dict]:
|
|
"""Project explicit capture receipts; never infer provenance from tool args."""
|
|
records = []
|
|
seen = set()
|
|
for tool in step.get("generic_tools") or []:
|
|
if not isinstance(tool, dict) or (
|
|
tool.get("type"),
|
|
tool.get("name"),
|
|
tool.get("status"),
|
|
) != ("tool", "android_file", "success"):
|
|
continue
|
|
payload = tool.get("payload")
|
|
receipt = payload.get("result") if isinstance(payload, dict) else None
|
|
if not isinstance(receipt, dict) or (
|
|
receipt.get("ok") is not True
|
|
or receipt.get("verified") is not True
|
|
or receipt.get("action") != "screenshot"
|
|
):
|
|
continue
|
|
serial, path, digest = (
|
|
receipt.get("device_serial"),
|
|
receipt.get("trace_path"),
|
|
receipt.get("sha256"),
|
|
)
|
|
if not all(isinstance(value, str) and value for value in (serial, path, digest)):
|
|
continue
|
|
captured_at = receipt.get("captured_at")
|
|
captured_at = captured_at if isinstance(captured_at, str) else None
|
|
key = (serial, path, digest, captured_at)
|
|
if key in seen:
|
|
# Nested tool decorators can record the same completed capture twice.
|
|
continue
|
|
seen.add(key)
|
|
records.append(
|
|
{
|
|
"source": {"kind": "android", "target_id": serial, "label": "安卓屏幕"},
|
|
"trace_id": tool.get("trace_id"),
|
|
"artifact_path": path,
|
|
"sha256": digest,
|
|
"captured_at": captured_at,
|
|
}
|
|
)
|
|
return records
|
|
|
|
|
|
def _step_reference(step: dict) -> dict:
|
|
value = step.get("step_id")
|
|
return {
|
|
"step_id": str(value) if isinstance(value, (str, UUID)) else None,
|
|
"step_number": step.get("step_number"),
|
|
}
|
|
|
|
|
|
def _represented_capture(action: dict, captures: list[dict]) -> bool:
|
|
if action.get("action") != "android_file":
|
|
return False
|
|
args = action.get("args") or {}
|
|
if not isinstance(args, dict) or args.get("action") != "screenshot":
|
|
return False
|
|
filename = args.get("filename")
|
|
return any(
|
|
capture["artifact_path"].replace("\\", "/").rsplit("/", 1)[-1] == filename
|
|
for capture in captures
|
|
)
|
|
|
|
|
|
def recorded_sources(steps: list[dict], *, limit: int = 100) -> dict:
|
|
sources = {}
|
|
unattributed = []
|
|
for step in steps:
|
|
result = step.get("last_execution_result")
|
|
execution = result.get("execution") if isinstance(result, dict) else None
|
|
records = execution if isinstance(execution, list) else step.get("action_taken")
|
|
records = records if isinstance(records, list) else [records] if records else []
|
|
captures = _android_screenshot_records(step)
|
|
records = [
|
|
record
|
|
for record in records
|
|
if not isinstance(record, dict)
|
|
or (
|
|
record.get("action")
|
|
not in {"save_note", "read_note", "list_notes", "report_task_status"}
|
|
and not _represented_capture(record, captures)
|
|
)
|
|
]
|
|
records = [*records, *captures]
|
|
if not records:
|
|
# A planning or completion turn with no recorded action is not a
|
|
# missing action source. Screenshots alone do not supply provenance.
|
|
continue
|
|
found = False
|
|
missing = False
|
|
for record in records:
|
|
if not isinstance(record, dict) or not isinstance(record.get("source"), dict):
|
|
missing = True
|
|
continue
|
|
source = record["source"]
|
|
raw_kind = source.get("kind")
|
|
kind = raw_kind if isinstance(raw_kind, str) and raw_kind in SOURCE_KINDS else "other"
|
|
target = source.get("target_id")
|
|
if not isinstance(target, str) or not target:
|
|
missing = True
|
|
continue
|
|
found = True
|
|
plugin = str(record.get("plugin_id") or "")[:100]
|
|
label = str(source.get("label") or "")[:500]
|
|
key = (plugin, kind, target[:256])
|
|
if key not in sources:
|
|
sources[key] = {
|
|
"id": hashlib.sha256(json.dumps(key).encode()).hexdigest()[:20],
|
|
"kind": kind,
|
|
"plugin_id": plugin,
|
|
"target_id": target[:256],
|
|
"label": label,
|
|
"category": "interface" if kind in {"android", "computer"} else "auxiliary",
|
|
"references": [],
|
|
"reference_count": 0,
|
|
}
|
|
reference = _step_reference(step)
|
|
for field in (
|
|
"call_id",
|
|
"source_id",
|
|
"trace_id",
|
|
"artifact_path",
|
|
"sha256",
|
|
"captured_at",
|
|
):
|
|
if isinstance(record.get(field), str) and record[field]:
|
|
reference[field] = record[field][: 2048 if field == "artifact_path" else 256]
|
|
if isinstance(record.get("read_only"), bool):
|
|
reference["read_only"] = record["read_only"]
|
|
entry = sources[key]
|
|
if not entry.get("label_varies") and entry["label"] != label:
|
|
# The same window can navigate between tabs/pages. Its first
|
|
# title is not a valid label for all later evidence.
|
|
entry["label"] = ""
|
|
entry["label_varies"] = True
|
|
entry["reference_count"] += 1
|
|
if len(entry["references"]) < 8:
|
|
entry["references"].append(reference)
|
|
if not found or missing:
|
|
# A screenshot or a model-authored URL does not supply missing provenance.
|
|
unattributed.append(_step_reference(step))
|
|
values = list(sources.values())
|
|
return {
|
|
"schema_version": 1,
|
|
"sources": values[:limit],
|
|
"total_sources": len(values),
|
|
"sources_omitted": max(0, len(values) - limit),
|
|
"unattributed_step_count": len(unattributed),
|
|
"unattributed_references": unattributed[:8],
|
|
"note": "来源清单仅来自执行记录，不代表断言已验证。界面证据与接口、终端、文件来源分开；界面结果不能单独证明后端持久状态。逐项结论及未验证边界以正式验收依据为准。缺失来源不按网址或业务名称推断执行平台。",
|
|
}
|