Surface Layer — Invalidation and Regeneration Runbook

from rdflib import Graph
from surface.invalidation import compute_read_source_digest

# Load the changed source
g = Graph()
g.parse("path/to/changed/source.ttl")

# Compute current digest for the source IRI
source_iri = "http://example.org/concepts/Coverage_Cyber"
current_digest = compute_read_source_digest(g, source_iri)
print(f"Current digest: {current_digest}")

from surface.invalidation import read_set_from_manifest

# Load the generated surface package
manifest_g = Graph()
manifest_g.parse("ontology/surface/execution/my-contract/manifest.ttl")

# Extract all recorded read-set entries
read_set = read_set_from_manifest(manifest_g)

# Find entries for the source
for entry in read_set:
    if entry['readSource'] == source_iri:
        print(f"Recorded digest: {entry['readHash']}")
        print(f"Recorded version: {entry['readVersion']}")
        break

from surface.invalidation import compare_read_set

# Get all recorded entries
all_entries = read_set_from_manifest(manifest_g)

# Find which entries are stale
stale_entries = compare_read_set(all_entries, {'source_iri': current_digest})

if stale_entries:
    print(f"Surface is stale: {len(stale_entries)} entry/ies changed")
else:
    print("Surface is fresh; no regeneration needed")

from surface.invalidation import plan_regeneration

# Identify changed sources
changed_sources = [source_iri]

# Get the regeneration scope
surfaces, mappings, artefacts = plan_regeneration(
    changed_sources=changed_sources,
    changed_mappings=[],
    changed_profiles=[],
    graph=manifest_g
)

print(f"Surfaces to regenerate: {len(surfaces)}")
print(f"Mappings to recompile: {len(mappings)}")
print(f"Artefacts to replace: {len(artefacts)}")

import subprocess

for surface_key in surfaces:
    contract_path = f"ontology/surface/examples/{surface_key}.ttl"
    print(f"Regenerating {surface_key}...")
    
    result = subprocess.run(
        ["python", "-m", "surface", "lower", "--contracts", contract_path],
        cwd="/Users/t4/work/lattice",
        capture_output=True
    )
    
    if result.returncode != 0:
        print(f"ERROR: Regeneration failed for {surface_key}")
        print(result.stderr.decode())
        return False

print("✅ All surfaces regenerated")

# For each affected mapping in MORK
for mapping_iri in mappings:
    # Identify which backend(s) target this mapping
    
    # SPARQL backend
    result = subprocess.run(
        ["python", "-m", "mork_compilers", "compile", "--backend", "sparql", mapping_iri],
        capture_output=True
    )
    
    # SHACL backend
    result = subprocess.run(
        ["python", "-m", "mork_compilers", "compile", "--backend", "shacl", mapping_iri],
        capture_output=True
    )
    
    print(f"✅ Recompiled {mapping_iri}")

changed_mapping_iri = "http://example.org/mappings/CoverageTypeProjection_v1"

print(f"Changed mapping: {changed_mapping_iri}")

from surface.invalidation import plan_regeneration

# Plan based on mapping change
surfaces, mappings, artefacts = plan_regeneration(
    changed_sources=[],
    changed_mappings=[changed_mapping_iri],
    changed_profiles=[],
    graph=manifest_g
)

print(f"Affected artefacts: {len(artefacts)}")

# Load MORK graph to find dependencies
mork_g = Graph()
mork_g.parse("ontology/mork/spec/Mork.ttl")

# Find what depends on the changed mapping
dependents = []
for s, p, o in mork_g.triples((None, mork_ns.dependsOnMapping, changed_mapping_iri)):
    dependents.append(s)

print(f"Dependent mappings: {dependents}")

# Follow generatedBy edges for artefacts
for artefact_iri in artefacts:
    print(f"Regenerate artefact: {artefact_iri}")

# A profile change is intentionally broad
changed_profile = "http://example.org/profiles/GeneralV2"

print(f"Changed profile: {changed_profile}")

from surface.invalidation import plan_regeneration

# Profile changes affect everything downstream
surfaces, mappings, artefacts = plan_regeneration(
    changed_sources=[],
    changed_mappings=[],
    changed_profiles=[changed_profile],
    graph=manifest_g
)

print(f"⚠️  Broad impact: {len(surfaces)} surfaces, {len(mappings)} mappings, {len(artefacts)} artefacts")

# Topological sort by dependency
from surface.invalidation import topo_sort_by_dependency

surfaces_ordered = topo_sort_by_dependency(surfaces, graph=manifest_g)

for surface_key in surfaces_ordered:
    print(f"Regenerating {surface_key}...")
    # [regeneration code from Scenario 1]

# After successful regeneration, update all manifests
for surface_key in surfaces:
    manifest_path = f"ontology/surface/execution/{surface_key}/manifest.ttl"
    g = Graph()
    g.parse(manifest_path)
    
    # Update profile reference (example)
    profile_old = rdflib.Namespace("http://example.org/profiles/GeneralV1")
    profile_new = rdflib.Namespace("http://example.org/profiles/GeneralV2")
    
    # Replace references
    for s, p, o in g.triples((None, rdflib.RDF.type, profile_old.SurfaceProfile)):
        g.remove((s, p, o))
        g.add((s, p, profile_new.SurfaceProfile))
    
    g.serialize(manifest_path, format="turtle")

canonicalisation_changed = True
recorded_version = "srf-canon/1"  # old
new_version = "srf-canon/2"       # new

print(f"Canonicalisation cutover: {recorded_version} → {new_version}")

from surface.invalidation import invalidate_all_hashes

# Every recorded srf:readHash, srf:semanticContentHash, srf:artefactHash is now stale
for surface_package in all_package_paths():
    manifest_g = Graph()
    manifest_g.parse(f"{surface_package}/manifest.ttl")
    
    # Remove all old hashes
    invalidate_all_hashes(manifest_g)
    
    # Will be recomputed during regeneration
    manifest_g.serialize(f"{surface_package}/manifest.ttl", format="turtle")

for surface_package in all_package_paths():
    g = Graph()
    contract_path = f"{surface_package}/contract.ttl"
    g.parse(contract_path)
    
    # Update canonicalisation version
    surface_profile = g.value(None, RDF.type, SRF.SurfaceProfile)
    g.set((surface_profile, SRF.canonicalisationVersion, Literal(new_version)))
    
    g.serialize(contract_path, format="turtle")

changed_sources = [
    "http://example.org/concepts/Coverage_Cyber",
    "http://example.org/concepts/Coverage_Auto",
]

surfaces, mappings, artefacts = plan_regeneration(
    changed_sources=changed_sources,
    changed_mappings=[],
    changed_profiles=[],
    graph=manifest_g
)

# Changed: skos:broader relations under Coverage_Auto hierarchy
# Impact: srf:matchesCoverageTypeAncestor closure needs refresh
# No impact: ScopeCoverageType_* class definitions (symbols stay the same)

# Monitor for changes to that external source
# Example: NFPA codes reference catalog

from surface.invalidation import watch_external_source

watch_external_source(
    source_iri="http://nfpa.org/codes/PerlTypeScheme",
    manifest_path="ontology/surface/execution/peril-index/manifest.ttl",
    polling_interval_hours=24
)

# Automatically regenerate if digest changes