#!/usr/bin/env python3
"""
spec_schedkit.py -- executable specification for schedkit.py.

Exit 0  == every contract held.
Exit 1  == at least one contract failed (the failure is printed).

Contract classes
  A. fixture cases     hand-computed waves, printed in full
  B. invariant checks  every output re-derived from first principles
  C. oracle check      find_cycles against an independent brute-force enumeration
  D. randomized        1,000 DAGs x 40 permutations, checked against the oracle
  E. determinism       repeated calls, build-order shuffles, deep graphs
  F. cycle handling    exact CycleError messages, DOT/JSON refusal
  G. scale             50,000 tasks, 100,000 edges, under a wall-clock budget
  H. cost shape        acyclic graphs must be linear; a 50,000-node ring must
                       be walked iteratively (the recursive build died at 2,000)
  I. linear extension  topological_order() must return the lexicographically
                       SMALLEST topological order -- the documented "min-id
                       first" contract -- checked against brute force over all
                       permutations, not against the module's own heap

The randomized section is the teeth: it compares this implementation against a
*different* algorithm (brute-force cycle enumeration over permutations), not
against itself. A tautological test suite cannot pass this file.
"""

import itertools
import json
import os
import random
import sys
import time
from typing import Dict, List, Sequence, Set, Tuple

# Import the schedkit *beside this file*, not one from a hard-coded path: the
# mutation harness copies both files into a scratch dir and mutates the copy.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import schedkit  # noqa: E402
from schedkit import CycleError, Graph, find_cycles, iter_plan, plan  # noqa: E402

FAILS: List[str] = []
CHECKS = 0
T_START = time.monotonic()


def check(name: str, got, want) -> None:
    global CHECKS
    CHECKS += 1
    if got != want:
        FAILS.append("%s\n       got : %r\n       want: %r" % (name, got, want))


def check_true(name: str, cond: bool, detail: str = "") -> None:
    global CHECKS
    CHECKS += 1
    if not cond:
        FAILS.append("%s %s" % (name, detail))


def section(title: str) -> None:
    print("\n== %s" % title)


# =========================================================== A. fixture cases
def canon_ids(cyc: Sequence[str]) -> Tuple[str, ...]:
    """Rotate a cycle so its lexicographically smallest id comes first.

    Mirrors schedkit._canon, but written here from the *contract* (a cycle is a
    ring, so any rotation names it) rather than copied from the module.
    """
    if not cyc:
        return tuple(cyc)
    i = min(range(len(cyc)), key=lambda k: (cyc[k], k))
    return tuple(cyc[i:]) + tuple(cyc[:i])


def build(edges, extra_tasks=()):
    g = Graph()
    for t in extra_tasks:
        g.add_task(t)
    return g.add_edges(edges)


def fixtures() -> None:
    section("A. fixtures")

    # A1: diamond. a -> b, a -> c, b -> d, c -> d
    g = build([("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")])
    check("A1 diamond waves", plan(g), (("a",), ("b", "c"), ("d",)))

    # A2: independent chains do not wait on each other
    g = build([("a1", "a2"), ("b1", "b2"), ("a2", "a3")])
    check("A2 two chains", plan(g), (("a1", "b1"), ("a2", "b2"), ("a3",)))

    # A3: empty graph
    check("A3 empty", plan(Graph()), ())

    # A4: single task, no deps
    check("A4 single", plan(build([], ["solo"])), (("solo",),))

    # A5: lexicographic order inside a wave (insertion order is reversed)
    g = Graph()
    for t in ["zeta", "alpha", "mike", "bravo"]:
        g.add_task(t)
    check("A5 wave order", plan(g), (("alpha", "bravo", "mike", "zeta"),))

    # A6: a long chain gives one wave per task
    g = build([(chr(97 + i), chr(98 + i)) for i in range(5)])
    check("A6 chain of 6", plan(g), tuple((c,) for c in "abcdef"))

    # A7: wave depth is prerequisite-depth, not "connected component"
    g = build([("root", "x"), ("root", "y"), ("x", "z"), ("y", "z"), ("unrelated", "u2")])
    check("A7 depth", plan(g), (("root", "unrelated"), ("u2", "x", "y"), ("z",)))

    # A8: the graph is untouched by planning (no mutation)
    g = build([("a", "b"), ("b", "c")])
    before = g.edges()
    plan(g)
    check("A8 no mutation", g.edges(), before)

    # A9: duplicate edge is idempotent
    g = Graph()
    g.add_edge("a", "b").add_edge("a", "b").add_edge("a", "b")
    check("A9 idempotent edges", g.edges(), (("a", "b"),))

    # A10: self-loop is a cycle, not a crash
    g = Graph()
    g.add_task("self", ["self"])
    check("A10 self loop", find_cycles(g), (("self",),))


# ====================================================== B. invariant checks
def check_invariants(name: str, g: Graph) -> None:
    waves = plan(g)
    flat = tuple(iter_plan(waves))

    # B1 partition: every task exactly once, in the graph
    check("%s: partition" % name, tuple(sorted(flat)), g.tasks())

    # B2 waves sorted ascending
    for i, w in enumerate(waves):
        check("%s: wave %d sorted" % (name, i), list(w), sorted(w))

    # B3 acyclicity of the emitted plan order
    pos = {t: i for i, t in enumerate(flat)}
    for pre, post in g.edges():
        check_true(
            "%s: order respects %s->%s" % (name, pre, post),
            pos[pre] < pos[post],
            "(pos %d !< %d)" % (pos[pre], pos[post]),
        )

    # B4 wave of t == 1 + max(wave of deps); 0 if no deps
    rank = {t: i for i, w in enumerate(waves) for t in w}
    for t in g.tasks():
        want = 1 + max((rank[d] for d in g.deps(t)), default=-1)
        check("%s: rank(%s)" % (name, t), rank[t], want)

    # B5 waves are contiguous and non-empty
    check_true("%s: non-empty waves" % name, all(waves), str(waves))

    # B6 plan_json is valid JSON and agrees with plan()
    doc = json.loads(schedkit.plan_json(g))
    check("%s: json waves" % name, tuple(tuple(w) for w in doc["waves"]), waves)
    check("%s: json n_tasks" % name, doc["validation"]["n_tasks"], len(g))


def invariants() -> None:
    section("B. invariants")
    cases = {
        "diamond": build([("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")]),
        "star": build([("h", "a"), ("h", "b"), ("h", "c"), ("h", "d")]),
        "empty": Graph(),
        "single": build([], ["only"]),
        "mesh": build(
            [("a", "b"), ("a", "c"), ("b", "d"), ("c", "d"), ("a", "d"), ("b", "c")]
        ),
        "two-roots": build([("r1", "x"), ("r2", "x"), ("x", "y")]),
        "fan-out-then-fan-in": build(
            [("r", "a"), ("r", "b"), ("r", "c"), ("a", "z"), ("b", "z"), ("c", "z")]
        ),
    }
    for name, g in cases.items():
        check_invariants(name, g)


# ============================================================= C. oracle check
def oracle_cycles(g: Graph) -> Tuple[Tuple[str, ...], ...]:
    """Independent brute force: DFS from every node, enumerate simple cycles.

    Deliberately a different algorithm from schedkit.find_cycles (plain DFS with
    path tracking rather than Johnson + Tarjan), so agreement is real evidence.
    """
    adj = {t: sorted(g.dependents(t)) for t in g.tasks()}
    found: Set[Tuple[str, ...]] = set()

    def canon(cyc: List[str]) -> Tuple[str, ...]:
        i = min(range(len(cyc)), key=lambda k: cyc[k])
        return tuple(cyc[i:] + cyc[:i])

    for start in sorted(adj):
        stack: List[Tuple[str, List[str]]] = [(start, [start])]
        while stack:
            node, path = stack.pop()
            for nxt in adj[node]:
                if nxt == start and len(path) >= 1:
                    found.add(canon(path))
                elif nxt not in path:
                    stack.append((nxt, path + [nxt]))
    return tuple(sorted(found))


def oracle() -> None:
    section("C. oracle: find_cycles vs brute-force DFS")
    cases = {
        "acyclic": build([("a", "b"), ("b", "c"), ("a", "c")]),
        "one-cycle": build([("a", "b"), ("b", "c"), ("c", "a")]),
        "two-cycles": build(
            [("a", "b"), ("b", "a"), ("x", "y"), ("y", "x"), ("a", "x")]
        ),
        "nested": build(
            [("a", "b"), ("b", "c"), ("c", "a"), ("b", "d"), ("d", "a")]
        ),
        "self-loop": (lambda: (lambda g: (g.add_task("s", ["s"]), g)[1])(Graph()))(),
        "figure-eight": build(
            [("a", "b"), ("b", "a"), ("b", "c"), ("c", "b"), ("a", "d"), ("d", "a")]
        ),
        "chorded": build(
            [("a", "b"), ("b", "c"), ("c", "d"), ("d", "a"), ("a", "c"), ("b", "d")]
        ),
        "empty": Graph(),
    }
    for name, g in cases.items():
        check("C oracle %s" % name, find_cycles(g), oracle_cycles(g))


# ============================================================ D. randomized
def random_dag(rng: random.Random, n: int, p: float) -> Graph:
    """Random DAG: edge i->j only when i<j, so acyclicity is by construction."""
    g = Graph()
    ids = ["t%03d" % i for i in range(n)]
    for t in ids:
        g.add_task(t)
    for i in range(n):
        for j in range(i + 1, n):
            if rng.random() < p:
                g.add_edge(ids[i], ids[j])
    return g


def randomized(trials: int = 1000, perms: int = 40, seed: int = 20261021) -> None:
    section("D. randomized differential (%d DAGs x %d permutations)" % (trials, perms))
    rng = random.Random(seed)
    t0 = time.monotonic()
    oracle_calls = 0

    for trial in range(trials):
        n = rng.randint(0, 7)
        p = rng.choice([0.0, 0.05, 0.15, 0.3, 0.6, 1.0])
        g = random_dag(rng, n, p)
        ids = g.tasks()
        base = plan(g)
        base_cycles = find_cycles(g)

        # D1: oracle agreement on cycles (small graphs only -- brute force)
        if n <= 6:
            check("D1 trial %d cycles" % trial, base_cycles, oracle_cycles(g))
            oracle_calls += 1

        # D2: the plan is a permutation of the tasks
        check("D2 trial %d partition" % trial, tuple(sorted(iter_plan(base))), ids)

        # D3: isomorphism invariance under `perms` random relabellings.
        # The relabelled plan must equal the original plan with every id
        # substituted and the wave re-sorted -- a different id set is the whole
        # point of the check, and wave order is defined by id order, not by the
        # order the original ids happened to sit in.
        for k in range(perms):
            perm = list(ids)
            rng.shuffle(perm)
            mapping = {old: "n%03d" % i for i, old in enumerate(perm)}
            g2 = g.relabel(mapping)
            want_plan = tuple(
                tuple(sorted(mapping[t] for t in w)) for w in base
            )
            check("D3 trial %d perm %d" % (trial, k), plan(g2), want_plan)
            want_cycles = tuple(
                sorted(canon_ids([mapping[t] for t in c]) for c in base_cycles)
            )
            check("D3c trial %d perm %d cycles" % (trial, k), find_cycles(g2), want_cycles)
            check("D3b trial %d perm %d task count" % (trial, k), len(g2), len(g))

        # D4: every edge is respected by the emitted order
        flat = list(iter_plan(base))
        pos = {t: i for i, t in enumerate(flat)}
        for pre, post in g.edges():
            check_true(
                "D4 trial %d edge %s->%s" % (trial, pre, post),
                pos[pre] < pos[post],
            )

    dt = time.monotonic() - t0
    print("   %d DAGs, %d oracle comparisons, %.3fs" % (trials, oracle_calls, dt))


# ============================================================== E. determinism
def determinism() -> None:
    section("E. determinism")
    edges = [("a", "b"), ("b", "c"), ("a", "c"), ("d", "e"), ("e", "c"), ("f", "b")]

    g = build(edges)
    runs = [plan(g) for _ in range(200)]
    check_true("E1 200 runs identical", all(r == runs[0] for r in runs))

    # build-order shuffle must not change the plan
    rng = random.Random(7)
    outs = set()
    for _ in range(50):
        shuffled = list(edges)
        rng.shuffle(shuffled)
        gg = Graph()
        for pre, post in shuffled:
            gg.add_edge(pre, post)
        outs.add(plan(gg))
    check("E2 build order irrelevant", outs, {plan(build(edges))})

    # deep graph must not blow the interpreter stack
    deep = Graph()
    N = 50000
    for i in range(N - 1):
        deep.add_edge("d%06d" % i, "d%06d" % (i + 1))
    t0 = time.monotonic()
    waves = plan(deep)
    dt = time.monotonic() - t0
    check("E3 deep chain waves", len(waves), N)
    check("E3b deep chain last", waves[-1], ("d%06d" % (N - 1),))
    print("   deep chain: %d tasks, %d waves, %.3fs" % (N, len(waves), dt))

    # wide graph: one root, 20k leaves
    wide = Graph()
    for i in range(20000):
        wide.add_edge("root", "leaf%05d" % i)
    t0 = time.monotonic()
    w = plan(wide)
    dt = time.monotonic() - t0
    check("E4 wide waves", (len(w), len(w[0]), len(w[1])), (2, 1, 20000))
    print("   wide graph: 20001 tasks, %d waves, %.3fs" % (len(w), dt))

    # build-order shuffle of a deep chain (insertion order reversed)
    rev = Graph()
    for i in range(N - 1, 0, -1):
        rev.add_edge("d%06d" % (i - 1), "d%06d" % i)
    check("E5 deep reversed identical", plan(rev), waves)


# =========================================================== F. cycle handling
def cycle_handling() -> None:
    section("F. cycle handling")
    g = build([("a", "b"), ("b", "c"), ("c", "a")])
    try:
        plan(g)
        check_true("F1 CycleError raised", False, "no exception")
    except CycleError as e:
        check("F1 CycleError cycles", e.cycles, (("a", "b", "c"),))
        check("F1b message", str(e), "dependency cycle detected: a -> b -> c -> a")

    # canonical rotation: same cycle entered from a different node
    g2 = build([("c", "a"), ("a", "b"), ("b", "c")])
    check("F2 canonical rotation", find_cycles(g2), (("a", "b", "c"),))

    # DOT and JSON must refuse a cyclic graph, not emit a broken plan
    for label, fn in (("DOT", schedkit.plan_graphviz), ("JSON", schedkit.plan_json)):
        try:
            fn(g)
            check_true("F3 %s refuses cycles" % label, False, "no exception")
        except CycleError:
            check_true("F3 %s refuses cycles" % label, True)

    # two disjoint cycles are both reported
    g3 = build([("a", "b"), ("b", "a"), ("x", "y"), ("y", "x")])
    check("F4 two cycles", find_cycles(g3), (("a", "b"), ("x", "y")))

    # a cycle reachable from an acyclic region still blocks the whole plan
    g4 = build([("root", "a"), ("a", "b"), ("b", "a")])
    try:
        plan(g4)
        check_true("F5 reachable cycle blocks", False, "no exception")
    except CycleError:
        check_true("F5 reachable cycle blocks", True)


# ==================================================================== G. scale
def scale() -> None:
    section("G. scale")
    budget = 20.0
    t0 = time.monotonic()

    # layered graph: 50k tasks, 100k edges, 10 layers
    L, W = 10, 5000
    g = Graph()
    layers = [["L%d_%05d" % (l, i) for i in range(W)] for l in range(L)]
    for l in layers:
        for t in l:
            g.add_task(t)
    for l in range(L - 1):
        for i in range(W):
            for k in range(2):  # 2 edges per task -> 100k edges
                g.add_edge(layers[l][i], layers[l + 1][(i + k) % W])
    t1 = time.monotonic()
    waves = plan(g)
    t2 = time.monotonic()
    check("G1 layered waves", (len(waves), tuple(len(w) for w in waves)), (L, tuple([W] * L)))
    check("G2 edge count", len(g.edges()), W * 2 * (L - 1))
    print(
        "   %d tasks, %d edges: build %.3fs, plan %.3fs"
        % (len(g), len(g.edges()), t1 - t0, t2 - t1)
    )
    check_true("G3 under budget", (t2 - t0) < budget, "(%.2fs >= %.0fs)" % (t2 - t0, budget))

    # relabeling 50k tasks must also be linear-ish
    t3 = time.monotonic()
    ids = g.tasks()
    mapping = {t: "R" + t[1:] for t in ids}
    g2 = g.relabel(mapping)
    w2 = plan(g2)
    t4 = time.monotonic()
    check("G4 relabel plan", [tuple(sorted(mapping[t] for t in w)) for w in waves],
          [tuple(sorted(t for t in w)) for w in w2])
    print("   relabel+plan 50k: %.3fs" % (t4 - t3))
    check_true("G5 relabel under budget", (t4 - t3) < budget, "(%.2fs)" % (t4 - t3))


# ========================================== H. cost shape (why the build was fixed)
def shapes() -> None:
    section("H. cost shape: acyclic graphs linear, big SCCs iterative")

    # H1: a 2,000-node chain contains no cycle at all. The previous build took
    # 5.6651s here (measured, quadratic); a linear one takes milliseconds. This
    # gate catches a relapse into re-scanning every adjacency list per vertex.
    chain = Graph()
    for i in range(1999):
        chain.add_edge("d%06d" % i, "d%06d" % (i + 1))
    t0 = time.monotonic()
    check("H1 chain has no cycles", find_cycles(chain), ())
    dt = time.monotonic() - t0
    print("   chain 2000: %.4fs (quadratic build measured 5.6651s)" % dt)
    check_true("H1b chain under 0.5s", dt < 0.5, "(%.3fs)" % dt)

    # H2: one 50,000-node ring is one SCC. Exactly one cycle, and it is found
    # without recursing -- the recursive build raised RecursionError at n=2000
    # with the default limit of 1000.
    n = 50000
    ring = Graph()
    for i in range(n):
        ring.add_edge("r%06d" % i, "r%06d" % ((i + 1) % n))
    t0 = time.monotonic()
    cyc = find_cycles(ring)
    dt = time.monotonic() - t0
    check("H2 ring one cycle", (len(cyc), len(cyc[0])), (1, n))
    check("H2b ring is the ring", cyc[0], tuple("r%06d" % i for i in range(n)))
    print("   ring 50000: %.3fs, recursionlimit %d" % (dt, sys.getrecursionlimit()))
    check_true("H2c ring under 10s", dt < 10.0, "(%.2fs)" % dt)

    # H3: 2,000 disjoint 5-rings -- every cycle reported, one SCC pass total.
    many = Graph()
    for r in range(2000):
        ids = ["g%04d_%d" % (r, i) for i in range(5)]
        for i in range(5):
            many.add_edge(ids[i], ids[(i + 1) % 5])
    t0 = time.monotonic()
    c = find_cycles(many)
    dt = time.monotonic() - t0
    check("H3 disjoint rings", len(c), 2000)
    print("   2000 disjoint 5-rings: %d cycles, %.3fs" % (len(c), dt))


# ========================================= I. the linear extension's own contract
def smallest_topo_oracle(g: Graph) -> Tuple[str, ...]:
    """The lexicographically smallest topological order, by brute force.

    Independent of the module and of any heap: walk every permutation of the
    task ids in lexicographic order (itertools yields them sorted when the input
    is sorted) and return the first one that respects every edge. Small graphs
    only -- n! is the cost.
    """
    ids = g.tasks()
    deps = {t: set(g.deps(t)) for t in ids}
    for perm in itertools.permutations(ids):
        seen: Set[str] = set()
        for t in perm:
            if not deps[t] <= seen:
                break
            seen.add(t)
        else:
            return perm
    return ()


def is_linear_extension(order: Sequence[str], g: Graph) -> bool:
    if sorted(order) != list(g.tasks()):
        return False
    pos = {t: i for i, t in enumerate(order)}
    return all(pos[pre] < pos[post] for pre, post in g.edges())


def linear_extension() -> None:
    section("I. topological_order is the SMALLEST linear extension")

    # I1: hand-computed. Kahn with a min-heap emits a, b, c, d; a plain stack
    # (`ready.pop()`) emits a, c, b, d -- the same valid order, but not the
    # smallest one, which is what the module's docstring promises.
    g = build([("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")])
    check("I1 diamond topo", g.topological_order(), ("a", "b", "c", "d"))

    # I2: tie-break is by id, not by insertion order or by stack position
    g = build([("root", "zeta"), ("root", "alpha"), ("root", "mike")])
    check("I2 tie-break by id", g.topological_order(), ("root", "alpha", "mike", "zeta"))

    # I3: two ready roots -- the smaller one goes first, its whole chain too
    g = build([("b1", "b2"), ("a1", "a2")])
    check("I3 two roots", g.topological_order(), ("a1", "a2", "b1", "b2"))

    # I4: chain and single task
    g = build([(chr(97 + i), chr(98 + i)) for i in range(5)])
    check("I4 chain topo", g.topological_order(), tuple("abcdef"))
    check("I4b single", build([], ["solo"]).topological_order(), ("solo",))
    check("I4c empty", Graph().topological_order(), ())

    # I5: differential against brute force over all permutations, 300 random DAGs
    rng = random.Random(4242)
    checked = 0
    for trial in range(300):
        n = rng.randint(0, 6)
        p = rng.choice([0.0, 0.1, 0.25, 0.5, 0.8, 1.0])
        gg = random_dag(rng, n, p)
        want = smallest_topo_oracle(gg)
        check("I5 trial %d smallest linear extension" % trial, gg.topological_order(), want)
        checked += 1
    print("   %d random DAGs compared against all-permutations brute force" % checked)

    # I6: insertion order must not matter -- ids alone decide the order
    edges = [("a", "b"), ("a", "c"), ("c", "d"), ("b", "e"), ("d", "e"), ("z", "e")]
    outs = set()
    for _ in range(50):
        shuffled = list(edges)
        rng.shuffle(shuffled)
        outs.add(build(shuffled).topological_order())
    check("I6 build order irrelevant", outs, {build(edges).topological_order()})

    # I7: still a valid linear extension at a size brute force cannot reach
    big = Graph()
    for i in range(3000):
        big.add_task("t%04d" % i)
    for i in range(0, 2999, 3):
        big.add_edge("t%04d" % i, "t%04d" % (i + 1))
    order = big.topological_order()
    check_true("I7 3000-node order valid", is_linear_extension(order, big))
    check("I7b 3000-node order size", len(order), 3000)

    # I8: a cyclic graph must raise, not silently truncate the order
    cyc = build([("a", "b"), ("b", "a")])
    try:
        cyc.topological_order()
        check_true("I8 cyclic raises", False, "no exception")
    except CycleError:
        check_true("I8 cyclic raises", True)


# ======================================================================= main
def main() -> int:
    print("schedkit spec -- %s" % schedkit.__file__)
    print("python %s" % sys.version.split()[0])

    fixtures()
    invariants()
    oracle()
    randomized()
    determinism()
    cycle_handling()
    scale()
    shapes()
    linear_extension()

    dt = time.monotonic() - T_START
    print("\n%d checks, %d failed, %.2fs" % (CHECKS, len(FAILS), dt))
    if FAILS:
        print("\nFAILURES (%d):" % len(FAILS))
        for f in FAILS[:15]:
            print("  - " + f)
        if len(FAILS) > 15:
            print("  ... and %d more" % (len(FAILS) - 15))
        return 1
    print("SPEC OK: all %d checks held" % CHECKS)
    return 0


if __name__ == "__main__":
    sys.exit(main())