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

Language-advantage pass on the FULL Go port (truth battery stays green,
now replayed through the parallel path):
- build_auto_context fans the oracle's four independent IO calls (warm
  route, ledger read, memory recall, always digest) out to goroutines —
  wall time is the slowest call, not the sum of four subprocesses;
  results assembled in the oracle's exact field order so output is
  byte-identical to the serial Python semantics
- subprocess + clock seams promoted to interfaces (CommandRunner, Clock)
  with RunnerFunc/ClockFunc adapters and compile-time assertions
  (var _ CommandRunner = RunnerFunc(defaultRunScript), var _ Clock = ...);
  the replay seam swap now rides the same interfaces the embedded callers
  use
- save_context tmp file made per-writer (os.CreateTemp + atomic rename):
  identical serial semantics, but concurrent hooks no longer clobber the
  oracle's shared .tmp path (its fixed tmp races ENOENT under concurrency)
- Phase B battery (phaseb_test.go): parallel-build parity vs frozen truth,
  counting-decorator seam swap (3 subprocess runs per build), 8-goroutine
  concurrent builds + concurrent hooks — all race-clean under -race

src/go/auto_context/main.go

@@ -27,6 +27,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 	"unicode"
 	"unicode/utf8"
@@ -116,10 +117,53 @@ var (
 	PythonBin  = defaultPythonBin
 	ScriptsDir = defaultScriptsDir
 	MemoryDir  = defaultMemoryDir
-	ClockNow   = defaultClockNow
-	RunScript  = defaultRunScript
 )
 
+// CommandRunner — the subprocess seam as an interface (Phase B): the
+// runner is swappable in embedded callers and tests (fake runners,
+// counting/pooled runners) without touching port code; the default
+// implementation is compile-time asserted against the contract.
+type CommandRunner interface {
+	Run(exe string, args []string, timeout time.Duration) ExecResult
+}
+
+// RunnerFunc adapts a plain function to CommandRunner.
+type RunnerFunc func(exe string, args []string, timeout time.Duration) ExecResult
+
+// Run satisfies CommandRunner.
+func (f RunnerFunc) Run(exe string, args []string, timeout time.Duration) ExecResult {
+	return f(exe, args, timeout)
+}
+
+// compile-time: the default subprocess implementation satisfies the seam
+var _ CommandRunner = RunnerFunc(defaultRunScript)
+
+// Clock — the wall-clock seam as an interface (Phase B): build_auto_context
+// reads it exactly once per build, like the oracle's datetime.now call.
+type Clock interface {
+	Now() time.Time
+}
+
+// ClockFunc adapts a plain function to Clock.
+type ClockFunc func() time.Time
+
+// Now satisfies Clock.
+func (f ClockFunc) Now() time.Time { return f() }
+
+// compile-time: the default clock satisfies the seam
+var _ Clock = ClockFunc(defaultClockNow)
+
+// defaultRunner / defaultClock — the real implementations, kept as named
+// seams so per-row swaps in replay.go can restore them.
+var defaultRunner CommandRunner = RunnerFunc(defaultRunScript)
+var defaultClock Clock = ClockFunc(defaultClockNow)
+
+// Runner — the swapped-per-truth-row subprocess seam.
+var Runner CommandRunner = defaultRunner
+
+// ClockNow — the swapped-per-truth-row clock seam.
+var ClockNow Clock = defaultClock
+
 func warmRouterPath() string { return filepath.Join(ScriptsDir(), "warm_router.py") }
 func memoryStorePath() string {
 	return filepath.Join(ScriptsDir(), "memory_store.py")
@@ -317,7 +361,7 @@ func pythonReprString(s string) string {
 // router subprocess and return the fired-neurons payload; ANY failure
 // (nonzero exit, timeout, bad JSON, empty stdout) degrades to None.
 func runWarmRouter(message string) any {
-	res := RunScript(PythonBin(), []string{warmRouterPath(), "route", message, "--json"}, 10*time.Second)
+	res := Runner.Run(PythonBin(), []string{warmRouterPath(), "route", message, "--json"}, 10*time.Second)
 	if res.Err == nil && res.ReturnCode == 0 && strings.TrimSpace(res.Stdout) != "" {
 		if v, err := pyParse(res.Stdout); err == nil {
 			return v
@@ -329,7 +373,7 @@ func runWarmRouter(message string) any {
 // runMemoryRecall — run_memory_recall: recall relevant entries; a
 // non-list payload degrades to [], a list is capped at budget entries.
 func runMemoryRecall(query string, budget int) []any {
-	res := RunScript(PythonBin(), []string{memoryStorePath(), "recall", query, "--json"}, 15*time.Second)
+	res := Runner.Run(PythonBin(), []string{memoryStorePath(), "recall", query, "--json"}, 15*time.Second)
 	if res.Err == nil && res.ReturnCode == 0 && strings.TrimSpace(res.Stdout) != "" {
 		if v, err := pyParse(res.Stdout); err == nil {
 			if list, ok := v.([]any); ok {
@@ -346,7 +390,7 @@ func runMemoryRecall(query string, budget int) []any {
 // runAlwaysDigest — run_always_digest: the always-on digest payload or
 // None on any failure.
 func runAlwaysDigest() any {
-	res := RunScript(PythonBin(), []string{warmRouterPath(), "always", "--json"}, 10*time.Second)
+	res := Runner.Run(PythonBin(), []string{warmRouterPath(), "always", "--json"}, 10*time.Second)
 	if res.Err == nil && res.ReturnCode == 0 && strings.TrimSpace(res.Stdout) != "" {
 		if v, err := pyParse(res.Stdout); err == nil {
 			return v
@@ -410,9 +454,16 @@ func recallQueryWords(message string) []string {
 
 // buildAutoContext — build_auto_context: warm neurons, active tasks,
 // memory recall (substantive messages only), always-on digest.
+//
+// Phase B (language advantage): the oracle's four IO calls are mutually
+// independent (pure reads of external state), so they fan out to
+// goroutines — wall time is the slowest call, not the sum of four
+// subprocesses. Results are assembled in the oracle's exact field order,
+// so output is byte-identical to the serial Python semantics (the battery
+// replays the frozen truth against THIS parallel path).
 func buildAutoContext(message string) *OrderedMap {
 	context := NewOrderedMap(
-		KV{"timestamp", isoformat(ClockNow().UTC())},
+		KV{"timestamp", isoformat(ClockNow.Now().UTC())},
 		KV{"message_preview", runeSlice(message, 200)},
 		KV{"warm_neurons", nil},
 		KV{"memory_recall", nil},
@@ -420,16 +471,9 @@ func buildAutoContext(message string) *OrderedMap {
 		KV{"active_tasks", nil},
 	)
 
-	warm := runWarmRouter(message)
-	if pyTruthy(warm) {
-		context.Set("warm_neurons", warm)
-	}
-
-	tasks := loadActiveTasks()
-	if len(tasks) > 0 {
-		context.Set("active_tasks", tasks)
-	}
-
+	// cheap, CPU-only: key-term extraction (no IO) stays on the caller
+	// goroutine so the fan-out carries the exact query the oracle computes
+	query := ""
 	if message != "" && utf8.RuneCountInString(strings.TrimSpace(message)) > 10 {
 		words := recallQueryWords(message)
 		if len(words) > 0 {
@@ -437,14 +481,41 @@ func buildAutoContext(message string) *OrderedMap {
 			if len(words) < n {
 				n = len(words)
 			}
-			recall := runMemoryRecall(strings.Join(words[:n], " "), 5)
-			if len(recall) > 0 {
-				context.Set("memory_recall", recall)
-			}
+			query = strings.Join(words[:n], " ")
 		}
 	}
 
-	digest := runAlwaysDigest()
+	var (
+		wg     sync.WaitGroup
+		warm   any
+		digest any
+		tasks  []any
+		recall []any
+	)
+	launch := func(f func()) {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			f()
+		}()
+	}
+	launch(func() { warm = runWarmRouter(message) })
+	launch(func() { tasks = loadActiveTasks() })
+	if query != "" {
+		launch(func() { recall = runMemoryRecall(query, 5) })
+	}
+	launch(func() { digest = runAlwaysDigest() })
+	wg.Wait()
+
+	if pyTruthy(warm) {
+		context.Set("warm_neurons", warm)
+	}
+	if len(tasks) > 0 {
+		context.Set("active_tasks", tasks)
+	}
+	if len(recall) > 0 {
+		context.Set("memory_recall", recall)
+	}
 	if pyTruthy(digest) {
 		context.Set("always_digest", digest)
 	}
@@ -454,15 +525,33 @@ func buildAutoContext(message string) *OrderedMap {
 
 // saveContext — save_context: atomic write (tmp file + rename) of the
 // context JSON, indent=2, ensure_ascii.
+//
+// Phase B: the tmp file is per-writer (os.CreateTemp in the target dir),
+// not the oracle's fixed CONTEXT_FILE+".tmp" — the rename semantics are
+// identical, but concurrent hooks no longer clobber each other's tmp
+// (the oracle's shared-tmp path races ENOENT under concurrency).
 func saveContext(context *OrderedMap) error {
-	if err := os.MkdirAll(filepath.Dir(contextFilePath()), 0o777); err != nil {
+	dir := filepath.Dir(contextFilePath())
+	if err := os.MkdirAll(dir, 0o777); err != nil {
 		return err
 	}
-	tmp := contextFilePath() + ".tmp"
-	if err := os.WriteFile(tmp, []byte(encodePythonJSON(context, 0)), 0o644); err != nil {
+	tmp, err := os.CreateTemp(dir, ".auto_context-*.tmp")
+	if err != nil {
 		return err
 	}
-	return os.Rename(tmp, contextFilePath())
+	name := tmp.Name()
+	defer os.Remove(name) // no-op after a successful rename
+	if _, err := tmp.Write([]byte(encodePythonJSON(context, 0))); err != nil {
+		tmp.Close()
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	if err := os.Chmod(name, 0o644); err != nil {
+		return err
+	}
+	return os.Rename(name, contextFilePath())
 }
 
 // handleHook — handle_hook: the on_message plugin hook. The truthiness

src/go/auto_context/phaseb_test.go

@@ -0,0 +1,133 @@
+package main
+
+// Phase B battery — pins the language-advantage guarantees: the subprocess
+// and clock seams are interfaces (compile-time asserted in main.go),
+// build_auto_context fans the oracle's four independent IO calls out to
+// goroutines with byte-identical results, and concurrent builds are
+// race-clean (run under -race in CI).
+
+import (
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+)
+
+// countingRunner — a CommandRunner decorator that counts invocations: the
+// seam swaps in without touching port code.
+type countingRunner struct {
+	mu    sync.Mutex
+	inner CommandRunner
+	count int
+}
+
+func (c *countingRunner) Run(exe string, args []string, timeout time.Duration) ExecResult {
+	c.mu.Lock()
+	c.count++
+	c.mu.Unlock()
+	return c.inner.Run(exe, args, timeout)
+}
+
+// swapSeams — install the frozen seams for one test; restore on cleanup.
+func swapSeams(t *testing.T, routeMode, alwaysMode, recallMode string) string {
+	t.Helper()
+	home := t.TempDir()
+	memdir := filepath.Join(home, "memory")
+	if err := os.MkdirAll(memdir, 0o777); err != nil {
+		t.Fatal(err)
+	}
+	MemoryDir = func() string { return memdir }
+	ScriptsDir = func() string { return "/tmp/actx_stubs" }
+	PythonBin = func() string { return "python3" }
+	Runner = buildFakeRunner(routeMode, alwaysMode, recallMode)
+	base, _ := time.Parse(time.RFC3339, "2026-10-11T12:00:00.123456+00:00")
+	ClockNow = ClockFunc(func() time.Time { return base })
+	t.Cleanup(func() {
+		MemoryDir = defaultMemoryDir
+		ScriptsDir = defaultScriptsDir
+		PythonBin = defaultPythonBin
+		Runner = defaultRunner
+		ClockNow = defaultClock
+	})
+	return memdir
+}
+
+// TestPhaseBParallelBuildParity — the frozen build truth replays green
+// against the GOROUTINE fan-out path (results assembled in oracle order
+// are byte-identical to the serial Python semantics).
+func TestPhaseBParallelBuildParity(t *testing.T) {
+	swapSeams(t, "ok", "ok", "ok")
+	doc := loadTruth(t)
+	for _, name := range []string{"build:full", "build:unicode-preview", "build:stopwords", "build:short-message"} {
+		row := findRow(t, name)
+		if msg := compareOne(doc, &row); msg != "" {
+			t.Fatalf("parallel build diverged on %s: %s", name, msg)
+		}
+	}
+}
+
+// TestPhaseBRunnerSeamIsAnInterface — a counting decorator rides the
+// CommandRunner seam; a full build issues exactly 3 subprocess runs (route,
+// recall, always — the ledger read is file IO, not a subprocess).
+func TestPhaseBRunnerSeamIsAnInterface(t *testing.T) {
+	swapSeams(t, "ok", "ok", "ok")
+	counter := &countingRunner{inner: Runner}
+	Runner = counter
+	ctx := buildAutoContext("remember the config parser override discussion tests")
+	if counter.count != 3 {
+		t.Fatalf("expected 3 subprocess runs per build, got %d", counter.count)
+	}
+	if warm := ctx.Get2("warm_neurons"); !pyTruthy(warm) {
+		t.Fatal("warm payload lost through the decorated seam")
+	}
+}
+
+// TestPhaseBConcurrentBuilds — 8 goroutines building contexts through the
+// shared fake runner must all agree (and stay race-clean).
+func TestPhaseBConcurrentBuilds(t *testing.T) {
+	swapSeams(t, "ok", "ok", "ok")
+	const n = 8
+	results := make([]string, n)
+	var wg sync.WaitGroup
+	for i := 0; i < n; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			results[i] = mustJSON(buildAutoContext("remember the config parser override discussion tests"))
+		}(i)
+	}
+	wg.Wait()
+	for i := 1; i < n; i++ {
+		if results[i] != results[0] {
+			t.Fatalf("concurrent build %d diverged:\n  first: %s\n  got:   %s", i, results[0], results[i])
+		}
+	}
+}
+
+// TestPhaseBConcurrentHooks — concurrent handle_hook calls write the
+// context file through per-writer tmp + atomic rename: every rename
+// succeeds (no shared-tmp ENOENT race, which the oracle's fixed
+// CONTEXT_FILE+".tmp" would hit) and every reader sees complete JSON.
+func TestPhaseBConcurrentHooks(t *testing.T) {
+	memdir := swapSeams(t, "ok", "ok", "ok")
+	data := NewOrderedMap(KV{"message", "remember the config parser override discussion tests"})
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if r := handleHook("on_message", data); r.Get2("status") != "ok" {
+				t.Errorf("hook failed: %v", mustJSON(r))
+			}
+		}()
+	}
+	wg.Wait()
+	raw, err := os.ReadFile(filepath.Join(memdir, "index", "auto_context.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pyParse(string(raw)); err != nil {
+		t.Fatalf("context file torn or invalid: %v", err)
+	}
+}

src/go/auto_context/replay.go

@@ -113,8 +113,8 @@ func execRow(doc *TruthDoc, row *TruthRow) (got any, exact bool, err error) {
 
 	defer func() {
 		MemoryDir = defaultMemoryDir
-		ClockNow = defaultClockNow
-		RunScript = defaultRunScript
+		Runner = defaultRunner
+		ClockNow = defaultClock
 		PythonBin = defaultPythonBin
 		ScriptsDir = defaultScriptsDir
 	}()
@@ -129,7 +129,7 @@ func execRow(doc *TruthDoc, row *TruthRow) (got any, exact bool, err error) {
 	if err != nil {
 		return nil, false, err
 	}
-	ClockNow = func() time.Time { return base }
+	ClockNow = ClockFunc(func() time.Time { return base })
 
 	// file seeds from setup
 	if f, ok := setup.Get("tasks_file"); ok {
@@ -160,77 +160,7 @@ func execRow(doc *TruthDoc, row *TruthRow) (got any, exact bool, err error) {
 		}
 		return "ok"
 	}
-	routeMode, alwaysMode, recallMode := mode("stub_route"), mode("stub_always"), mode("stub_recall")
-
-	RunScript = func(exe string, args []string, timeout time.Duration) ExecResult {
-		var out string
-		code := 0
-		base := filepath.Base(args[0])
-		if base == "warm_router.py" {
-			if len(args) > 1 && args[1] == "always" {
-				switch alwaysMode {
-				case "ok":
-					out = encodePythonJSONCompact(NewOrderedMap(
-						KV{"digest", "daily-summary"}, KV{"items", 7},
-						KV{"topics", []any{"memory", "tasks", "recall"}}))
-				case "fail":
-					code = 1
-				case "badjson":
-					out = "not-json{{{"
-				case "empty":
-					// print nothing
-				}
-			} else {
-				msg := ""
-				if len(args) > 2 {
-					msg = args[2]
-				}
-				switch routeMode {
-				case "ok":
-					out = encodePythonJSONCompact(NewOrderedMap(
-						KV{"echo_message", msg},
-						KV{"fired", []any{"warm-alpha", "warm-beta", "warm-gamma"}},
-						KV{"matched", 3}, KV{"route", "focus"}))
-				case "list":
-					out = "[1, 2]"
-				case "emptylist":
-					out = "[]"
-				case "firedint":
-					out = encodePythonJSONCompact(NewOrderedMap(
-						KV{"fired", 3}, KV{"echo_message", msg}))
-				case "fail":
-					code = 1
-				case "badjson":
-					out = "not-json{{{"
-				case "empty":
-					// print nothing
-				}
-			}
-		} else {
-			query := ""
-			if len(args) > 2 {
-				query = args[2]
-			}
-			switch recallMode {
-			case "ok":
-				items := []any{NewOrderedMap(KV{"echo_query", query})}
-				for _, e := range recallEntries {
-					items = append(items, e)
-				}
-				out = encodePythonJSONCompact(items)
-			case "nonlist":
-				out = encodePythonJSONCompact(NewOrderedMap(
-					KV{"query", query}, KV{"note", "not a list"}))
-			case "fail":
-				code = 1
-			case "badjson":
-				out = "<xml?>"
-			case "emptyout":
-				// print nothing
-			}
-		}
-		return ExecResult{Stdout: out, ReturnCode: code}
-	}
+	Runner = buildFakeRunner(mode("stub_route"), mode("stub_always"), mode("stub_recall"))
 
 	scrub := func(v any) any {
 		return scrubPaths(v, memdir)
@@ -327,6 +257,82 @@ func execRow(doc *TruthDoc, row *TruthRow) (got any, exact bool, err error) {
 	return nil, false, fmt.Errorf("unknown row op %s (%s)", row.Op, row.Name)
 }
 
+// buildFakeRunner — the env-driven stub scripts' behavior as a
+// CommandRunner (mirrors /tmp/actx_stubs byte-for-byte): payloads ECHO the
+// caller's message/query so the frozen truth pins the exact strings the
+// port hands to the subprocess seam. Reused by the Phase B battery.
+func buildFakeRunner(routeMode, alwaysMode, recallMode string) RunnerFunc {
+	return func(exe string, args []string, timeout time.Duration) ExecResult {
+		var out string
+		code := 0
+		base := filepath.Base(args[0])
+		if base == "warm_router.py" {
+			if len(args) > 1 && args[1] == "always" {
+				switch alwaysMode {
+				case "ok":
+					out = encodePythonJSONCompact(NewOrderedMap(
+						KV{"digest", "daily-summary"}, KV{"items", 7},
+						KV{"topics", []any{"memory", "tasks", "recall"}}))
+				case "fail":
+					code = 1
+				case "badjson":
+					out = "not-json{{{"
+				case "empty":
+					// print nothing
+				}
+			} else {
+				msg := ""
+				if len(args) > 2 {
+					msg = args[2]
+				}
+				switch routeMode {
+				case "ok":
+					out = encodePythonJSONCompact(NewOrderedMap(
+						KV{"echo_message", msg},
+						KV{"fired", []any{"warm-alpha", "warm-beta", "warm-gamma"}},
+						KV{"matched", 3}, KV{"route", "focus"}))
+				case "list":
+					out = "[1, 2]"
+				case "emptylist":
+					out = "[]"
+				case "firedint":
+					out = encodePythonJSONCompact(NewOrderedMap(
+						KV{"fired", 3}, KV{"echo_message", msg}))
+				case "fail":
+					code = 1
+				case "badjson":
+					out = "not-json{{{"
+				case "empty":
+					// print nothing
+				}
+			}
+		} else {
+			query := ""
+			if len(args) > 2 {
+				query = args[2]
+			}
+			switch recallMode {
+			case "ok":
+				items := []any{NewOrderedMap(KV{"echo_query", query})}
+				for _, e := range recallEntries {
+					items = append(items, e)
+				}
+				out = encodePythonJSONCompact(items)
+			case "nonlist":
+				out = encodePythonJSONCompact(NewOrderedMap(
+					KV{"query", query}, KV{"note", "not a list"}))
+			case "fail":
+				code = 1
+			case "badjson":
+				out = "<xml?>"
+			case "emptyout":
+				// print nothing
+			}
+		}
+		return ExecResult{Stdout: out, ReturnCode: code}
+	}
+}
+
 // scrubPaths — normalize the row's memory-dir prefix to @MEMDIR@ so the
 // frozen want (recorded with the generator's temp dirs) compares equal.
 func scrubPaths(v any, memdir string) any {