"""Fictional static dependency graph; Python 3.9+, standard library only.

Run: python3 knowledge_graph_demo.py
All records are teaching examples, not real packages or advisories.
The traversal returns one shortest witness per reachable service, not all paths.
"""

from collections import defaultdict, deque


SERVICES = {
    "svc:order-api": "team:commerce",
    "svc:report-api": "team:data",
}
DEPENDENCIES = [
    ("svc:order-api", "pkg:checkout-sdk:2.1", "build-order-1"),
    ("pkg:checkout-sdk:2.1", "pkg:parser:1.4", "manifest-checkout-1"),
    ("svc:report-api", "pkg:parser:1.4", "build-report-1"),
]


def affected_services(target, dependencies, services):
    """Return reachable services and a shortest path with edge evidence.

    Inputs must already be filtered for snapshot, runtime scope and permissions.
    An empty result alone does not distinguish unknown targets from known leaves.
    Path copying keeps this example readable, but adds cost beyond plain BFS.
    """
    reverse = defaultdict(list)
    for user, dependency, evidence in dependencies:
        reverse[dependency].append((user, evidence))

    queue = deque([(target, [target], [])])
    seen = {target}
    result = {}
    while queue:
        current, path, evidence = queue.popleft()
        for user, source in reverse[current]:
            if user in seen:
                continue
            seen.add(user)
            next_path = [user] + path
            next_evidence = [source] + evidence
            if user in services:
                result[user] = (next_path, next_evidence)
            queue.append((user, next_path, next_evidence))
    return result


def check_examples():
    """Executable checks of the cases explained in article four."""
    target = "pkg:parser:1.4"
    actual = affected_services(target, DEPENDENCIES, SERVICES)
    assert set(actual) == set(SERVICES)
    assert actual["svc:order-api"] == (
        ["svc:order-api", "pkg:checkout-sdk:2.1", target],
        ["build-order-1", "manifest-checkout-1"],
    )
    assert actual["svc:report-api"] == (
        ["svc:report-api", target], ["build-report-1"],
    )
    assert set(affected_services(
        "pkg:checkout-sdk:2.1", DEPENDENCIES, SERVICES
    )) == {"svc:order-api"}
    assert affected_services("pkg:unknown:0", DEPENDENCIES, SERVICES) == {}
    assert affected_services(target, [], SERVICES) == {}

    cycle = DEPENDENCIES + [(target, "pkg:checkout-sdk:2.1", "cycle-demo")]
    assert affected_services(target, cycle, SERVICES) == actual
    duplicate = DEPENDENCIES + [DEPENDENCIES[0]]
    assert affected_services(target, duplicate, SERVICES) == actual

    multiple = DEPENDENCIES + [("svc:order-api", target, "direct-demo")]
    assert affected_services(target, multiple, SERVICES)["svc:order-api"] == (
        ["svc:order-api", target], ["direct-demo"],
    )
    # Validate that each returned witness is made of real, correctly oriented edges.
    for path, evidence in actual.values():
        assert len(evidence) == len(path) - 1
        assert all(
            (path[i], path[i + 1], source) in DEPENDENCIES
            for i, source in enumerate(evidence)
        )


def main():
    check_examples()
    target = "pkg:parser:1.4"
    known_nodes = set(SERVICES)
    for source, destination, _ in DEPENDENCIES:
        known_nodes.update((source, destination))
    if target not in known_nodes:
        print("Target not found in this snapshot; no safety conclusion is possible.")
        return
    print("Snapshot: demo-2026-10-01 (fictional, runtime dependencies only)")
    print("Checks passed. Review candidates, not confirmed exploitable services:")
    for service, (path, evidence) in sorted(
        affected_services(target, DEPENDENCIES, SERVICES).items()
    ):
        print(f"{service} | {SERVICES[service]}")
        print("  path: " + " -> ".join(path))
        print("  evidence: " + ", ".join(evidence))
    print("Unverified: production deployment and template-extension condition.")


if __name__ == "__main__":
    main()