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

typed CorruptEventStoreError (errors.As + offset>=2 message fix), journalMu
serialized appends, atomic eeUUIDNext mint; phase_b race battery 8x4 fan-out
with id-uniqueness law; re-verified freeze 61/61, parity 73/73, build/vet/
gofmt/race, pytest both pins (eb2-*.exit evidence)

docs/dichotomy-rebase-honest-audit.md

@@ -2,16 +2,16 @@
 
 _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL only when 80%+ of its Python functions have name-preserving counterparts in the Go/Rust implementation — directories without the functions are facades and count as NOT TOUCHED._
 
-**FULL completion: 55%** (105/192 runtime modules at FULL)
+**FULL completion: 59%** (114/192 runtime modules at FULL)
 
-Function-level detail: 70% (1608/2282 runtime functions matched) — secondary; includes in-flight work
+Function-level detail: 72% (1658/2287 runtime functions matched) — secondary; includes in-flight work
 
 | Status | Modules | Lines | Meaning |
 |---|---|---|---|
-| FULL | 105 | 44333 | 80%+ of functions ported |
-| PARTIAL | 48 | 18960 | 20–80% — real logic exists, gaps named below |
+| FULL | 114 | 47584 | 80%+ of functions ported |
+| PARTIAL | 39 | 16224 | 20–80% — real logic exists, gaps named below |
 | NOT TOUCHED | 39 | 11471 | no name-preserving implementation |
-| Total runtime | 192 | 74764 | |
+| Total runtime | 192 | 75279 | |
 
 ## NOT TOUCHED — the port queue, in priority order (largest first)
 
@@ -65,30 +65,22 @@ Function-level detail: 70% (1608/2282 runtime functions matched) — secondary;
 | memory_store | 2333 | 30/68 | _ctx_budget, now_iso, ensure_memory_type_column, _embed, _temp_json, _entities_to_json |
 | warm_router | 1518 | 17/33 | _ensure_use_history_table, record_outcome, record_neuron_fire, base_level_activation, _db, _aaak |
 | tool_executor | 1836 | 33/47 | _execute_tool_now, serialize_prepared_tool, deserialize_prepared_tool, confirm_write, _resolve_execution_cwd, _run_command |
-| token_optimizer | 514 | 16/25 | get_optimizer, _tc_core, _to_core, to_dict, _to_state, _apply_state |
 | time_awareness | 402 | 12/20 | _clock_args, _ta_core, get_time_awareness, to_dict, _to_state, _apply_state |
 | intent_continuity | 284 | 6/13 | _now, _raise_core_err, to_dict, from_dict, save, _ledger_state |
-| voice_sync | 299 | 3/10 | _now, get_delivery_register_from_db, get_delivery_register, update_rune_md, check_sync, handle_hook |
 | audit_ledger | 212 | 2/8 | _compute_hash, get_entries, log_promotion, log_gate_result, log_drift_alert, main |
-| context_pressure | 262 | 3/9 | log_measurement, read_log, log_stats, _now, handle_hook, main |
 | core_server | 234 | 6/12 | make_server, log_message, _send_json, _send_ndjson, do_GET, do_POST |
-| model_registry | 253 | 2/8 | test_registry, to_dict, get_profile, get_enabled_profiles, get_profiles_by_tier, list_profiles |
 | progressive_escalation | 279 | 2/8 | execute_with_escalation, test_escalation, handle_hook, to_dict, execute, verify |
 | voicebox | 164 | 5/11 | _run, start_recording, stop_recording, _transcribe_and_type, on_press, on_release |
 | persona_definition | 210 | 2/7 | _by_topic, compile_full, emit_constitution, emit_identity, emit_persona |
 | task_persist | 269 | 5/10 | _save, export, restore, main, handle_hook |
-| tiered_executor | 291 | 3/8 | _ctx_budget, test_executor, to_dict, to_dict, _execute_model |
 | construction_commands | 268 | 1/5 | _refuse, _validate, _accept, dispatch |
 | drift_observability | 399 | 5/9 | _do_core, main, _load_alerts, _save_alerts |
-| layout_contract | 336 | 2/6 | define_region, bind_component, amend_layout_contract, record_layout_observation |
 | pins | 282 | 7/11 | _store_path, _detail_blob, unpin, list_pins |
 | prompt_sync | 487 | 11/15 | _now, _rust_available, get_store, main |
-| self_improve | 246 | 1/5 | _now, suggest_improvements, handle_hook, main |
 | task_gate | 301 | 5/9 | _tg_core, _now, list_gates, main |
 | tier_manager | 262 | 6/10 | demotion_sweep, get_tier_config, should_search, should_compress |
 | connection_pool | 226 | 7/10 | get_pool, _kernel_state, _get_connection_core |
 | construction_activation | 237 | 1/4 | _scaffold_for, _auto_structural_awareness, dispatch_from_decision |
-| construction_projection | 254 | 1/4 | _contract_entity, validate, write_mini_doc |
 | dependency_tracker | 202 | 2/5 | check_propagation, get_full_map, main |
 | graph_projection | 134 | 3/6 | _model_to_dicts, write_all, _now |
 | state_checkpoint | 182 | 6/9 | _sck_core, _load_ledger, _save_ledger |
@@ -96,7 +88,6 @@ Function-level detail: 70% (1608/2282 runtime functions matched) — secondary;
 | cognitive_budgeter | 221 | 1/3 | test_budgeter, to_dict |
 | inference_trace | 204 | 4/6 | record_trace, read_traces |
 | response_adapter | 222 | 5/7 | test_adapter, to_dict |
-| standardized_providers | 281 | 3/5 | list_providers, connect_all_local |
 | structural_migration | 77 | 2/4 | apply_migration, consent_brief |
 | subconscious_pass | 236 | 3/5 | main, get_recent_findings |
 | subsystem_health | 213 | 4/6 | set_auto_filer, configure_logging |
@@ -150,6 +141,7 @@ Function-level detail: 70% (1608/2282 runtime functions matched) — secondary;
 | build_mode | 521 | 33/33 | GO |
 | growth_delta | 520 | 17/18 | GO |
 | event_wait | 518 | 14/14 | GO |
+| token_optimizer | 514 | 25/25 | RUST |
 | cache | 492 | 17/17 | GO |
 | claim_audit | 484 | 14/14 | RUST |
 | model_connect | 471 | 11/11 | GO |
@@ -162,31 +154,39 @@ Function-level detail: 70% (1608/2282 runtime functions matched) — secondary;
 | witness | 446 | 15/15 | GO |
 | cron_engine | 440 | 27/27 | GO |
 | model_profiles | 438 | 15/15 | GO |
+| layout_contract | 434 | 6/6 | RUST |
 | microkernel | 434 | 12/12 | GO |
 | model_routing | 432 | 13/13 | GO |
+| voice_sync | 426 | 12/12 | RUST |
 | night_mode | 421 | 19/19 | GO |
 | construction_mode | 412 | 5/5 | RUST |
+| self_improve | 384 | 7/8 | RUST |
 | build_basis | 375 | 11/13 | RUST |
 | attention_capsule | 365 | 6/7 | RUST |
 | testing_mode | 362 | 11/13 | RUST |
 | drift_cache | 357 | 12/14 | RUST |
 | provider_budget | 357 | 20/20 | GO |
 | drift-ledger | 355 | 14/14 | unclassified |
+| context_pressure | 351 | 9/9 | RUST |
 | builder_bridge | 326 | 11/11 | GO |
 | drift_compression | 323 | 7/7 | RUST |
 | outcome_ledger | 318 | 7/8 | RUST |
+| construction_projection | 317 | 4/4 | RUST |
 | message_queue | 316 | 13/13 | GO |
 | lexicon_evolution | 304 | 7/8 | unclassified |
 | event_envelope | 303 | 8/8 | GO |
 | stage_guards | 303 | 9/9 | GO |
 | message_graph | 295 | 12/13 | unclassified |
 | toolchain | 295 | 5/6 | RUST |
+| tiered_executor | 291 | 8/8 | GO |
 | salience_gate | 283 | 6/6 | RUST |
+| standardized_providers | 281 | 5/5 | GO |
 | shard_manager | 276 | 9/9 | RUST |
 | auto_context | 273 | 8/8 | GO |
 | spine_language | 269 | 5/5 | RUST |
 | construction_patterns_builtin | 266 | 1/1 | RUST |
 | iteration | 261 | 7/8 | RUST |
+| model_registry | 253 | 8/8 | GO |
 | event_bus | 244 | 7/7 | GO |
 | openai_compat_transport | 227 | 9/9 | GO |
 | evidence_probe | 213 | 9/9 | GO |

src/go/event-bus/envelope.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"time"
 	"unicode"
 )
@@ -162,22 +163,24 @@ func resetEEUUID() {
 	if v := os.Getenv("EE_UUID_BASE"); v != "" {
 		fmt.Sscanf(v, "%d", &base)
 	}
-	eeUUIDNext = base - 1 // eeDetHex increments BEFORE use (count(1) yields 1 first)
+	atomic.StoreInt64(&eeUUIDNext, base-1) // eeDetHex increments BEFORE use (count(1) yields 1 first)
 }
 
 // eeDetHex — the uuid4() stand-in. Under the determinism pins (battery +
 // parity set EE_FROZEN_NOW / EE_UUID_BASE) it is the rig's counter;
 // production paths mint a random 32-hex uuid like the oracle's uuid4().
+// The counter is atomic: concurrent Emit callers mint distinct ids
+// (Phase B concurrent-journal law).
 func eeDetHex() string {
-	eeUUIDNext++
+	n := atomic.AddInt64(&eeUUIDNext, 1)
 	if os.Getenv("EE_FROZEN_NOW") == "" && os.Getenv("EE_UUID_BASE") == "" {
 		b := make([]byte, 16)
 		if _, err := cryptorand.Read(b); err != nil {
-			return fmt.Sprintf("%032x", eeUUIDNext) // unreachable in practice
+			return fmt.Sprintf("%032x", n) // unreachable in practice
 		}
 		return hex.EncodeToString(b)
 	}
-	return fmt.Sprintf("%032x", eeUUIDNext)
+	return fmt.Sprintf("%032x", n)
 }
 
 // eeMintEventID — records.new_id: "evt_" + uuid4().hex[:12].

src/go/event-bus/phase_b_bus_test.go

@@ -0,0 +1,94 @@
+package main
+
+// phase_b_bus_test.go — the Phase B language-advantage pass (dichotomy
+// wave): concurrent Emit fan-out through the serialized journal writer
+// (-race clean; the Python oracle creates a NEW DurableEventStore per
+// emit — its per-instance RLock cannot coordinate cross-instance appends,
+// the Go writer can), and the typed CorruptEventStoreError
+// (errors.As discrimination, not string matching).
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+)
+
+// TestConcurrentEmitRace — 8 goroutines × 4 emits through Emit: all 32
+// appends survive as well-formed sorted-key JSONL rows. Run under -race.
+// No EE_FROZEN_NOW pin: the counter's %032x prefix collides at [:12] for
+// small n, so id uniqueness is only meaningful on the real uuid4 path.
+func TestConcurrentEmitRace(t *testing.T) {
+	sandbox := t.TempDir()
+	t.Setenv("RUNE_MEMORY_DIR", sandbox)
+	t.Setenv("EB_SANDBOX_ROOT", sandbox)
+
+	const workers, perWorker = 8, 4
+	resetEEUUID()
+	var wg sync.WaitGroup
+	errs := make(chan error, workers*perWorker)
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func(w int) {
+			defer wg.Done()
+			for i := 0; i < perWorker; i++ {
+				if _, err := Emit("user.message", "user",
+					ordKV("w", w, "i", i)); err != nil {
+					errs <- err
+				}
+			}
+		}(w)
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		t.Fatalf("concurrent emit: %v", err)
+	}
+
+	rows, err := ReplayOrdered(StorePath())
+	if err != nil {
+		t.Fatalf("replay: %v", err)
+	}
+	if len(rows) != workers*perWorker {
+		t.Fatalf("journal rows = %d, want %d", len(rows), workers*perWorker)
+	}
+	seen := make(map[string]bool, len(rows))
+	for _, r := range rows {
+		if et, _ := r.vals["event_type"].(string); et != "user.message" {
+			t.Fatalf("corrupt row rendered: %v", r.vals["event_type"])
+		}
+		id, _ := r.vals["event_id"].(string)
+		if seen[id] {
+			t.Fatalf("duplicate event_id under concurrency: %s", id)
+		}
+		seen[id] = true
+		if _, ok := r.vals["payload_hash"]; !ok {
+			t.Fatalf("row missing payload_hash: %v", r.vals)
+		}
+	}
+}
+
+// TestCorruptEventTyped — the store failure is a TYPE, not a string.
+func TestCorruptEventTyped(t *testing.T) {
+	sandbox := t.TempDir()
+	t.Setenv("RUNE_MEMORY_DIR", sandbox)
+	if err := os.MkdirAll(filepath.Dir(StorePath()), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(StorePath(), []byte("garbage line\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	_, err := ReplayOrdered(StorePath())
+	if err == nil {
+		t.Fatal("corrupt store must raise")
+	}
+	var typed *CorruptEventStoreError
+	if !errors.As(err, &typed) {
+		t.Fatalf("not typed CorruptEventStoreError: %v (%T)", err, err)
+	}
+	if typed.Offset != 1 || !strings.Contains(typed.Error(), "corrupt event at offset 1: Expecting value: line 1 column 1 (char 0)") {
+		t.Fatalf("message contract broken: %q", typed.Error())
+	}
+}

src/go/event-bus/store.go

@@ -28,8 +28,27 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 )
 
+// CorruptEventStoreError — the TypedError form of event_store.py's
+// DurableEventStore.CorruptEventStoreError (Phase B: callers discriminate
+// via errors.As at compile time instead of string matching).
+type CorruptEventStoreError struct {
+	Offset int
+	Cause  string
+}
+
+func (e *CorruptEventStoreError) Error() string {
+	return fmt.Sprintf("corrupt event at offset %d: %s", e.Offset, e.Cause)
+}
+
+// journalMu — Phase B concurrent-emit coordination: in-process appends to
+// the canonical journal are fully serialized (the Python oracle's
+// per-instance RLock cannot coordinate separate DurableEventStore
+// instances; the Go port can).
+var journalMu sync.Mutex
+
 // ── _store_path — the canonical journal path ──────────────────────────────
 
 // pyExpand — os.path.expandvars then expanduser (memory_env.expand).
@@ -418,8 +437,11 @@ func Emit(eventType string, source interface{}, payload interface{}) (*ordObj, e
 }
 
 // appendStoreLine — DurableEventStore.append+flush: mkdir parents, one
-// append line + "\n", fsync (durable right after emit).
+// append line + "\n", fsync (durable right after emit); the mutex
+// serializes concurrent in-process emitters end to end.
 func appendStoreLine(path, line string) error {
+	journalMu.Lock()
+	defer journalMu.Unlock()
 	if dir := filepath.Dir(path); dir != "" {
 		if err := os.MkdirAll(dir, 0o755); err != nil {
 			return err
@@ -478,15 +500,11 @@ func ReplayOrdered(path string) ([]*ordObj, error) {
 		}
 		v, lerr := pyLoads(line)
 		if lerr != nil {
-			return nil, fmt.Errorf(
-				"corrupt event at offset %d: Expecting value: line %d column 1 (char 0)",
-				offset, offset)
+			return nil, &CorruptEventStoreError{Offset: offset, Cause: "Expecting value: line 1 column 1 (char 0)"}
 		}
 		o, ok := v.(*ordObj)
 		if !ok {
-			return nil, fmt.Errorf(
-				"corrupt event at offset %d: Expecting value: line %d column 1 (char 0)",
-				offset, offset)
+			return nil, &CorruptEventStoreError{Offset: offset, Cause: "Expecting value: line 1 column 1 (char 0)"}
 		}
 		rows = append(rows, o)
 	}

verification/eb-summary.md

@@ -3,6 +3,26 @@
 Worktree: /mnt/workdrive/rune-wave-dd (detached, locked) · Oracle: src/rune/event_bus.py (244L, 7 fns)
 Before: 5/7 PARTIAL (missing _store_path, handle_hook) → After: 7/7 FULL (GO side).
 
+## Phase B re-verification (post language-advantage pass, evidence eb2-*)
+
+Code deltas: typed `CorruptEventStoreError` (errors.As, message contract
+"corrupt event at offset N: Expecting value: line 1 column 1 (char 0)" —
+fixes the latent offset≥2 message bug where the line number mirrored the
+offset), `journalMu` serializing in-process appends end to end, atomic
+`eeUUIDNext` mint counter (concurrent Emit mints distinct ids), and
+`phase_b_bus_test.go` (8×4 concurrent-emit fan-out -race clean with
+event_id uniqueness assertion + typed-error discrimination test).
+
+| Gate | Evidence (exit) | Result |
+|---|---|---|
+| freeze --verify | verification/eb2-freeze-verify.log | OK: 61 cases byte-identical (exit 0) |
+| live parity | verification/eb2-parity.log | parity: 73/73 match, negative control detected (exit 0) |
+| go build ./... | verification/eb2-gobuild.log | OK (exit 0) |
+| go vet / gofmt -l | verification/eb2-govet.log / eb2-gofmt.log | clean (exit 0 / empty) |
+| battery + race | verification/eb2-gotest.log | ok rune/event-bus 1.060s (exit 0) |
+| pytest live / RUNE_EVENT_BUS_CORE=python | verification/eb2-pytest-live.log / eb2-pytest-python-pin.log | 4 passed both pins (exit 0) |
+| audit regen | verification/eb2-audit.log | event_bus 7/7 GO · FULL 55% (105/192) (exit 0) |
+
 ## Evidence index (cross-checked by Read before attestation)
 
 | Gate | Command | Evidence (exit) |

verification/eb2-audit.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-freeze-verify.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-gobuild.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-gofmt.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-gotest.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-govet.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-parity.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-pytest-live.exit

@@ -0,0 +1 @@
+exit=0

verification/eb2-pytest-python-pin.exit

@@ -0,0 +1 @@
+exit=0