Listing: Listing

from fractions import Fraction
from itertools import product

def validate_graph(parents):
    nodes = set(parents)
    if any(p not in nodes for ps in parents.values() for p in ps):
        raise ValueError("undeclared parent vertex")
    done = set()
    while len(done) < len(nodes):
        ready = {
            v for v in nodes - done
            if set(parents[v]) <= done
        }
        if not ready:
            raise ValueError("cycle in declared network")
        done |= ready

def ancestors(parents, target, blocked=()):
    if target not in parents:
        raise ValueError("undeclared target")
    blocked = set(blocked)
    if target in blocked:
        return set()
    seen, pending = set(), list(parents[target])
    while pending:
        v = pending.pop()
        if v in blocked or v in seen:
            continue
        seen.add(v)
        pending.extend(parents[v])
    return seen

def audit_declared_cut(parents, input_vertices, encoding_events,
                       query_event, decoder_event, cut_vertices):
    validate_graph(parents)
    all_named = (set(input_vertices) | set(encoding_events)
                 | set(cut_vertices)
                 | {query_event, decoder_event})
    if not all_named <= set(parents):
        raise ValueError("audit references undeclared vertices")

    query_reaches_encoding = any(
        query_event in ancestors(parents, e)
        for e in encoding_events
    )
    bypass_sources = set(input_vertices) & ancestors(
        parents, decoder_event, blocked=cut_vertices
    )
    return {
        "late_query_graph_check": not query_reaches_encoding,
        "declared_cut_path_check": not bypass_sources,
        "bypass_sources": sorted(bypass_sources),
        "physical_completeness_verified": False,
        "freshness_verified": False,
        "capacity_verified": False,
        "joint_resource_independence_verified": False,
    }

def exact_two_state_benchmark():
    inputs = tuple(product((0, 1), repeat=2))
    best, count = Fraction(0, 1), 0
    for encoder in product((0, 1), repeat=4):
        for decoder in product((0, 1), repeat=4):
            successes = 0
            for i, x in enumerate(inputs):
                m = encoder[i]
                for j in range(2):
                    successes += decoder[2 * m + j] == x[j]
            best = max(best, Fraction(successes, 8))
            count += 1
    return best, count

def proposed_checks():
    # Proposed assertions only. No executed pass is reported.
    graph = {
        "X1": (),
        "X2": (),
        "fresh_query_source": (),
        "E1": ("X1",),
        "E2": ("E1", "X2"),
        "cut_memory": ("E2",),
        "storage": ("cut_memory",),
        "query": ("fresh_query_source",),
        "decoder": ("storage", "query"),
        "archive": ("X1", "X2", "query"),
        "score": ("archive", "decoder"),
    }
    report = audit_declared_cut(
        graph, ("X1", "X2"), ("E1", "E2"),
        "query", "decoder", ("cut_memory",)
    )
    assert report["late_query_graph_check"]
    assert report["declared_cut_path_check"]
    assert not report["freshness_verified"]

    # Add an omitted pulse-log route around the declared cut.
    bad_log = dict(graph)
    bad_log["pulse_log"] = ("E1",)
    bad_log["decoder"] = ("storage", "query", "pulse_log")
    report = audit_declared_cut(
        bad_log, ("X1", "X2"), ("E1", "E2"),
        "query", "decoder", ("cut_memory",)
    )
    assert not report["declared_cut_path_check"]
    assert "X1" in report["bypass_sources"]

    # Add advance access to the query.
    bad_query = dict(graph)
    bad_query["E1"] = ("X1", "query")
    report = audit_declared_cut(
        bad_query, ("X1", "X2"), ("E1", "E2"),
        "query", "decoder", ("cut_memory",)
    )
    assert not report["late_query_graph_check"]

    optimum, visited = exact_two_state_benchmark()
    assert optimum == Fraction(3, 4)
    assert visited == 256