dtact 0.2.8

"""
verify_fsm.py — DTA-V3 Scheduler FSM Finite Verifier
=====================================================
Exhaustive small-scale simulation of the DTA-V3 chunk state machine
for N ∈ {2, 3, 4} workers and max_hops = floor(N/2).

Verifies three correctness properties on every reachable state:
  1. No Task Loss     — Φ(σ) = const for every live transition
  2. Deadlock Freedom — wait-for graph G_wf is always empty
  3. Livelock Freedom — no chunk reaches hop_count > max_hops

Mathematical objects correspond to the "Formal Mathematical Model"
chapter of main.tex (formerly a standalone formal_model.tex before
the paper was consolidated into a single file):
  τ  → task index (int)
  c  → chunk (tasks, count, hop_count)
  σ  → system state (dict: site_name → frozenset of task indices)
  Φ  → total particle count (number operator N̂)
  V  → Lyapunov potential = max_hops - hop_count

The simulation models the *chunk-level* FSM, not the full system
(which would be too large to enumerate).  Each chunk carries a
single task (count=1) for simplicity; the multi-task case follows
by linearity of the conservation law.

Usage:
    python verify_fsm.py          # runs all scenarios
    python verify_fsm.py --verbose # prints each state transition
"""

import sys
import argparse
from collections import deque
from dataclasses import dataclass, field
from typing import FrozenSet, Optional, List, Tuple, Dict

# ─────────────────────────────────────────────────────────────
# Data structures
# ─────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class Chunk:
    """A chunk carrying a single task τ with hop counter h."""
    tau: int          # task index
    hop_count: int    # h ∈ {0, …, max_hops}

@dataclass(frozen=True)
class State:
    """
    Occupation-number state σ: a snapshot of one task τ's location.

    site is one of:
      "ext"          — not yet admitted (source)
      "M_{i}_{j}"   — in mailbox from worker i to worker j
      "Q_{i}"       — in worker i's local queue
      "W"            — in the warehouse
      "E_{i}"       — executing on worker i
      "sink"         — completed (absorbing)

    hop is the chunk's current hop_count (only meaningful when in M or W).
    """
    site: str
    hop: int          # current hop_count of the chunk

# ─────────────────────────────────────────────────────────────
# Transition generator
# ─────────────────────────────────────────────────────────────

def successors(state: State, N: int, max_hops: int) -> List[Tuple[str, State]]:
    """
    Return list of (operator_name, next_state) for all valid transitions
    from `state` in a system with N workers and given max_hops.

    Each transition corresponds to one operator Ô_* from Definition 5.2
    of the "Formal Mathematical Model" chapter of main.tex.
    """
    site = state.site
    h    = state.hop
    results = []

    # ── Ô_ext : ext → M_{ext,i} for each worker i ──────────────
    if site == "ext":
        for i in range(N):
            results.append((
                f"O_ext(→W{i})",
                State(site=f"M_ext_{i}", hop=0)
            ))
        return results

    # ── External mailbox M_ext_i → Q_i ─────────────────────────
    if site.startswith("M_ext_"):
        i = int(site.split("_")[-1])
        results.append((
            f"O_push-local(ext→Q{i})",
            State(site=f"Q_{i}", hop=h)
        ))
        return results

    # ── Internal mailbox M_{i}_{j} ─────────────────────────────
    if site.startswith("M_") and not site.startswith("M_ext"):
        parts = site.split("_")
        i, j = int(parts[1]), int(parts[2])

        # O_push-local: chunk accepted into Q_j
        results.append((
            f"O_push-local(M{i}{j}→Q{j})",
            State(site=f"Q_{j}", hop=h)
        ))

        # O_hop: re-deflect to another mailbox (only if h < max_hops)
        if h < max_hops:
            for j2 in range(N):
                if j2 != i and j2 != j:  # avoid self-column and current target
                    results.append((
                        f"O_hop(M{i}{j}→M{i}{j2}, h={h}→{h+1})",
                        State(site=f"M_{i}_{j2}", hop=h+1)
                    ))

        # O_park: park in warehouse (when h == max_hops OR all mailboxes full)
        # We always offer this as a non-deterministic choice to cover
        # the "all mailboxes full" scenario.
        results.append((
            f"O_park(M{i}{j}→W, h={h})",
            State(site="W", hop=h)
        ))

        return results

    # ── Local queue Q_i ─────────────────────────────────────────
    if site.startswith("Q_"):
        i = int(site.split("_")[1])

        # O_dispatch: pop and execute
        results.append((
            f"O_dispatch(Q{i}→E{i})",
            State(site=f"E_{i}", hop=h)
        ))

        # O_mailbox: deflect to a mailbox (only offered as possibility)
        for j in range(N):
            if j != i:
                results.append((
                    f"O_mailbox(Q{i}→M{i}{j})",
                    State(site=f"M_{i}_{j}", hop=0)
                ))

        return results

    # ── Warehouse W ─────────────────────────────────────────────
    if site == "W":
        # O_drain: any worker drains the warehouse into their Q
        for i in range(N):
            results.append((
                f"O_drain(W→Q{i})",
                State(site=f"Q_{i}", hop=h)
            ))
        return results

    # ── Executing E_i ───────────────────────────────────────────
    if site.startswith("E_"):
        i = int(site.split("_")[1])

        # O_finish: task completes → sink
        results.append((
            f"O_finish(E{i}→sink)",
            State(site="sink", hop=h)
        ))

        # O_requeue: Notified → back to Q_i
        results.append((
            f"O_requeue(E{i}→Q{i})",
            State(site=f"Q_{i}", hop=h)
        ))

        return results

    # ── Sink (absorbing) ─────────────────────────────────────────
    if site == "sink":
        return []  # no outgoing transitions

    raise ValueError(f"Unknown site: {site}")

# ─────────────────────────────────────────────────────────────
# Property checkers
# ─────────────────────────────────────────────────────────────

LIVE_SITES = lambda site: site not in ("ext", "sink")

def particle_count(state: State) -> int:
    """
    Φ(σ) for a single-task system:
      1 if task is live (not in ext or sink)
      0 if in sink (completed)
      -1 sentinel if in ext (not yet admitted)
    """
    if state.site == "sink":
        return 0
    if state.site == "ext":
        return -1   # not yet admitted; Φ not defined
    return 1

def check_conservation(parent: State, op: str, child: State) -> Optional[str]:
    """
    Property 1: No Task Loss.
    Every live transition must preserve Φ or decrease it by 1 (finish).
    Returns an error string if violated, else None.
    """
    phi_before = particle_count(parent)
    phi_after  = particle_count(child)
    if phi_before == -1:
        # Admission step: Φ goes from undefined to 1. OK.
        return None
    delta = phi_after - phi_before
    if delta == 0:
        return None   # conservation ✓
    if delta == -1 and child.site == "sink":
        return None   # O_finish: Φ decreases by 1 to 0 ✓
    return (f"CONSERVATION VIOLATED: {parent.site} →[{op}]→ {child.site}: "
            f"ΔΦ = {delta}")

def check_hop_bound(state: State, max_hops: int) -> Optional[str]:
    """
    Property 3: Livelock Freedom.
    The hop_count must never exceed max_hops.
    Returns an error string if violated, else None.
    """
    if state.hop > max_hops:
        return (f"LIVELOCK VIOLATION: hop_count={state.hop} > "
                f"max_hops={max_hops} at site {state.site}")
    return None

def check_deadlock(state: State, N: int, max_hops: int) -> Optional[str]:
    """
    Property 2: Deadlock Freedom.
    A state is deadlocked if it is non-sink and has no successors.
    (In our model this should never happen; sink is the only state
    with no successors.)
    Returns an error string if violated, else None.
    """
    if state.site == "sink":
        return None   # absorbing, expected
    succs = successors(state, N, max_hops)
    if not succs:
        return (f"DEADLOCK: state {state} has no successors "
                f"but is not the sink")
    return None

# ─────────────────────────────────────────────────────────────
# BFS exhaustive exploration
# ─────────────────────────────────────────────────────────────

def explore(N: int, verbose: bool = False) -> Dict[str, int]:
    """
    BFS over all reachable states starting from 'ext' for a system
    with N workers.  Returns a summary dict with counts.
    """
    max_hops = N // 2
    init = State(site="ext", hop=0)

    visited: set  = set()
    queue         = deque([init])
    errors: List[str] = []
    stats = {
        "states_visited": 0,
        "transitions":    0,
        "sink_reached":   0,
        "errors":         0,
        "max_hop_seen":   0,
    }

    print(f"\n{'='*60}")
    print(f"Exploring N={N}, max_hops={max_hops}")
    print(f"{'='*60}")

    while queue:
        state = queue.popleft()
        if state in visited:
            continue
        visited.add(state)
        stats["states_visited"] += 1

        # Check livelock property at this state
        err = check_hop_bound(state, max_hops)
        if err:
            errors.append(err)
            print(f"  ✗ {err}")

        # Check deadlock property at this state
        err = check_deadlock(state, N, max_hops)
        if err:
            errors.append(err)
            print(f"  ✗ {err}")

        if state.site == "sink":
            stats["sink_reached"] += 1
            if verbose:
                print(f"  [SINK reached] {state}")
            continue

        for op, child in successors(state, N, max_hops):
            stats["transitions"] += 1
            stats["max_hop_seen"] = max(stats["max_hop_seen"], child.hop)

            # Check conservation
            err = check_conservation(state, op, child)
            if err:
                errors.append(err)
                print(f"  ✗ {err}")

            if verbose:
                print(f"  {state.site}(h={state.hop}) "
                      f"--[{op}]--> {child.site}(h={child.hop})")

            if child not in visited:
                queue.append(child)

    stats["errors"] = len(errors)

    # ── Summary ─────────────────────────────────────────────────
    status = "✓ ALL PROPERTIES HOLD" if not errors else f"✗ {len(errors)} VIOLATIONS"
    print(f"\n  States visited : {stats['states_visited']}")
    print(f"  Transitions    : {stats['transitions']}")
    print(f"  Sink reached   : {stats['sink_reached']}")
    print(f"  Max hop seen   : {stats['max_hop_seen']} (≤ max_hops={max_hops})")
    print(f"  Errors         : {stats['errors']}")
    print(f"  Result         : {status}")

    return stats

# ─────────────────────────────────────────────────────────────
# Dedicated property report
# ─────────────────────────────────────────────────────────────

def lyapunov_report(N: int):
    """
    Print the Lyapunov potential V(c) = max_hops - h along each
    hop chain path, confirming strict decrease.
    """
    max_hops = N // 2
    print(f"\n  Lyapunov trace (N={N}, max_hops={max_hops}):")
    h = 0
    while h <= max_hops:
        V = max_hops - h
        fate = "→ O_hop (V decreases)" if h < max_hops else "→ O_park FORCED (V=0, absorbing)"
        print(f"    h={h:2d}  V={V:2d}  {fate}")
        h += 1

# ─────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="DTA-V3 FSM verifier")
    parser.add_argument("--verbose", action="store_true",
                        help="Print every state transition")
    args = parser.parse_args()

    print("DTA-V3 Scheduler FSM Finite Verifier")
    print("Formal model: main.tex, chapter 'Formal Mathematical Model'")
    print("Properties checked:")
    print("  [1] No Task Loss     — Φ conserved on every live transition")
    print("  [2] Deadlock Freedom — every non-sink state has ≥1 successor")
    print("  [3] Livelock Freedom — hop_count never exceeds max_hops")

    all_ok = True
    for N in [2, 3, 4, 8]:
        stats = explore(N, verbose=args.verbose)
        lyapunov_report(N)
        if stats["errors"] > 0:
            all_ok = False

    print("\n" + "="*60)
    if all_ok:
        print("RESULT: All three correctness properties verified for N ∈ {2,3,4,8}.")
        print("  — Φ is conserved on every live transition (no task loss).")
        print("  — Every non-sink state has at least one successor (no deadlock).")
        print("  — hop_count never exceeds max_hops (livelock freedom / Lyapunov).")
    else:
        print("RESULT: One or more violations found. See output above.")
        sys.exit(1)

if __name__ == "__main__":
    main()