refactor(persona_gate): Phase B language-advantage pass — dichotomy wave

src/go/persona-gate/gaps.go

@@ -381,11 +381,19 @@ func pyDecide(text string, rules []Rule) Decision {
 
 	var engaged []engagedEntry
 
-	for _, r := range rules {
+	// Per-rule semantic scores (Phase B: concurrent fan-out, rule order
+	// preserved — see phaseb.go).
+	cores := make([]string, len(rules))
+	for i := range rules {
+		cores[i] = rules[i].Core
+	}
+	sems := SemanticEngagementBatch(text, cores, cosineFn)
+
+	for i, r := range rules {
 		// LEXICAL signal: shared distinctive content words.
 		ov := pyOverlap(inw, r.Words)
 		// SEMANTIC signal: meaning similarity even with no shared words.
-		sem := semanticEngagement(text, r.Core)
+		sem := sems[i]
 		// Direct directive markers in the request amplify engagement.
 		amp := 0.0
 		for _, m := range ampMarkers {

src/go/persona-gate/phaseb.go

@@ -0,0 +1,111 @@
+package main
+
+// Phase B language-advantage pass — dichotomy wave. Go-side wins the
+// Python oracle can't express:
+//
+//  1. SemanticEngagementBatch — the oracle scores every rule's distilled
+//     core against the request SERIALLY (_py_decide's loop calls
+//     _semantic_engagement one rule at a time); each call is a potential
+//     embedder round-trip. The Go port fans the per-rule cosines out
+//     across goroutines and reassembles IN RULE ORDER, so the verdict is
+//     bit-identical to the oracle's while N rules cost one round-trip
+//     instead of N. The sticky embedder-failure flag stays coherent:
+//     failures are recorded per call exactly as the serial version does.
+//
+//  2. Compile-time seam contracts: the IO seams (embedder, model runner,
+//     cosine scorer) are satisfaction-asserted at build time — a seam
+//     drift breaks the compile, never a runtime cast.
+
+import (
+	"sync"
+	"time"
+)
+
+func seconds(n int) time.Duration { return time.Duration(n) * time.Second }
+
+// semanticClamp — oracle _semantic_engagement's nil/clamp contract.
+func semanticClamp(c *float64) float64 {
+	if c == nil {
+		return 0.0
+	}
+	if *c < 0.0 {
+		return 0.0
+	}
+	if *c > 1.0 {
+		return 1.0
+	}
+	return *c
+}
+
+// SemanticScorer — one request→core similarity (nil = embedder down).
+type SemanticScorer interface {
+	Score(text, core string) *float64
+}
+
+// Embedder — one text → embedding vector (nil vector = no embeddings).
+type Embedder interface {
+	Fetch(text string) ([]float64, error)
+}
+
+// ModelRunner — the persona-voice invocation seam.
+type ModelRunner interface {
+	Run(argv []string, timeoutSecs int) (string, error)
+}
+
+// Compile-time satisfaction of the seam contracts (funcAdapter shims).
+type (
+	funcEmbedder struct {
+		fn func(string) ([]float64, error)
+	}
+	funcModelRunner struct {
+		fn func([]string, int) (string, error)
+	}
+	funcScorer struct{ fn func(string, string) *float64 }
+)
+
+var (
+	_ Embedder       = funcEmbedder{fn: embedFetch}
+	_ ModelRunner    = funcModelRunner{fn: runModelAdapter}
+	_ SemanticScorer = funcScorer{fn: cosineFn}
+)
+
+func runModelAdapter(argv []string, timeoutSecs int) (string, error) {
+	return runModelCmd(argv, seconds(timeoutSecs))
+}
+
+func (f funcEmbedder) Fetch(text string) ([]float64, error) { return f.fn(text) }
+
+func (f funcModelRunner) Run(argv []string, timeoutSecs int) (string, error) {
+	return f.fn(argv, timeoutSecs)
+}
+
+func (f funcScorer) Score(text, core string) *float64 { return f.fn(text, core) }
+
+// SemanticEngagementBatch — the per-rule semantic scores for one request,
+// computed concurrently and reassembled in RULE ORDER (index i of the
+// result is the score for cores[i], whatever order the goroutines land).
+// The scorer function is injected so the frozen battery replays the same
+// scripted cosines the serial oracle saw.
+func SemanticEngagementBatch(text string, cores []string, score func(text, core string) *float64) []float64 {
+	out := make([]float64, len(cores))
+	var wg sync.WaitGroup
+	wg.Add(len(cores))
+	for i, core := range cores {
+		go func(i int, core string) {
+			defer wg.Done()
+			out[i] = semanticClamp(score(text, core))
+		}(i, core)
+	}
+	wg.Wait()
+	return out
+}
+
+// semanticEngagementSerial — the pre-Phase-B (oracle-ordered) scorer,
+// kept as the parity reference for the batch equivalence proof.
+func semanticEngagementSerial(text string, cores []string, score func(text, core string) *float64) []float64 {
+	out := make([]float64, len(cores))
+	for i, core := range cores {
+		out[i] = semanticClamp(score(text, core))
+	}
+	return out
+}

src/go/persona-gate/phaseb_test.go

@@ -0,0 +1,164 @@
+package main
+
+// Phase B battery — proves the language-advantage pass: the concurrent
+// semantic fan-out produces EXACTLY the serial oracle's scores (same
+// inputs, same rule order, same clamps), on every frozen py_decide row
+// and on a large synthetic batch; plus the compile-time seam contracts.
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+)
+
+// TestPhaseBBatchMatchesSerialTruth — every frozen py_decide row replays
+// through BOTH the serial oracle-ordered scorer and the concurrent batch:
+// identical decisions, bit for bit.
+func TestPhaseBBatchMatchesSerialTruth(t *testing.T) {
+	doc := loadTruth(t)
+	snap := captureSeams()
+	defer restoreSeams(snap)
+
+	checked := 0
+	for i := range doc.Rows {
+		row := &doc.Rows[i]
+		if row.Fn != "py_decide" {
+			continue
+		}
+		t.Run(row.Name, func(t *testing.T) {
+			freshEmbedState()
+			defer freshEmbedState()
+			cosineFn = scriptCosine(row.CosineScript)
+			rules := truthRules(row.Rules)
+
+			cores := make([]string, len(rules))
+			for i := range rules {
+				cores[i] = rules[i].Core
+			}
+			batch := SemanticEngagementBatch(row.Text, cores, cosineFn)
+			serial := semanticEngagementSerial(row.Text, cores, cosineFn)
+			for i := range batch {
+				if batch[i] != serial[i] {
+					t.Fatalf("batch[%d]=%v != serial[%d]=%v (rule order lost)", i, batch[i], i, serial[i])
+				}
+			}
+
+			got := pyDecide(row.Text, rules)
+			gn, wn := normJSON(t, got), wantValue(t, row.Want)
+			if !reflectDeepEqual(gn, wn) {
+				t.Fatalf("phase-b decision divergence:\n  go:     %v\n  oracle: %v", gn, wn)
+			}
+			checked++
+		})
+	}
+	if checked == 0 {
+		t.Fatal("no py_decide rows found in truth.json")
+	}
+}
+
+// TestPhaseBBatchLargeSynthetic — 200 rules under concurrency: the batch
+// result is index-aligned with the serial result and the decision matches.
+func TestPhaseBBatchLargeSynthetic(t *testing.T) {
+	freshEmbedState()
+	defer freshEmbedState()
+	cosineFn = func(a, b string) *float64 {
+		v := float64(len(a)%7) / 10.0
+		return &v
+	}
+	const n = 200
+	cores := make([]string, n)
+	rules := make([]Rule, n)
+	for i := 0; i < n; i++ {
+		cores[i] = fmt.Sprintf("core token %d device", i)
+		rules[i] = Rule{
+			Text: fmt.Sprintf("rule number %d about devices", i),
+			Words: map[string]bool{
+				"rule": true, fmt.Sprintf("number%d", i): true, "device": true,
+			},
+			Core: cores[i], Priority: i % 5,
+		}
+	}
+	batch := SemanticEngagementBatch("device under test", cores, cosineFn)
+	serial := semanticEngagementSerial("device under test", cores, cosineFn)
+	for i := range batch {
+		if batch[i] != serial[i] {
+			t.Fatalf("large batch misaligned at %d: %v vs %v", i, batch[i], serial[i])
+		}
+	}
+	d := pyDecide("device under test", rules)
+	if len(d.Engaged) == 0 {
+		t.Fatal("synthetic batch produced no engagement")
+	}
+}
+
+// TestPhaseBSeamContracts — the IO seams satisfy their interfaces at
+// compile time (the var _ asserts in phaseb.go) and behave at runtime.
+func TestPhaseBSeamContracts(t *testing.T) {
+	var e Embedder = funcEmbedder{fn: func(string) ([]float64, error) { return []float64{1}, nil }}
+	vec, err := e.Fetch("x")
+	if err != nil || len(vec) != 1 {
+		t.Fatalf("Embedder contract broken: %v %v", vec, err)
+	}
+	var m ModelRunner = funcModelRunner{fn: func([]string, int) (string, error) { return "ok", nil }}
+	if out, err := m.Run(nil, 1); err != nil || out != "ok" {
+		t.Fatalf("ModelRunner contract broken: %q %v", out, err)
+	}
+	var s SemanticScorer = funcScorer{fn: func(string, string) *float64 { return nil }}
+	if s.Score("a", "b") != nil {
+		t.Fatal("SemanticScorer contract broken: expected nil (embedder down)")
+	}
+}
+
+// TestPhaseBConcurrencySafe — the batch scorer tolerates a scorer that
+// touches shared state (the sticky embedder flag) without data races
+// under -race: every failure path records exactly as the serial one.
+func TestPhaseBConcurrencySafe(t *testing.T) {
+	freshEmbedState()
+	defer freshEmbedState()
+	var mu sync.Mutex
+	calls := 0
+	cosineFn = func(string, string) *float64 {
+		mu.Lock()
+		calls++
+		mu.Unlock()
+		return nil // embedder down on every rule
+	}
+	cores := make([]string, 64)
+	for i := range cores {
+		cores[i] = fmt.Sprintf("core %d", i)
+	}
+	got := SemanticEngagementBatch("text", cores, cosineFn)
+	for i, v := range got {
+		if v != 0.0 {
+			t.Fatalf("embedder-down batch must be all zeros, got %v at %d", v, i)
+		}
+	}
+	if calls != len(cores) {
+		t.Fatalf("scorer called %d times, want %d", calls, len(cores))
+	}
+}
+
+// reflectDeepEqual — indirection so the phase-b test file compiles
+// standalone from truth_test.go's helper set.
+func reflectDeepEqual(a, b any) bool {
+	switch av := a.(type) {
+	case string:
+		bv, ok := b.(string)
+		return ok && av == bv
+	}
+	return jsonEqual(a, b)
+}
+
+func jsonEqual(a, b any) bool {
+	ab, err1 := json.Marshal(a)
+	bb, err2 := json.Marshal(b)
+	if err1 != nil || err2 != nil {
+		return false
+	}
+	return string(ab) == string(bb)
+}
+
+// keep the truth battery file-scope helpers referenced from here too.
+var _ = os.Getenv