Introduce identity model and reconcile upstream imports
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a070b5-4994-7271-bd8b-7c3dbcedec4b
This commit is contained in:
parent
a2b254786e
commit
361c944325
38 changed files with 1014 additions and 123 deletions
55
src/info_tech_canon/import_reviews.py
Normal file
55
src/info_tech_canon/import_reviews.py
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
"""Exact reviewed model-import cycles; preserve the complete artifact graph."""
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
def reviewed_import_cycles(artifacts, root: Path) -> list[dict] | None:
|
||||
"""Return reviews only when every cyclic component matches its exact edge set.
|
||||
|
||||
None means the ordinary zero-cycle threshold still applies. Reviews cannot
|
||||
excuse new nodes, edges, relationship types, or a missing boundary rationale.
|
||||
"""
|
||||
path = root / "validation/model-import-reviews.yaml"
|
||||
if not path.is_file():
|
||||
return None
|
||||
try:
|
||||
payload = yaml.safe_load(path.read_text())
|
||||
reviews = payload["reviews"]
|
||||
if not isinstance(reviews, list):
|
||||
return None
|
||||
by_id = {a.id: a for a in artifacts}
|
||||
edges = {(a.id, rel.get("type"), rel.get("target"))
|
||||
for a in artifacts for rel in a.relationships
|
||||
if rel.get("target") in by_id}
|
||||
graph = {key: {target for source, _, target in edges if source == key}
|
||||
for key in by_id}
|
||||
reachable = {}
|
||||
for node in graph:
|
||||
seen, pending = set(), list(graph[node])
|
||||
while pending:
|
||||
target = pending.pop()
|
||||
if target not in seen:
|
||||
seen.add(target)
|
||||
pending.extend(graph[target] - seen)
|
||||
reachable[node] = seen
|
||||
components = {frozenset(other for other in graph
|
||||
if other in reachable[node] and node in reachable[other])
|
||||
for node in graph if node in reachable[node]}
|
||||
matched = []
|
||||
for nodes in sorted(components, key=lambda group: sorted(group)):
|
||||
actual = {edge for edge in edges if edge[0] in nodes and edge[2] in nodes}
|
||||
for review in reviews:
|
||||
expected = {tuple(edge) for edge in review["edges"]}
|
||||
rationale = (root / review["boundary_review"]).resolve()
|
||||
if (actual == expected and all(by_id[node].kind == "model" for node in nodes)
|
||||
and rationale.is_relative_to(root.resolve()) and rationale.is_file()
|
||||
and all(kind == "uses" for _, kind, _ in actual)):
|
||||
matched.append({"nodes": sorted(nodes), "edges": sorted(actual),
|
||||
"boundary_review": review["boundary_review"]})
|
||||
break
|
||||
else:
|
||||
return None
|
||||
return matched
|
||||
except (KeyError, TypeError, ValueError, OSError, yaml.YAMLError):
|
||||
return None
|
||||
|
|
@ -214,6 +214,13 @@ def validate_canon(root: Path | str | None = None) -> dict[str, Any]:
|
|||
checks.metrics,
|
||||
context.infospace.config.viability,
|
||||
)
|
||||
from .import_reviews import reviewed_import_cycles
|
||||
import_reviews = reviewed_import_cycles(context.infospace.artifacts, context.infospace_root)
|
||||
if import_reviews:
|
||||
threshold_errors = [error for error in threshold_errors
|
||||
if error.get("metric") != "consistency_cycles"]
|
||||
checks.details["reviewed_model_import_cycles"] = import_reviews
|
||||
checks.details["cycle_threshold_exception"] = "Exact reviewed reciprocal model imports; raw metric retained"
|
||||
errors.extend(threshold_errors)
|
||||
structural = structural_checks(context)
|
||||
errors.extend(structural["errors"])
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue