lean.tmpl

{{define "main" -}}
-- Code generated by tools/codegen from model.json; DO NOT EDIT.
--
-- Proof form: the net is a claim about every reachable marking, and the claim
-- is discharged before the program is allowed to exist. See FORMS.md; output
-- must match parity/trace.golden.
--
-- The marking here is a *multiset* — a count per place — and enablement is
-- plain P/T semantics with no `capacity: 1` clause. Every theorem below is
-- checked by `decide`: the kernel evaluates the breadth-first search over the
-- reachable state space ({{.ReachCount}} markings) during elaboration. Make
-- the net unsafe — say, add a second arc producing into a marked place — and
-- this file stops compiling. The other languages' proof forms run the same
-- check at startup; Lean runs it before `main` can be defined.
--
-- tools/codegen derives the theorems from model.json: 1-safety (every place
-- has capacity 1), one conservation theorem per basis vector of the left
-- nullspace of the incidence matrix C (yᵀC = 0, C = places × transitions),
-- and — because the net is conflict-free — the unique deadlock marking, which
-- codegen found by its own BFS under the canonical semantics and Lean
-- re-derives here under P/T semantics.

inductive State where
{{- range .Places}}
  | {{.}}
{{- end}}
  deriving DecidableEq, Repr

def State.name : State → String
{{- range .Places}}
  | .{{.}} => "{{.}}"
{{- end}}

-- Every place, sorted by name. Markings index into this list by position.
def State.all : List State :=
  [{{range $i, $p := .Places}}{{if $i}}, {{end}}.{{$p}}{{end}}]

-- The multiset marking: how many tokens each place holds. Association by
-- position into State.all keeps it a plain List Nat, which is what lets
-- `decide` evaluate everything below inside the kernel.
abbrev Marking := List Nat

def indexOf (p : State) : Nat :=
  (State.all.findIdx? (· == p)).getD 0

def count (m : Marking) (p : State) : Nat :=
  m.getD (indexOf p) 0

def initialMarking : Marking :=
  State.all.map fun p =>
    if [{{range $i, $p := .Initial}}{{if $i}}, {{end}}State.{{$p}}{{end}}].contains p then 1 else 0

-- One transition: inputs consumed, outputs produced, guards read but not
-- consumed, blockers read and required empty. The same shape the contract
-- form's entry points take.
structure Transition where
  name     : String
  inputs   : List State
  outputs  : List State
  guards   : List State
  blockers : List State

-- The net, in the scheduler's candidate order: sorted by transition name. This
-- is what implements the canonical scheduling policy (FORMS.md) — fire the
-- enabled transition whose name is lexicographically least. codegen emits the
-- list sorted, so the policy is enforced by the generator rather than by hand.
def transitions : List Transition :=
  [{{range $i, $t := .Transitions}}{{if $i}},
   {{end}} ⟨"{{$t.Name}}", [{{range $j, $p := $t.Inputs}}{{if $j}}, {{end}}.{{$p}}{{end}}], [{{range $j, $p := $t.Outputs}}{{if $j}}, {{end}}.{{$p}}{{end}}], [{{range $j, $p := $t.Guards}}{{if $j}}, {{end}}.{{$p}}{{end}}], [{{range $j, $p := $t.Blockers}}{{if $j}}, {{end}}.{{$p}}{{end}}]⟩{{end}} ]

-- Plain P/T enablement: inputs and guards marked, blockers empty. Deliberately
-- no "outputs clear" clause — that the clause is never needed is theorem 1.
def enabled (m : Marking) (t : Transition) : Bool :=
  t.inputs.all (fun p => count m p ≥ 1)
    && t.guards.all (fun p => count m p ≥ 1)
    && t.blockers.all (fun p => count m p == 0)

def fireCounts (m : Marking) (t : Transition) : Marking :=
  State.all.map fun p =>
    count m p - (if t.inputs.contains p then 1 else 0)
              + (if t.outputs.contains p then 1 else 0)

def successors (m : Marking) : List Marking :=
  (transitions.filter (enabled m ·)).map (fireCounts m)

-- Breadth-first search over the reachable state space. The frontier bound is
-- 2^{{len .Places}}: past it, some marking must exceed one token, and theorem 1 has
-- already failed. This net reaches {{.ReachCount}} markings; the fuel is never the limit.
def explore : Nat → List Marking → List Marking → List Marking
  | 0, seen, _ => seen
  | _ + 1, seen, [] => seen
  | fuel + 1, seen, m :: frontier =>
      let fresh := (successors m).filter (fun n => !seen.contains n)
      explore fuel (seen ++ fresh) (frontier ++ fresh)

def reachable : List Marking :=
  explore {{.Fuel}} [initialMarking] [initialMarking]

def oneSafe (m : Marking) : Bool :=
  m.all (· ≤ 1)

def isDeadlock (m : Marking) : Bool :=
  transitions.all (fun t => !enabled m t)

-- The BFS closed: every successor of a reachable marking was reached, so the
-- theorems below quantify over the whole state space, not a fuel-truncated
-- prefix of it.
theorem reachable_closed :
    reachable.all (fun m => (successors m).all (reachable.contains ·)) = true := by
  decide
-- Theorem 1: 1-safety is structural (the loader admits only capacity-1 models,
-- so this theorem is stated for every generated net). No reachable marking holds two tokens
-- anywhere, even though nothing above stops a transition firing into a marked
-- place. This is what licenses the set representation every other form uses.
theorem reachable_one_safe : reachable.all oneSafe = true := by decide

-- P-invariants. Each weight vector y below satisfies yᵀC = 0 for the incidence
-- matrix C, so Σ y[p]·M[p] is the same in every reachable marking as in the
-- initial one. codegen computed the basis by row-reducing Cᵀ over ℚ; Lean does
-- not trust that computation — it re-checks the consequence on every reachable
-- marking. Weights are Int because a net that destroys tokens has invariants
-- with negative entries.
def weighted (w : List Int) (m : Marking) : Int :=
  (List.zipWith (fun (wi : Int) (c : Nat) => wi * (c : Int)) w m).sum
{{range .Invariants}}
-- Invariant {{.Index}}: {{join .Terms " "}} = {{.Initial}}
def invariant{{.Index}} : List Int := [{{range $i, $w := .Weights}}{{if $i}}, {{end}}{{$w}}{{end}}]

theorem invariant{{.Index}}_conserved :
    reachable.all (fun m => weighted invariant{{.Index}} m == {{.Initial}}) = true := by
  decide
{{end}}
{{- if .Final}}
def finalMarking : Marking :=
  State.all.map fun p => if [{{range $i, $p := .Final}}{{if $i}}, {{end}}State.{{$p}}{{end}}].contains p then 1 else 0

-- Theorem 2: the outcome is confluent. The net is conflict-free — no two
-- transitions consume the same place — so every maximal firing sequence stops
-- at the same marking, and it is {{"{"}}{{join .Final ","}}{{"}"}}. The golden
-- trace's final line is the only place the net can stop.
theorem unique_deadlock :
    (reachable.filter isDeadlock) = [finalMarking] := by decide
{{else}}
-- The net is not conflict-free, or has more than one terminal marking
-- ({{len .Deadlocks}} found), so no confluence theorem is stated.
{{end}}
-- What remains is the interpreter form's scheduler, running on a net the
-- compiler has already checked.

def insertSorted (x : String) : List String → List String
  | [] => [x]
  | y :: ys => if x < y then x :: y :: ys else y :: insertSorted x ys

def sortStrings (xs : List String) : List String :=
  xs.foldr insertSorted []

-- Canonical trace rendering: marked places, sorted by name, comma-joined.
def render (m : Marking) : String :=
  String.intercalate "," (sortStrings ((State.all.filter (count m · ≥ 1)).map State.name))

partial def executeProcess (step : Nat) (m : Marking) : IO Unit := do
  match transitions.find? (enabled m ·) with
  | none => pure ()
  | some t =>
      let next := fireCounts m t
      IO.println s!"Step #{step + 1}: {t.name} => {render next}"
      executeProcess (step + 1) next

def main : IO Unit :=
  executeProcess 0 initialMarking
{{end}}