from __future__ import annotations import hashlib import json import re from typing import Any from .schema import EvidenceSpec, SourceBinding, SourceRecord, unresolved_binding _INDEX = re.compile(r"\[([0-9]+)\]") def json_pointer(path: str | None) -> str: """Convert the visualization's legacy dotted paths to RFC 6901 pointers.""" value = str(path or "").strip() if not value: return "" if " / " in value: return "" value = _INDEX.sub(r".\1", value) parts = [part for part in value.split(".") if part] return "/" + "/".join( part.replace("~", "~0").replace("/", "~1") for part in parts ) def pointer_lookup(value: Any, pointer: str) -> tuple[bool, Any]: if not pointer: return True, value current = value for raw in pointer.lstrip("/").split("/"): part = raw.replace("~1", "/").replace("~0", "~") if isinstance(current, list): try: current = current[int(part)] except (ValueError, IndexError): return False, None elif isinstance(current, dict): if part not in current: return False, None current = current[part] else: return False, None return True, current def pointer_value(value: Any, pointer: str) -> Any: return pointer_lookup(value, pointer)[1] def source_digest(value: Any) -> str: serialized = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) return hashlib.sha256(serialized.encode("utf-8")).hexdigest() class SourceResolver: """Maps semantic CardData fields to deduplicated physical source records. It deliberately does not infer adoption from matching text. The only evidence inputs are the field's declared relation and physical origin. """ def __init__( self, *, bundle: dict[str, Any], event_details: dict[int, dict[str, Any]], ) -> None: self.bundle = bundle self.event_details = event_details self.sources: dict[str, SourceRecord] = {} self._span_offsets: dict[tuple[int, str, str], int] = {} def binding_for_field( self, field: dict[str, Any], *, binding_id: str, role: str | None = None, ) -> SourceBinding: source = field.get("source") if isinstance(field.get("source"), dict) else {} source_kind = str(source.get("kind") or "") source_ref = str(source.get("ref") or "") raw, physical = self._raw_source(source_kind, source_ref, source) if source_kind == "calculation": raw = { "value": field.get("value"), "calculation": source.get("label"), } source_id = self._source_id(source_kind, source_ref, source, physical) field_path = str(source.get("fieldPath") or "") pointer = json_pointer(field_path) resolution = "resolved" if raw is None: missing = self._ensure_missing_source( source_id, source, reason=f"未找到 {source.get('label') or source_ref or '原始记录'}", ) return unresolved_binding( binding_id, reason=missing["locator"]["reason"], role=role or self._role(field), source_id=source_id, code="source-record-not-saved", ) selector: dict[str, Any] found, selected = pointer_lookup(raw, pointer) if pointer else (True, raw) if source_kind == "calculation": selector = {"kind": "json-pointer", "path": "/value"} selected = field.get("value") elif source_kind == "artifact" and field_path in {"snapshotRef", "recordCounts", "historicalVersion"}: selector = {"kind": "whole-record"} selected = field.get("value") elif pointer and found: selector = {"kind": "json-pointer", "path": pointer} elif pointer: # The legacy CardData path may refer to a derived projection rather # than a persisted record. Keep the source visible, but do not # pretend that the pointer resolves against it. selector = { "kind": "unresolved", "reason": f"原始记录中无法安全定位字段 {field_path}", "code": "field-path-mismatch", } selected = field.get("value") resolution = "partial" else: selector = {"kind": "whole-record"} selected = field.get("value") record = self._ensure_source(source_id, source_kind, source, raw, physical) record.setdefault("selectedValues", []).append( { "bindingId": binding_id, "fieldId": field.get("id"), "label": field.get("label"), "path": pointer or field_path or None, "value": selected, } ) transform = self._transform(source_kind, source) if ( transform.get("kind") == "direct" and found and not _equivalent_value(field.get("value"), selected) ): transform = { "kind": "parsed", "operation": "business-projection-v1", "outputKey": str(field.get("id") or binding_id), } binding: SourceBinding = { "id": binding_id, "sourceId": source_id, "role": role or self._role(field), # type: ignore[typeddict-item] "selector": selector, "transform": transform, "evidence": self._evidence(field, source_kind), "resolution": resolution, # type: ignore[typeddict-item] } if transform.get("kind") == "parsed": binding["evidence"]["confidence"] = "deterministic" if binding["resolution"] != "resolved" and transform.get("kind") == "direct": binding["transform"] = { "kind": "parsed", "operation": "unresolved-source-projection-v1", "outputKey": str(field.get("id") or binding_id), } binding["evidence"]["confidence"] = "unconfirmed" if "未安全关联" in str(source.get("label") or ""): binding["resolution"] = "unsafe" binding["evidence"]["confidence"] = "unconfirmed" return binding def binding_for_event( self, event_id: int, *, binding_id: str, path: str = "", role: str = "process", availability: str = "produced-by-run", ) -> SourceBinding: event = self.event_details.get(int(event_id)) source_id = f"event:{event_id}" if event is None: self._ensure_missing_source( source_id, {"label": f"Event {event_id}", "ref": source_id}, reason=f"Event {event_id} 或 Event Body 未保存", ) return unresolved_binding( binding_id, reason=f"Event {event_id} 或 Event Body 未保存", role=role, source_id=source_id, code="event-body-not-saved", ) source = { "kind": "runtime-event", "label": str(event.get("event_name") or f"Event {event_id}"), "ref": source_id, "fieldPath": path, } pointer = json_pointer(path) found, selected = pointer_lookup(event, pointer) if pointer else (True, event) resolution = "resolved" if found else "partial" if not pointer: selector = {"kind": "whole-record"} elif found: selector = {"kind": "json-pointer", "path": pointer} else: selector = { "kind": "unresolved", "reason": f"Event {event_id} 中不存在字段 {path}", "code": "field-path-mismatch", } record = self._ensure_source(source_id, "runtime-event", source, event, event) record.setdefault("selectedValues", []).append( {"bindingId": binding_id, "fieldId": binding_id, "label": source["label"], "path": pointer or None, "value": selected} ) return { "id": binding_id, "sourceId": source_id, "role": role, # type: ignore[typeddict-item] "selector": selector, "transform": {"kind": "direct"}, "evidence": { "availability": availability, "adoption": "not-applicable" if availability == "produced-by-run" else "not-recorded", "confidence": "exact", }, # type: ignore[typeddict-item] "resolution": resolution, # type: ignore[typeddict-item] } def text_span_binding( self, event_id: int, *, binding_id: str, path: str, exact_text: Any, role: str, ) -> SourceBinding: event = self.event_details.get(int(event_id)) source_id = f"event:{event_id}" if event is None: self._ensure_missing_source( source_id, {"label": f"Event {event_id}", "ref": source_id}, reason=f"Event {event_id} 或 Event Body 未保存", ) return unresolved_binding( binding_id, reason=f"Event {event_id} 或 Event Body 未保存", role=role, source_id=source_id, code="event-body-not-saved", ) pointer = json_pointer(path) raw_text = pointer_value(event, pointer) needle = str(exact_text or "") if not isinstance(raw_text, str) or not needle: binding = self.binding_for_event( event_id, binding_id=binding_id, path=path, role=role, availability="returned-to-agent" if role == "output" else "direct-read", ) binding["transform"] = { "kind": "parsed", "operation": "structured-text-section-v1", "outputKey": binding_id, } binding["evidence"]["confidence"] = "deterministic" return binding span_key = (int(event_id), pointer, needle) search_from = self._span_offsets.get(span_key, 0) start = raw_text.find(needle, search_from) if start < 0 and search_from: start = raw_text.find(needle) if start < 0: binding = self.binding_for_event( event_id, binding_id=binding_id, path=path, role=role, availability="returned-to-agent" if role == "output" else "direct-read", ) binding["transform"] = { "kind": "parsed", "operation": "normalized-text-section-v1", "outputKey": binding_id, } binding["evidence"]["confidence"] = "deterministic" return binding record = self._ensure_source( source_id, "runtime-event", {"label": str(event.get("event_name") or f"Event {event_id}"), "ref": source_id}, event, event, ) record.setdefault("selectedValues", []).append( {"bindingId": binding_id, "fieldId": binding_id, "label": binding_id, "path": pointer, "value": needle} ) self._span_offsets[span_key] = start + len(needle) return { "id": binding_id, "sourceId": source_id, "role": role, # type: ignore[typeddict-item] "selector": { "kind": "text-span", "path": pointer, "startCodePoint": start, "endCodePoint": start + len(needle), "exactText": needle, "sourceDigest": source_digest(raw_text), }, "transform": {"kind": "direct"}, "evidence": { "availability": "returned-to-agent" if role == "output" else "direct-read", "adoption": "not-recorded", "confidence": "exact", }, "resolution": "resolved", } def add_unresolved_source(self, binding: SourceBinding) -> None: source_id = binding["sourceId"] reason = str(binding.get("selector", {}).get("reason") or "原始记录无法定位") self._ensure_missing_source( source_id, {"label": "记录缺失"}, reason=reason, ) def add_log_anchor( self, event_id: int, module: dict[str, Any], *, event_msg_id: str ) -> str: source_id = f"log:event:{event_id}" self.sources[source_id] = { "id": source_id, "kind": "log-anchor", "label": f"锚定日志模块 · Event {event_id}", "locator": { "table": "script_build_log", "eventId": event_id, "msgId": event_msg_id, "logAnchor": module.get("anchor"), }, "completeness": "complete", "resultState": "present", "truncated": False, "selectedValues": [], "rawRecord": module.get("content"), } return source_id def empty_database_result_binding( self, *, binding_id: str, table: str, filters: dict[str, Any], label: str, role: str = "output", ) -> SourceBinding: """Represent a successful zero-row lookup without calling it missing.""" filter_key = ",".join(f"{key}={filters[key]}" for key in sorted(filters)) source_id = f"db-result:{table}:{filter_key or 'all'}" self.sources[source_id] = { "id": source_id, "kind": "database", "label": label, "locator": { "table": table, "filters": filters, "resultCount": 0, }, "completeness": "complete", "resultState": "empty", "truncated": False, "selectedValues": [{ "bindingId": binding_id, "fieldId": binding_id, "label": "查询结果", "path": "", "value": [], }], "rawRecord": [], } return { "id": binding_id, "sourceId": source_id, "role": role, # type: ignore[typeddict-item] "selector": {"kind": "whole-record"}, "transform": { "kind": "calculated", "operation": "数据库查询结果计数为 0", }, "evidence": { "availability": "visualization-derived", "adoption": "not-applicable", "confidence": "deterministic", }, "resolution": "resolved", } def members_binding( self, field: dict[str, Any], *, binding_id: str, members: list[dict[str, Any]], reducer: str, role: str = "output", ) -> SourceBinding: """Create a deterministic aggregate while retaining every member ref.""" base_field = { **field, "source": { "kind": "calculation", "label": reducer, "ref": f"calculation:{binding_id}", "fieldPath": "value", }, "relation": "persisted-output", } base = self.binding_for_field(base_field, binding_id=binding_id, role=role) member_selectors: list[dict[str, Any]] = [] member_resolutions: list[str] = [] for index, member in enumerate(members, 1): member_binding = self.binding_for_field( member, binding_id=f"{binding_id}:member:{index}", role=role, ) member_selectors.append( { "sourceId": member_binding["sourceId"], "selector": member_binding["selector"], } ) member_resolutions.append(str(member_binding.get("resolution") or "missing")) member_source = self.sources.get(member_binding["sourceId"]) if member_source and member_source.get("selectedValues"): member_source["selectedValues"][-1]["aggregateBindingId"] = binding_id if not member_selectors: base["selector"] = { "kind": "unresolved", "reason": "确定性聚合没有找到可定位的参与记录", "code": "binding-map-missing", } base["resolution"] = "missing" base["evidence"]["confidence"] = "unconfirmed" return base base["selector"] = { "kind": "members", "members": member_selectors, "reducer": reducer, } base["transform"] = {"kind": "calculated", "operation": reducer} base["evidence"] = { "availability": "visualization-derived", "adoption": "not-applicable", "confidence": "deterministic", } if any(value in {"missing", "unsafe"} for value in member_resolutions): base["resolution"] = "partial" base["evidence"]["confidence"] = "unconfirmed" elif any(value == "partial" for value in member_resolutions): base["resolution"] = "partial" return base def _raw_source( self, source_kind: str, source_ref: str, source: dict[str, Any] ) -> tuple[Any, dict[str, Any]]: if source_kind == "runtime-event": event_id = _suffix_int(source_ref) if event_id is not None: event = self.event_details.get(event_id) return event, event or {} # A runtime source is only trustworthy when it names one concrete # Event. Treating an empty ref as "all related Events" previously # let a multipath evaluator impersonate a missing script_evaluator # in round summaries. return None, {"detailRef": source_ref, "reason": "未保存可定位的 Event ref"} if source_kind == "database": return self._database_record(source_ref, str(source.get("label") or "")) if source_kind == "artifact": if "branch:" in source_ref: branch_id = _suffix_int(source_ref) row = _by_int(self.bundle.get("branches"), "branch_id", branch_id) # CardData paths are rooted at the branch record # (`candidate_snapshot.snapshot`), so retain that root rather # than pre-extracting candidate_snapshot and breaking the # declared JSON Pointer. return row, {"table": "script_build_branch", "recordId": (row or {}).get("id")} artifact = self.bundle.get("currentArtifact") return artifact, {"artifactRef": source_ref or "artifact:base:current"} if source_kind == "calculation": return {"value": None, "calculation": source.get("label")}, {"calculation": source.get("label")} return None, {"ref": source_ref} def _database_record(self, source_ref: str, label: str) -> tuple[Any, dict[str, Any]]: normalized = label.lower() if source_ref.startswith("missing-convergence:"): parts = source_ref.split(":", 4) record_type = parts[1] if len(parts) > 1 else "unknown" round_index = _integer(parts[3]) if len(parts) > 3 else None branch_ids = [ _integer(value) for value in (parts[4].split(",") if len(parts) > 4 else []) if _integer(value) is not None ] return [], { "tables": ( ["script_build_event", "script_build_event_body"] if record_type == "review" else ["script_build_multipath_decision"] ), "filters": { "round_index": round_index, "branch_ids": branch_ids, }, "resultCount": 0, } if "script_build_record" in normalized or source_ref == "run:objective" or source_ref == "run:final-result": row = self.bundle.get("record") return row, {"table": "script_build_record", "recordId": (row or {}).get("id")} if "data_decision" in normalized or source_ref.startswith("data-decision:"): row = _by_int(self.bundle.get("dataDecisions"), "id", _suffix_int(source_ref)) return row, {"table": "script_build_data_decision", "recordId": (row or {}).get("id")} if "multipath" in normalized or source_ref.startswith("multipath-decision:"): row = _by_int(self.bundle.get("multipathDecisions"), "id", _suffix_int(source_ref)) return row, {"table": "script_build_multipath_decision", "recordId": (row or {}).get("id")} if "domain_info" in normalized or source_ref.startswith("domain-info:"): row = _by_int(self.bundle.get("domainInfo"), "id", _suffix_int(source_ref)) return row, {"table": "script_build_domain_info", "recordId": (row or {}).get("id")} if "script_build_round" in normalized or source_ref.startswith("round:") and ":branch:" not in source_ref: round_index = _round_index(source_ref) row = _by_int(self.bundle.get("rounds"), "round_index", round_index) return row, {"table": "script_build_round", "recordId": (row or {}).get("id")} if "script_build_branch" in normalized or ":branch:" in source_ref: branch_id = _branch_id(source_ref) row = _by_int(self.bundle.get("branches"), "branch_id", branch_id) return row, {"table": "script_build_branch", "recordId": (row or {}).get("id")} return None, {"table": label or "unknown", "ref": source_ref} def _ensure_source( self, source_id: str, source_kind: str, source: dict[str, Any], raw: Any, physical: dict[str, Any], ) -> SourceRecord: existing = self.sources.get(source_id) if existing is not None: return existing kind = { "database": "database", "runtime-event": "runtime-event", "artifact": "artifact", "calculation": "calculation", "log-anchor": "log-anchor", }.get(source_kind, "missing") if "历史回退" in str(source.get("label") or ""): kind = "historical-fallback" completeness = "complete" truncated = False omitted = 0 if kind == "runtime-event" and isinstance(raw, dict): sides = [raw.get("input"), raw.get("output")] if raw.get("input") is None and raw.get("output") is None: completeness = "partial" for side in sides: if isinstance(side, dict) and side.get("truncated"): truncated = True omitted += int(side.get("omittedCharacters") or 0) record: SourceRecord = { "id": source_id, "kind": kind, # type: ignore[typeddict-item] "label": str(source.get("label") or source.get("ref") or source_id), "locator": physical, "completeness": "partial" if truncated else completeness, # type: ignore[typeddict-item] "resultState": ( "empty" if raw == [] or ( kind == "calculation" and isinstance(raw, dict) and raw.get("value") == [] ) else "present" ), "truncated": truncated, "selectedValues": [], "rawRecord": raw, } if omitted: record["omittedCharacters"] = omitted if isinstance(raw, dict) and kind == "runtime-event": record["status"] = str(raw.get("status") or "unknown") record["durationMs"] = _integer(raw.get("duration_ms")) record["locator"] = { "eventId": raw.get("id"), "eventName": raw.get("event_name"), "eventType": raw.get("event_type"), "table": "script_build_event + script_build_event_body", } self.sources[source_id] = record return record def _ensure_missing_source( self, source_id: str, source: dict[str, Any], *, reason: str ) -> SourceRecord: record = self.sources.get(source_id) if record is None: record = { "id": source_id, "kind": "missing", "label": str(source.get("label") or source_id), "locator": {"ref": source.get("ref"), "reason": reason}, "completeness": "missing", "resultState": "missing", "truncated": False, "selectedValues": [], "rawRecord": None, } self.sources[source_id] = record return record @staticmethod def _source_id( source_kind: str, source_ref: str, source: dict[str, Any], physical: dict[str, Any], ) -> str: if source_kind == "runtime-event" and _suffix_int(source_ref) is not None: return f"event:{_suffix_int(source_ref)}" if source_kind == "database" and physical.get("table"): return f"db:{physical['table']}:{physical.get('recordId') or source_ref or 'unknown'}" if source_kind == "artifact": return f"artifact:{source_ref or physical.get('artifactRef') or 'unknown'}" key = f"{source_kind}:{source_ref}:{source.get('label')}" return f"source:{hashlib.sha1(key.encode('utf-8')).hexdigest()[:12]}" @staticmethod def _role(field: dict[str, Any]) -> str: relation = str(field.get("relation") or "") return { "direct-input": "input", "previous-result": "basis", "standing-constraint": "constraint", "explicit-basis": "basis", "available-upstream": "basis", "persisted-output": "output", "run-output": "output", }.get(relation, "basis") @staticmethod def _transform(source_kind: str, source: dict[str, Any]) -> dict[str, Any]: if source_kind == "calculation": return {"kind": "calculated", "operation": str(source.get("label") or "确定性计算")} if "历史回退" in str(source.get("label") or ""): return { "kind": "fallback", "selectedSourceId": str(source.get("ref") or ""), "candidateSourceIds": [str(source.get("ref") or "")], } return {"kind": "direct"} @staticmethod def _evidence(field: dict[str, Any], source_kind: str) -> EvidenceSpec: relation = str(field.get("relation") or "") if source_kind == "calculation": return { "availability": "visualization-derived", "adoption": "not-applicable", "confidence": "deterministic", } if relation == "explicit-basis": return { "availability": "direct-read", "adoption": "explicitly-adopted", "confidence": "exact", } if relation in {"direct-input", "standing-constraint"}: return { "availability": "direct-read", "adoption": "not-recorded", "confidence": "exact", } if relation == "previous-result": return { "availability": "returned-to-agent" if source_kind == "runtime-event" else "available-upstream", "adoption": "not-recorded", "confidence": "exact" if source_kind == "runtime-event" else "safe-association", } if relation == "available-upstream": return { "availability": "available-upstream", "adoption": "not-recorded", "confidence": "safe-association", } if relation == "persisted-output": availability = "returned-to-agent" if source_kind == "runtime-event" else "produced-by-run" return { "availability": availability, "adoption": "not-recorded" if availability == "returned-to-agent" else "not-applicable", "confidence": "exact", } if relation == "run-output": return { "availability": "produced-by-run", "adoption": "not-applicable", "confidence": "exact", } return { "availability": "available-upstream", "adoption": "not-recorded", "confidence": "unconfirmed", } def _integer(value: Any) -> int | None: try: return int(value) except (TypeError, ValueError): return None def _suffix_int(value: str) -> int | None: return _integer(str(value or "").rsplit(":", 1)[-1]) def _round_index(ref: str) -> int | None: match = re.search(r"(?:^|:)round:(\d+)", f":{ref}") return _integer(match.group(1)) if match else None def _branch_id(ref: str) -> int | None: match = re.search(r":branch:(\d+)", ref) return _integer(match.group(1)) if match else None def _by_int(items: Any, key: str, wanted: int | None) -> dict[str, Any] | None: if wanted is None: return None return next( ( item for item in items or [] if isinstance(item, dict) and _integer(item.get(key)) == wanted ), None, ) def _equivalent_value(left: Any, right: Any) -> bool: try: return json.dumps(left, ensure_ascii=False, sort_keys=True, default=str) == json.dumps( right, ensure_ascii=False, sort_keys=True, default=str ) except (TypeError, ValueError): return str(left) == str(right)