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

src/go/context-broker/phaseb.go

@@ -0,0 +1,106 @@
+package main
+
+// Phase B language-advantage pass — dichotomy wave. Three Go-side wins the
+// Python oracle can't express:
+//
+//  1. Concurrent window discovery + rolling-store access under -race: the
+//     broker's registries are mutex-guarded; Go proves thread-safety that
+//     the Python module-level dicts only get from the GIL.
+//  2. EstimateAll — batch token estimation fans out across goroutines and
+//     reassembles in input order (the estimator is pure, so the batch is
+//     embarrassingly parallel; the Python loop is serial).
+//  3. A typed BlockSource interface + name-keyed admission registry with
+//     compile-time satisfaction assertions (Python dispatches through
+//     isinstance checks; Go turns the dispatch into data).
+
+import (
+	"sort"
+	"sync"
+)
+
+// ── 1. concurrent registries ────────────────────────────────────────────────
+
+// DiscoverWindowsConcurrent resolves several models' windows in parallel.
+// Each goroutine captures its own discover func (no shared-global swap);
+// registry writes stay serialized by the same mutex the single-call path
+// uses; results reassemble in input order regardless of discovery latency.
+func DiscoverWindowsConcurrent(modelIDs []string, discover func(string) int) []int {
+	out := make([]int, len(modelIDs))
+	var wg sync.WaitGroup
+	for i, id := range modelIDs {
+		wg.Add(1)
+		go func(i int, id string) {
+			defer wg.Done()
+			out[i] = resolveWindow(id, discover)
+		}(i, id)
+	}
+	wg.Wait()
+	return out
+}
+
+// RollingKeys — a consistent snapshot of every persisted session key
+// (source + session), sorted. Reads under the rolling lock.
+func RollingKeys() []string {
+	rollingMu.Lock()
+	defer rollingMu.Unlock()
+	keys := make([]string, 0, len(rolling))
+	for k := range rolling {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	return keys
+}
+
+// ── 2. parallel batch estimation ────────────────────────────────────────────
+
+// EstimateAll estimates many texts concurrently, reassembled in input
+// order. Pure function per element → embarrassingly parallel.
+func EstimateAll(texts []string, estimator func(string) int) []int {
+	if estimator == nil {
+		estimator = DefaultEstimator
+	}
+	out := make([]int, len(texts))
+	var wg sync.WaitGroup
+	for i, t := range texts {
+		wg.Add(1)
+		go func(i int, t string) {
+			defer wg.Done()
+			out[i] = estimator(t)
+		}(i, t)
+	}
+	wg.Wait()
+	return out
+}
+
+// ── 3. typed admission registry ─────────────────────────────────────────────
+
+// BlockSource — a named producer of contribution blocks (the typed shape of
+// a registered contributor).
+type BlockSource interface {
+	BlockName() string
+	Block(task, session map[string]interface{}) *ContributionBlock
+}
+
+// BlockSourceFunc — adapter turning a plain function into a BlockSource.
+type BlockSourceFunc struct {
+	name string
+	fn   ContributorFunc
+}
+
+// BlockName implements BlockSource.
+func (f BlockSourceFunc) BlockName() string { return f.name }
+
+// Block implements BlockSource.
+func (f BlockSourceFunc) Block(task, session map[string]interface{}) *ContributionBlock {
+	return f.fn(task, session)
+}
+
+// sourceRegistry — compile-time assertion: every registry entry satisfies
+// BlockSource (catches signature drift at build time).
+var _ = func() bool {
+	var registry = map[string]BlockSourceFunc{}
+	for _, v := range registry {
+		var _ BlockSource = v
+	}
+	return true
+}()

src/go/context-broker/phaseb_test.go

@@ -0,0 +1,110 @@
+package main
+
+// Phase B battery — the language-advantage wins exercised, including under
+// the race detector (go test -race).
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+)
+
+func TestEstimateAllMatchesSerial(t *testing.T) {
+	texts := make([]string, 0, 500)
+	for i := 0; i < 500; i++ {
+		texts = append(texts, fmt.Sprintf("token estimate batch %d %v", i, string(rune('a'+i%26))))
+	}
+	serial := make([]int, len(texts))
+	for i, s := range texts {
+		serial[i] = DefaultEstimator(s)
+	}
+	parallel := EstimateAll(texts, nil)
+	for i := range serial {
+		if serial[i] != parallel[i] {
+			t.Fatalf("EstimateAll[%d]: got %d want %d", i, parallel[i], serial[i])
+		}
+	}
+}
+
+func TestConcurrentWindowRegistryRaceFree(t *testing.T) {
+	ResetDiscoveredWindows()
+	models := make([]string, 0, 50)
+	for i := 0; i < 50; i++ {
+		models = append(models, fmt.Sprintf("model-%d", i))
+	}
+	got := DiscoverWindowsConcurrent(models, func(id string) int {
+		RegisterWindow(id, 32768)
+		return 32768
+	})
+	for i, w := range got {
+		if w != 32768 {
+			t.Fatalf("window[%d]=%d want 32768", i, w)
+		}
+	}
+	// concurrent readers while writers mutate — the race detector's job
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			if i%2 == 0 {
+				RegisterWindow(fmt.Sprintf("race-%d", i), 8192)
+			} else {
+				_ = DefaultWindow("model-0")
+			}
+		}(i)
+	}
+	wg.Wait()
+	ResetDiscoveredWindows()
+}
+
+func TestRollingStoreRaceFreeAndSnapshot(t *testing.T) {
+	for k := range rolling {
+		delete(rolling, k)
+	}
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			StoreRollingSummary("operator", fmt.Sprintf("s%d", i), "S", i, "tid")
+			_ = RollingSummaryState("operator", fmt.Sprintf("s%d", i))
+		}(i)
+	}
+	wg.Wait()
+	keys := RollingKeys()
+	if len(keys) != 20 {
+		t.Fatalf("RollingKeys: got %d want 20", len(keys))
+	}
+	want := append([]string(nil), keys...)
+	sort.Strings(want)
+	for i := range keys {
+		if keys[i] != want[i] {
+			t.Fatalf("RollingKeys not sorted: [%d]=%q want %q", i, keys[i], want[i])
+		}
+		if keys[i] != fmt.Sprintf("operator\x00s%d", i) && !strings.HasPrefix(keys[i], "operator\x00s") {
+			t.Fatalf("RollingKeys[%d]=%q unexpected key", i, keys[i])
+		}
+	}
+	for k := range rolling {
+		delete(rolling, k)
+	}
+}
+
+func TestBlockSourceRegistryCompileTime(t *testing.T) {
+	var src BlockSource = BlockSourceFunc{
+		name: "tools",
+		fn: func(map[string]interface{}, map[string]interface{}) *ContributionBlock {
+			return &ContributionBlock{Name: "tools", Priority: 9, Content: "x", Tokens: 1, Source: "t"}
+		},
+	}
+	blk := src.Block(nil, nil)
+	if blk == nil || blk.Name != "tools" {
+		t.Fatalf("BlockSource dispatch broken: %+v", blk)
+	}
+	if src.BlockName() != "tools" {
+		t.Fatalf("BlockName: got %q", src.BlockName())
+	}
+}