Pipeline.lean

View source on GitHub

Step

An enumeration of the six roles. Five are forward stages that process data sequentially. Consolidate is the backward pass that updates policy inside the substrate.

inductive Step where
  | perceive | cache | filter
  | attend | consolidate | remember

InterfaceTypes

The seven types that flow between pipeline stages: raw input, encoded, indexed, selected, ranked, policy, and persisted. Each stage consumes one type and produces the next.

structure InterfaceTypes where
  raw : Type        -- environment input
  encoded : Type    -- after Perceive
  indexed : Type    -- after Cache
  selected : Type   -- after Filter
  ranked : Type     -- after Attend
  policy : Type     -- parameterizes Attend; written by Consolidate
  persisted : Type  -- after Remember (feeds back to Perceive)

Pipeline

A complete pipeline bundles five forward kernels and one backward kernel. All are monadic — when the monad is Id, everything is deterministic. Attend takes a policy parameter; Consolidate reads persisted state and updates policy.

structure Pipeline (M : Type → Type) [Monad M] (I : InterfaceTypes) where
  perceive    : Kernel M I.raw I.encoded
  cache       : Kernel M I.encoded I.indexed
  filter      : Kernel M I.indexed I.selected
  attend      : I.policy → Kernel M I.selected I.ranked
  remember    : Kernel M I.ranked I.persisted
  consolidate : I.policy → Kernel M I.persisted I.policy

; The six-stage pipeline as function composition
; Each stage: one type in, one type out

(define (perceive raw)
  (string-append "enc:" raw))

(define (cache encoded)
  (list encoded (string-append encoded "#2")))

(define (filter indexed)
  (list (car indexed)))  ; keep only first

(define (attend policy selected)
  (map (lambda (s) (string-append s "|" policy))
       selected))

(define (remember ranked) ranked)  ; lossless

; Forward pass: compose perceive through attend
(define (forward raw policy)
  (attend policy
    (filter (cache (perceive raw)))))

(display "forward('hello', 'top-1'): ")
(display (forward "hello" "top-1"))
(newline)
(display "forward('hello', 'diverse'): ")
(display (forward "hello" "diverse"))

# The six-stage pipeline as function composition
# Each stage: one type in, one type out

def perceive(raw):
    return "enc:" + raw

def cache(encoded):
    return [encoded, encoded + "#2"]

def filter_stage(indexed):
    return [indexed[0]]  # keep only first

def attend(policy, selected):
    return [s + "|" + policy for s in selected]

def remember(ranked):
    return ranked  # lossless

# Forward pass: compose perceive through attend
def forward(raw, policy):
    return attend(policy, filter_stage(cache(perceive(raw))))

print(f"forward('hello', 'top-1'): {forward('hello', 'top-1')}")
print(f"forward('hello', 'diverse'): {forward('hello', 'diverse')}")

Pipeline.forward

The forward data path: raw input flows through Perceive, Cache, Filter, and Attend (parameterized by current policy), producing ranked output. Written as explicit binds for proof tractability (easier for Lean to unfold).

Pipeline.cycle

One full cycle: run the forward pass, persist the result via Remember, then run Consolidate to derive updated policy. Returns the new policy and persisted state as a pair.

Pipeline.iterate

Run the pipeline n times over a stream of inputs. Each iteration chains via bind, threading the policy through cycles.

Connections

• Kleisli.lean — imported; every pipeline stage is a Kernel
• Contracts.lean — attaches postconditions to each pipeline stage
• Handshake.lean — proves the stages compose correctly via handshake conditions
• Uniqueness.lean — proves this ordering is the only valid one