File size: 6,306 Bytes
23ade4e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """Runtime/legacy graph input adapter and immutable resolved assembly carrier."""
from __future__ import annotations
from dataclasses import dataclass
import json
from types import MappingProxyType
from typing import Any, Mapping
from runtime.eventframe.contracts import ArticleLocalRuntimeResult
from .assembly import ArticleLocalKGAssembler
from .contracts import ArticleLocalKnowledgeGraphResult, GRAPH_SCHEMA_VERSION
def deep_freeze(value: Any) -> Any:
"""Recursively detach and freeze a JSON-compatible value."""
if isinstance(value, Mapping):
return MappingProxyType({str(key): deep_freeze(item) for key, item in value.items()})
if isinstance(value, (list, tuple)):
return tuple(deep_freeze(item) for item in value)
return value
def deep_thaw(value: Any) -> Any:
"""Return a detached JSON-compatible copy of a frozen carrier value."""
if isinstance(value, Mapping):
return {str(key): deep_thaw(item) for key, item in value.items()}
if isinstance(value, tuple):
return [deep_thaw(item) for item in value]
return value
def _json_copy(value: Mapping[str, Any]) -> dict[str, Any]:
return json.loads(json.dumps(value, ensure_ascii=False, allow_nan=False))
@dataclass(frozen=True, slots=True)
class ResolvedAssemblyState:
"""Immutable assembly-only view of already resolved runtime identities."""
source_kind: str
source_graph: Mapping[str, Any]
article: Mapping[str, Any]
resolved_events: tuple[Mapping[str, Any], ...]
resolved_entities: tuple[Mapping[str, Any], ...]
statements: tuple[Mapping[str, Any], ...]
time_expressions: tuple[Mapping[str, Any], ...]
semantic_edges: tuple[Mapping[str, Any], ...]
unmaterialized_evidence: Mapping[str, Any]
source_lanes: Mapping[str, Any]
coverage: Mapping[str, Any]
runtime_provenance: Mapping[str, Any]
@classmethod
def from_legacy_graph(
cls,
graph: Mapping[str, Any],
*,
source_kind: str,
) -> "ResolvedAssemblyState":
detached = _json_copy(graph)
if detached.get("schema_version") != GRAPH_SCHEMA_VERSION:
raise ValueError("resolved assembly state requires the canonical legacy graph")
nodes = detached.get("nodes", [])
edges = detached.get("edges", [])
member_event_ids = {
str(edge["source_id"])
for edge in edges
if edge.get("edge_type") == "MEMBER_OF_EVENT"
}
resolved_events = [node for node in nodes if node.get("kind") == "LOCAL_EVENT"]
resolved_events.extend(
node
for node in nodes
if node.get("kind") == "EVENT" and str(node["node_id"]) not in member_event_ids
)
frozen_graph = deep_freeze(detached)
return cls(
source_kind=source_kind,
source_graph=frozen_graph,
article=deep_freeze(detached["article"]),
resolved_events=tuple(deep_freeze(row) for row in resolved_events),
resolved_entities=tuple(
deep_freeze(node) for node in nodes if node.get("kind") == "LOCAL_ENTITY"
),
statements=tuple(
deep_freeze(node) for node in nodes if node.get("kind") == "STATEMENT"
),
time_expressions=tuple(
deep_freeze(node) for node in nodes if node.get("kind") == "TIME"
),
semantic_edges=tuple(
deep_freeze(edge)
for edge in edges
if edge.get("edge_type") != "MEMBER_OF_EVENT"
),
unmaterialized_evidence=deep_freeze(
detached.get("unmaterialized_evidence", {})
),
source_lanes=deep_freeze(detached.get("source_lanes", {})),
coverage=deep_freeze(detached.get("lane_statuses", {})),
runtime_provenance=deep_freeze(detached.get("provenance", {})),
)
def identity_ids_by_scope(self) -> Mapping[str, tuple[str, ...]]:
return MappingProxyType(
{
"EVENT": tuple(str(row["node_id"]) for row in self.resolved_events),
"ENTITY": tuple(str(row["node_id"]) for row in self.resolved_entities),
"STATEMENT": tuple(str(row["node_id"]) for row in self.statements),
"TIME": tuple(str(row["node_id"]) for row in self.time_expressions),
}
)
def identity_ids(self) -> tuple[str, ...]:
return tuple(
identity
for scope in ("EVENT", "ENTITY", "STATEMENT", "TIME")
for identity in self.identity_ids_by_scope()[scope]
)
def to_legacy_graph(self) -> dict[str, Any]:
return deep_thaw(self.source_graph)
class AssemblyInputAdapter:
"""Adapt only canonical runtime/legacy contracts without model or Gold imports."""
policy_id = "ASSEMBLY_INPUT_ADAPTER_V1"
def __init__(self, legacy_assembler: ArticleLocalKGAssembler) -> None:
self._legacy_assembler = legacy_assembler
def adapt(
self,
source: (
ArticleLocalRuntimeResult
| ArticleLocalKnowledgeGraphResult
| Mapping[str, Any]
),
) -> ResolvedAssemblyState:
if isinstance(source, ArticleLocalRuntimeResult):
graph = self._legacy_assembler.assemble(source).to_dict()
source_kind = "ARTICLE_LOCAL_RUNTIME_RESULT"
elif isinstance(source, ArticleLocalKnowledgeGraphResult):
graph = source.to_dict()
source_kind = "LEGACY_ASSEMBLY_RESULT"
elif isinstance(source, Mapping):
snapshot = _json_copy(source)
if snapshot.get("schema_version") == GRAPH_SCHEMA_VERSION:
graph = snapshot
source_kind = "LEGACY_ASSEMBLY_MAPPING"
else:
graph = self._legacy_assembler.assemble(snapshot).to_dict()
source_kind = "RUNTIME_RESULT_MAPPING"
if snapshot != _json_copy(source):
raise ValueError("assembly input adapter mutated its source mapping")
else:
raise TypeError("unsupported assembly input contract")
return ResolvedAssemblyState.from_legacy_graph(graph, source_kind=source_kind)
|