feat: runnable cascade demo over production-shaped logs

.gitignore

@@ -1,4 +1,6 @@
 temp/
 bin/
 coverage.out
-.claude/
+prod.log
+classified.jsonl
+examples/review.jsonl

Makefile

@@ -68,4 +68,4 @@ clean:
 
 ## bench: run benchmarks across workspace modules
 bench:
-	go test -bench=. -benchmem -run=^$$ ./shared/pkg/stage/rules/... ./shared/pkg/engine/... ./shared/pkg/tokenizer/wordpiece/... ./shared/pkg/sink/jsonl/... ./shared/pkg/source/textfile/... ./shared/pkg/source/jsonl/...
+	go test -bench=. -benchmem -run=^$$ ./shared/pkg/stage/rules/... ./shared/pkg/stage/schema/... ./shared/pkg/engine/... ./shared/pkg/classifier/... ./shared/pkg/logfields/... ./shared/pkg/tokenizer/wordpiece/... ./shared/pkg/sink/jsonl/... ./shared/pkg/source/textfile/... ./shared/pkg/source/jsonl/...

README.md

@@ -18,7 +18,7 @@ source -> [ stage 1: rules ] -> [ stage 2: model ] -> [ stage N ] -> sink
                └── exit early ───────┴── exit early    uncertain -> review sink
 ```
 
-Everything is an interface: input sources, classification stages, and output sinks are pluggable modules. Today: text files and stdin. Tomorrow: whatever implements `Source`.
+Everything is an interface: input sources, classification stages, and output sinks are pluggable modules. Today: text files, JSONL event streams, and stdin. Tomorrow: whatever implements `Source`.
 
 For working examples of each contract, see [`textfile`](shared/pkg/source/textfile) (a `Source`), [`rules`](shared/pkg/stage/rules) (a `Stage`), and [`jsonl`](shared/pkg/sink/jsonl) (a `Sink`). [`docs/architecture.md`](docs/architecture.md) explains how the pieces fit and why the core stays payload-agnostic.
 
@@ -44,14 +44,34 @@ Each classified record is one JSON object on stdout with its category,
 confidence, the matched rule's reason, and (for events) the decoded fields.
 Records no rule matches are counted and reported on stderr.
 
+The full multi-stage cascade runs from a fresh clone too. `genprod.go` writes
+production-shaped logs (access lines, probes, app chatter, payments, auth
+events, warns, errors, and a slice nothing recognizes; deterministic by seed),
+and `classmesh.yaml` declares a two-tier cascade over them: rules first, a
+gated model stand-in for the leftovers, review for what neither tier can
+decide, health-check noise dropped by route:
+
+```
+go run examples/genprod.go -n 1000000 > prod.log
+classmesh validate --config examples/classmesh.yaml
+classmesh run --config examples/classmesh.yaml prod.log > classified.jsonl
+```
+
+With the default seed, the million lines classify in under two seconds:
+the rules tier decides 88%, the model tier 6%, and 6% lands in
+`examples/review.jsonl`. The stderr stats line reads
+`processed=1000000 classified=940162 review=59838 by_stage=map[model:60009 rules:880153]`;
+classified counts the health-check records the noise route then discards, so
+`classified.jsonl` holds 720,490 lines.
+
 ### Cascade config
 
 A whole multi-stage cascade can be declared in one versioned YAML file, checked
 with `validate` and run with `run --config`:
 
 ```
-classmesh validate --config classmesh.yaml          # parse + validate only
-classmesh run --config classmesh.yaml app.log       # build and run it
+classmesh validate --config examples/classmesh.yaml   # parse + validate only
+classmesh run --config examples/classmesh.yaml app.log # build and run it
 ```
 
 ```yaml
@@ -84,10 +104,11 @@ Measured on a single core (AMD Ryzen 7 3800X, `make bench`):
 |---|---|---|---|
 | Rules stage, first-rule hit | 46 ns | ~22M records/sec | 0 |
 | Rules stage, worst case (20-rule walk, regex-heavy) | ~6-7 µs | ~150k records/sec | 0 |
-| Full pipeline (engine + rules + sink) | ~500 ns | ~2M records/sec | 0 |
+| Full pipeline (engine + rules, discard sink) | ~500 ns | ~2M records/sec | 0 |
 
 Per-record cost depends on your ruleset: order rules by expected volume so the
-hot path exits early.
+hot path exits early. The pipeline row isolates engine + rules behind a discard
+sink; the JSONL output path is benchmarked separately in `sink/jsonl`.
 
 The comparison that motivates the cascade: classifying 1M short log lines with
 a budget LLM API (~25 input + 5 output tokens each at $0.15/$0.60 per million

docs/architecture.md

@@ -33,8 +33,9 @@ logs, events, and structured records without a parallel type per payload.
 
 A pipeline is assembled from four interfaces, each in its own package:
 
-- `source.Source` yields records until it is drained. Text files and stdin
-  implement it today; a CSV reader or a network stream could implement it next.
+- `source.Source` yields records until it is drained. Text files, JSONL
+  streams, and stdin implement it today; a CSV reader or a network stream could
+  implement it next.
 - `stage.Stage` classifies a record or reports `ErrUnclassified`. Stages range
   from deterministic rule matching to in-process models to remote calls.
 - `sink.Sink` consumes a record together with its classification: stdout, a

examples/classmesh.yaml

@@ -0,0 +1,11 @@
+# Production-log cascade demo: rules tier -> gated model stand-in -> review,
+# noise dropped by route. Commands: README, Examples section.
+version: 1
+input: { type: text }
+stages:
+  - { id: rules, type: rules, path: prod-rules.yml, gate: 1.0 }
+  - { id: model, type: mock, path: mock.yml, gate: 0.7 }
+sink: { type: jsonl, stream: stdout }
+review: { type: jsonl, path: review.jsonl }
+routes:
+  noise: { type: drop }

examples/genprod.go

@@ -0,0 +1,119 @@
+//go:build ignore
+
+// genprod writes n production-shaped log lines to stdout (deterministic by
+// seed): access logs, probes, app chatter, payments, auth events, warns,
+// errors, and a slice nothing matches. go run examples/genprod.go -n 1000000
+package main
+
+import (
+	"bufio"
+	"flag"
+	"fmt"
+	"math/rand"
+	"os"
+	"strings"
+	"time"
+)
+
+func main() {
+	n := flag.Int("n", 1_000_000, "number of log lines to generate")
+	seed := flag.Int64("seed", 42, "PRNG seed; same seed, same output")
+	flag.Parse()
+
+	w := bufio.NewWriterSize(os.Stdout, 1<<20)
+	r := rand.New(rand.NewSource(*seed))
+	ts := time.Date(2026, 7, 10, 8, 0, 0, 0, time.UTC)
+
+	paths := []string{"/api/v1/orders/%d", "/api/v1/users/%d", "/api/v1/search?q=q%d", "/api/v1/carts/%d/items", "/assets/app.js", "/login"}
+	methods := []string{"GET", "GET", "GET", "POST", "PUT"}
+	services := []string{"gateway", "orders", "users", "payments", "search", "carts"}
+	infos := []string{
+		"info service=%s cache miss key=user:%d refill_ms=%d",
+		"info service=%s job completed name=reindex-%d duration_ms=%d",
+		"info service=%s connection pool size=%d idle=%d",
+	}
+	odd := []string{
+		"tick %d drift %dms",
+		"peer gossip state=stable nodes=%d epoch=%d",
+		"compaction window advanced segment=%d ratio=0.%d",
+	}
+
+	for i := 0; i < *n; i++ {
+		ts = ts.Add(time.Duration(r.Intn(4)) * time.Millisecond)
+		stamp := ts.Format(time.RFC3339Nano)
+		svc := services[r.Intn(len(services))]
+		draw := r.Float64()
+		switch {
+		case draw < 0.28: // http access, 2xx
+			fmt.Fprintf(w, "%s info service=%s http request method=%s path=%s status=%d latency_ms=%d ip=10.%d.%d.%d\n",
+				stamp, svc, methods[r.Intn(len(methods))], pick(r, paths), 200+[]int{0, 0, 0, 1, 4}[r.Intn(5)], 1+r.Intn(180), r.Intn(256), r.Intn(256), 1+r.Intn(254))
+		case draw < 0.50: // probes: pure noise
+			if r.Intn(3) == 0 {
+				fmt.Fprintf(w, "GET /readiness 200\n")
+			} else {
+				fmt.Fprintf(w, "GET /healthz 200\n")
+			}
+		case draw < 0.62: // app info chatter
+			f := infos[r.Intn(len(infos))]
+			fmt.Fprintf(w, "%s "+f+"\n", stamp, svc, r.Intn(100000), r.Intn(900))
+		case draw < 0.70: // payments
+			verb := []string{"succeeded", "succeeded", "declined", "failed"}[r.Intn(4)]
+			fmt.Fprintf(w, "%s info service=payments payment %s for order %d amount=%d.%02d currency=USD\n",
+				stamp, verb, r.Intn(200000), 1+r.Intn(400), r.Intn(100))
+		case draw < 0.76: // auth signals
+			switch r.Intn(3) {
+			case 0:
+				fmt.Fprintf(w, "%s blocked: unauthorized token for user %d\n", stamp, r.Intn(9000))
+			case 1:
+				fmt.Fprintf(w, "%s login failed user=u%d attempts=%d\n", stamp, r.Intn(9000), 1+r.Intn(5))
+			default:
+				fmt.Fprintf(w, "%s rate limited ip=10.%d.%d.%d window=60s\n", stamp, r.Intn(256), r.Intn(256), 1+r.Intn(254))
+			}
+		case draw < 0.82: // http client/server errors
+			fmt.Fprintf(w, "%s info service=%s http request method=%s path=%s status=%d latency_ms=%d ip=10.%d.%d.%d\n",
+				stamp, svc, methods[r.Intn(len(methods))], pick(r, paths), []int{404, 404, 429, 500, 503}[r.Intn(5)], 1+r.Intn(900), r.Intn(256), r.Intn(256), 1+r.Intn(254))
+		case draw < 0.88: // warns: the model tier's job
+			switch r.Intn(3) {
+			case 0:
+				fmt.Fprintf(w, "%s warn service=%s slow query duration_ms=%d table=orders_%d\n", stamp, svc, 800+r.Intn(2400), 1+r.Intn(40))
+			case 1:
+				fmt.Fprintf(w, "%s warn service=%s queue depth=%d threshold=500\n", stamp, svc, 500+r.Intn(2000))
+			default:
+				fmt.Fprintf(w, "%s warn service=%s retrying request attempt=%d upstream=inventory\n", stamp, svc, 2+r.Intn(4))
+			}
+		case draw < 0.94: // severe errors
+			switch r.Intn(4) {
+			case 0:
+				fmt.Fprintf(w, "%s error service=%s upstream timeout after %dms trace=%08x\n", stamp, svc, 1000+r.Intn(4000), r.Uint32())
+			case 1:
+				fmt.Fprintf(w, "%s error service=%s request failed status=503 trace=%08x attempt=%d\n", stamp, svc, r.Uint32(), 1+r.Intn(3))
+			case 2:
+				fmt.Fprintf(w, "%s error service=%s panic recovered in handler trace=%08x goroutine=%d\n", stamp, svc, r.Uint32(), r.Intn(90000))
+			default:
+				fmt.Fprintf(w, "%s OOMKilled container=%s-%d restart_count=%d\n", stamp, svc, r.Intn(40), 1+r.Intn(6))
+			}
+		case draw < 0.97: // business events: matched low-confidence by the model tier
+			if r.Intn(2) == 0 {
+				fmt.Fprintf(w, "%s shipment %d dispatched from tlv warehouse=%c\n", stamp, r.Intn(999999), 'A'+rune(r.Intn(4)))
+			} else {
+				fmt.Fprintf(w, "%s inventory sync started region=eu-%d items=%d\n", stamp, 1+r.Intn(3), r.Intn(50000))
+			}
+		default: // chatter no tier recognizes: exercises the review path
+			f := odd[r.Intn(len(odd))]
+			fmt.Fprintf(w, "%s "+f+"\n", stamp, r.Intn(100000), 1+r.Intn(500))
+		}
+	}
+	if err := w.Flush(); err != nil {
+		fmt.Fprintf(os.Stderr, "genprod: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+// pick returns a path template with its id filled in when it has one.
+func pick(r *rand.Rand, paths []string) string {
+	p := paths[r.Intn(len(paths))]
+	if strings.Contains(p, "%d") {
+		return fmt.Sprintf(p, r.Intn(90000))
+	}
+	return p
+}

examples/mock.yml

@@ -0,0 +1,13 @@
+# Model-tier stand-in: sub-1.0 confidences exercise the 0.7 gate in classmesh.yaml.
+matchers:
+  - contains: ["warn "]
+    category: anomaly
+    confidence: 0.85
+
+  - contains: ["shipment ", "inventory "]
+    category: logistics
+    confidence: 0.6
+
+default:
+  category: unknown
+  confidence: 0.4

examples/prod-rules.yml

@@ -0,0 +1,32 @@
+# Ruleset for the genprod demo. Ordered by volume; first match wins.
+rules:
+  - id: health-noise
+    category: noise
+    contains: ["/healthz", "/readiness"]
+
+  - id: access-ok
+    category: traffic
+    contains: ["status=200", "status=201", "status=204"]
+
+  - id: app-ops
+    category: ops
+    contains: ["cache miss", "job completed", "connection pool"]
+
+  - id: payment-event
+    category: billing
+    regex: ["payment (succeeded|declined|failed)"]
+
+  - id: auth-signal
+    category: auth
+    any:
+      - contains: "unauthorized"
+      - contains: "login failed"
+      - contains: "rate limited"
+
+  - id: alert
+    category: alert
+    any:
+      - regex: "status=[45][0-9][0-9]"
+      - contains: "error "
+      - contains: "panic"
+      - contains: "OOMKilled"

services/cli/internal/app/example_config_test.go

@@ -0,0 +1,114 @@
+package app
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// The committed cascade example must stay runnable from a fresh clone.
+func TestExampleCascadeConfig(t *testing.T) {
+	t.Run("validate", func(t *testing.T) {
+		var out, errOut bytes.Buffer
+		err := Run(context.Background(), []string{"validate", "--config", examplePath(t, "classmesh.yaml")},
+			Streams{In: strings.NewReader(""), Out: &out, Err: &errOut})
+		if err != nil {
+			t.Fatalf("Run() error = %v, stderr=%s", err, errOut.String())
+		}
+		if !strings.Contains(out.String(), "structurally valid") {
+			t.Fatalf("stdout = %q, want a validity confirmation", out.String())
+		}
+	})
+
+	t.Run("run", func(t *testing.T) {
+		dir := t.TempDir()
+		for _, name := range []string{"classmesh.yaml", "prod-rules.yml", "mock.yml"} {
+			data, err := os.ReadFile(examplePath(t, name))
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := os.WriteFile(filepath.Join(dir, name), data, 0o644); err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		tests := []struct {
+			line     string
+			category string // expected category for classified records
+			stage    string // expected deciding stage
+			drop     bool   // routed to drop: classified but absent from stdout
+			review   bool   // neither tier decided: lands in review.jsonl
+		}{
+			{line: "GET /healthz 200", category: "noise", drop: true},
+			{line: "2026-07-10T08:00:00Z info service=gateway http request method=GET path=/api/v1/orders/1 status=200 latency_ms=5 ip=10.0.0.1", category: "traffic", stage: "rules"},
+			{line: "2026-07-10T08:00:00Z info service=payments payment declined for order 7 amount=10.00 currency=USD", category: "billing", stage: "rules"},
+			{line: "2026-07-10T08:00:01Z blocked: unauthorized token for user 3", category: "auth", stage: "rules"},
+			{line: "2026-07-10T08:00:02Z error service=api upstream timeout after 3000ms trace=00000001", category: "alert", stage: "rules"},
+			{line: "2026-07-10T08:00:03Z warn service=db slow query duration_ms=1900 table=orders_7", category: "anomaly", stage: "model"},
+			{line: "2026-07-10T08:00:04Z shipment 84712 dispatched from tlv warehouse=A", review: true},
+			{line: "tick 8912 drift 3ms", review: true},
+		}
+
+		var input strings.Builder
+		for _, tc := range tests {
+			input.WriteString(tc.line)
+			input.WriteByte('\n')
+		}
+		inPath := filepath.Join(dir, "input.log")
+		if err := os.WriteFile(inPath, []byte(input.String()), 0o644); err != nil {
+			t.Fatal(err)
+		}
+
+		var out, errOut bytes.Buffer
+		err := Run(context.Background(), []string{"run", "--config", filepath.Join(dir, "classmesh.yaml"), inPath},
+			Streams{In: strings.NewReader(""), Out: &out, Err: &errOut})
+		if err != nil {
+			t.Fatalf("Run() error = %v, stderr=%s", err, errOut.String())
+		}
+		if !strings.Contains(errOut.String(), "processed=8 classified=6 review=2") {
+			t.Fatalf("stats = %q, want processed=8 classified=6 review=2", errOut.String())
+		}
+
+		reviewData, err := os.ReadFile(filepath.Join(dir, "review.jsonl"))
+		if err != nil {
+			t.Fatalf("review sink: %v", err)
+		}
+		stdout := out.String()
+		for _, tc := range tests {
+			switch {
+			case tc.drop:
+				if strings.Contains(stdout, tc.line) {
+					t.Errorf("dropped line %q reached stdout", tc.line)
+				}
+			case tc.review:
+				if !strings.Contains(string(reviewData), tc.line) {
+					t.Errorf("line %q missing from review.jsonl", tc.line)
+				}
+				if strings.Contains(stdout, tc.line) {
+					t.Errorf("review line %q reached stdout", tc.line)
+				}
+			default:
+				rec := ""
+				for _, r := range strings.Split(strings.TrimSpace(stdout), "\n") {
+					if strings.Contains(r, tc.line) {
+						rec = r
+						break
+					}
+				}
+				if rec == "" {
+					t.Errorf("line %q missing from stdout", tc.line)
+					continue
+				}
+				if !strings.Contains(rec, `"category":"`+tc.category+`"`) {
+					t.Errorf("line %q: record %s, want category %q", tc.line, rec, tc.category)
+				}
+				if !strings.Contains(rec, `"stage":"`+tc.stage+`"`) {
+					t.Errorf("line %q: record %s, want stage %q", tc.line, rec, tc.stage)
+				}
+			}
+		}
+	})
+}

shared/pkg/tokenizer/wordpiece/wordpiece.go

@@ -1,13 +1,13 @@
 // Package wordpiece implements BERT-style WordPiece tokenization in pure Go,
 // with no cgo and no external tokenizer runtime. It turns text into the token
-// IDs an ONNX BERT/MiniLM classifier expects: basic tokenization (Unicode
+// IDs a BERT-vocabulary model expects: basic tokenization (Unicode
 // cleanup, optional lowercasing and accent stripping, punctuation and CJK
 // splitting) followed by greedy longest-match-first subword segmentation
 // against a fixed vocabulary, wrapped with [CLS] and [SEP].
 //
-// It exists so the in-process model stage can stay a single static binary
-// (CGO_ENABLED=0): the common alternative, the rust tokenizers library, needs
-// cgo and a statically linked archive. We own this instead.
+// It is intended to feed a future in-process model stage so the binary can
+// stay CGO_ENABLED=0; no such stage is wired yet. The common alternative, the
+// rust tokenizers library, needs cgo and a statically linked archive.
 package wordpiece
 
 import (