feat(event_bus): close PARTIAL gaps — dichotomy wave
This commit is contained in:
Nick committed
1 parent
dba124bfce
commit
a8d7867143
15 files changed
+3792
-423
No files matched your search
@@ -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: 58%** (112/192 runtime modules at FULL)
|
||||
**FULL completion: 55%** (105/192 runtime modules at FULL)
|
||||
|
||||
Function-level detail: 72% (1653/2287 runtime functions matched) — secondary; includes in-flight work
|
||||
Function-level detail: 70% (1608/2282 runtime functions matched) — secondary; includes in-flight work
|
||||
|
||||
| Status | Modules | Lines | Meaning |
|
||||
|---|---|---|---|
|
||||
| FULL | 112 | 47023 | 80%+ of functions ported |
|
||||
| PARTIAL | 41 | 16722 | 20–80% — real logic exists, gaps named below |
|
||||
| FULL | 105 | 44333 | 80%+ of functions ported |
|
||||
| PARTIAL | 48 | 18960 | 20–80% — real logic exists, gaps named below |
|
||||
| NOT TOUCHED | 39 | 11471 | no name-preserving implementation |
|
||||
| Total runtime | 192 | 75216 | |
|
||||
| Total runtime | 192 | 74764 | |
|
||||
|
||||
## NOT TOUCHED — the port queue, in priority order (largest first)
|
||||
|
||||
@@ -65,18 +65,25 @@ Function-level detail: 72% (1653/2287 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 |
|
||||
@@ -87,9 +94,9 @@ Function-level detail: 72% (1653/2287 runtime functions matched) — secondary;
|
||||
| state_checkpoint | 182 | 6/9 | _sck_core, _load_ledger, _save_ledger |
|
||||
| warm_neuron_store | 122 | 3/6 | _db, neuron_slugs, handle_hook |
|
||||
| cognitive_budgeter | 221 | 1/3 | test_budgeter, to_dict |
|
||||
| event_bus | 244 | 5/7 | _store_path, handle_hook |
|
||||
| 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 |
|
||||
@@ -143,7 +150,6 @@ Function-level detail: 72% (1653/2287 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 |
|
||||
@@ -156,20 +162,16 @@ Function-level detail: 72% (1653/2287 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 |
|
||||
@@ -179,15 +181,13 @@ Function-level detail: 72% (1653/2287 runtime functions matched) — secondary;
|
||||
| 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 |
|
||||
| project_registry | 205 | 10/12 | RUST |
|
||||
|
||||
@@ -0,0 +1,286 @@
|
||||
#!/usr/bin/env python3
|
||||
"""check_event_bus_parity.py — live oracle-vs-Go parity sweep for the
|
||||
event_bus gap closure (#2479, contract factory-gap-eb).
|
||||
|
||||
The frozen envelope (truth_bus.json, 61 cases) pins the surface byte-
|
||||
identically. THIS script drives the LIVE Python oracle (under the
|
||||
scripts/eb_freeze_seams rig: frozen clock + deterministic uuid counter +
|
||||
hermetic journal sandbox) and the Go binary (`op` mode, EE_FROZEN_NOW +
|
||||
per-op reset + the same sandbox normalization envs) across the same
|
||||
branches plus live extras, comparing the PYTHON-json byte form of every
|
||||
result:
|
||||
|
||||
* store_path — the env-derivation matrix via SUBPROCESSES (sandbox/,
|
||||
empty/, RUNE_HOME/, ~/ expansion, $HOME expansion, unset default),
|
||||
normalized to EB_SANDBOX / EB_HOME / REPO_ROOT placeholders.
|
||||
* event_types — the vocabulary table in definition order.
|
||||
* content_hash — any-shape payloads (sort_keys, ensure_ascii).
|
||||
* emit/emit_line — envelope build + durable append + journal LINE bytes,
|
||||
incl. the loud ValueError branches (dict/int/NoneType/float/bool/list).
|
||||
* log — filter + [-limit:] truncation matrix (0 → ALL, -N → front drop),
|
||||
the typed corrupt-file raise, and the cross-writer probe (oracle
|
||||
writes, Go replays the SAME file via the "replay" op).
|
||||
* stats — (total, by_type) first-occurrence order.
|
||||
* handle_hook — prefix, payload shaping (raw[:200], str(v)[:200] with
|
||||
repr forms + truncation, falsy data), journal after each hook.
|
||||
* main — captured CLI stdout + exit (emit/log/stats/hook, --json modes).
|
||||
|
||||
Usage: PYTHONPATH=scripts/eb_freeze_seams:src python3 scripts/check_event_bus_parity.py
|
||||
[--bin src/go/event-bus/event-bus]
|
||||
Exit 0 on 100% parity (incl. the negative control detecting perturbation),
|
||||
1 with the first divergences, 2 if the negative control fails to detect.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO / "src"))
|
||||
|
||||
FROZEN_NOW = "2026-10-14T11:22:33.123456+00:00"
|
||||
|
||||
import rune.event_bus as eb # noqa: E402 (rig active via PYTHONPATH)
|
||||
|
||||
SANDBOX = eb.EB_SANDBOX # the rig's hermetic journal dir (shared both sides)
|
||||
|
||||
|
||||
def run_go(binpath, env, req):
|
||||
p = subprocess.run([str(binpath), "op"], input=json.dumps(req),
|
||||
capture_output=True, text=True, env=env)
|
||||
if p.returncode != 0:
|
||||
raise RuntimeError(f"go op {req.get('op')} exited {p.returncode}: "
|
||||
f"{p.stdout} {p.stderr}")
|
||||
return json.loads(p.stdout)["json"]
|
||||
|
||||
|
||||
def go_env(extra=None):
|
||||
env = dict(os.environ)
|
||||
env["EE_FROZEN_NOW"] = FROZEN_NOW
|
||||
env["RUNE_MEMORY_DIR"] = SANDBOX
|
||||
env["EB_SANDBOX_ROOT"] = SANDBOX
|
||||
env["EB_REPO_ROOT"] = str(REPO)
|
||||
if extra:
|
||||
env.update(extra)
|
||||
return env
|
||||
|
||||
|
||||
def normalize(s, subs):
|
||||
for value, placeholder in subs:
|
||||
s = s.replace(value, placeholder)
|
||||
return s.replace("/mnt/workdrive/rune", "REPO_ROOT")
|
||||
|
||||
|
||||
def oracle_subproc(env, code):
|
||||
p = subprocess.run([sys.executable, "-c", code], capture_output=True,
|
||||
text=True, env=env, cwd=str(REPO))
|
||||
return p
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--bin", default=str(REPO / "src/go/event-bus/event-bus"))
|
||||
args = ap.parse_args()
|
||||
binpath = Path(args.bin)
|
||||
if not binpath.exists():
|
||||
print(f"Go binary missing: {binpath} (go build -o event-bus . in src/go/event-bus)")
|
||||
return 1
|
||||
|
||||
fails = total = 0
|
||||
|
||||
def check(name, oracle_json, go_json):
|
||||
nonlocal fails, total
|
||||
total += 1
|
||||
if oracle_json != go_json:
|
||||
fails += 1
|
||||
print(f"DIVERGE {name}:\n oracle: {oracle_json[:300]}\n"
|
||||
f" go: {go_json[:300]}")
|
||||
|
||||
# ── store_path — the env-derivation matrix (subprocess pairs) ─────
|
||||
# The ORACLE side runs WITHOUT the rig (PYTHONPATH=src only): these
|
||||
# rows are pure path derivation (no clock/uuid mints). The GO side
|
||||
# normalizes through the EB_* placeholder envs.
|
||||
home1 = Path(tempfile.mkdtemp(prefix="eb-parity-home-"))
|
||||
home2 = Path(tempfile.mkdtemp(prefix="eb-parity-home2-"))
|
||||
variants = [
|
||||
("default", {"RUNE_MEMORY_DIR": None, "RUNE_HOME": None},
|
||||
{"/mnt/workdrive/rune": "REPO_ROOT"}, {}),
|
||||
("sb", {"RUNE_MEMORY_DIR": SANDBOX}, {(SANDBOX): "EB_SANDBOX"},
|
||||
{"EB_SANDBOX_ROOT": SANDBOX}),
|
||||
("empty", {"RUNE_MEMORY_DIR": ""}, {}, {}),
|
||||
("rune_home", {"RUNE_MEMORY_DIR": None, "RUNE_HOME": str(home1)},
|
||||
{str(home1): "EB_HOME"}, {"EB_HOME_ROOT": str(home1)}),
|
||||
("tilde", {"RUNE_MEMORY_DIR": "~/eb-x", "RUNE_HOME": None, "HOME": str(home2)},
|
||||
{str(home2): "EB_HOME"}, {"EB_HOME_ROOT": str(home2)}),
|
||||
("dollar", {"RUNE_MEMORY_DIR": "$HOME/eb-v", "RUNE_HOME": None, "HOME": str(home1)},
|
||||
{str(home1): "EB_HOME"}, {"EB_HOME_ROOT": str(home1)}),
|
||||
]
|
||||
for name, env_over, subs, go_norms in variants:
|
||||
base = {k: v for k, v in os.environ.items()
|
||||
if k not in ("RUNE_MEMORY_DIR", "RUNE_HOME")}
|
||||
base["PYTHONPATH"] = str(REPO / "src")
|
||||
py_env = dict(base)
|
||||
for k, v in env_over.items():
|
||||
if v is None:
|
||||
py_env.pop(k, None)
|
||||
else:
|
||||
py_env[k] = v
|
||||
code = "import sys; sys.path.insert(0, %r); import rune.event_bus as eb; print(eb._store_path())" % str(REPO / "src")
|
||||
p = oracle_subproc(py_env, code)
|
||||
oracle_path = p.stdout.strip()
|
||||
go_env_v = go_env({**go_norms})
|
||||
for k, v in env_over.items():
|
||||
if v is None:
|
||||
go_env_v.pop(k, None)
|
||||
else:
|
||||
go_env_v[k] = v
|
||||
got = run_go(binpath, go_env_v, {"op": "store_path"})
|
||||
check(f"store_path:{name}",
|
||||
json.dumps(normalize(oracle_path, list(subs.items()))),
|
||||
got)
|
||||
|
||||
# ── event_types — vocabulary table ─────────────────────────────────
|
||||
check("event_types", json.dumps(eb.eb_event_types()),
|
||||
run_go(binpath, go_env(), {"op": "event_types"}))
|
||||
|
||||
# ── content_hash — any-shape payloads ──────────────────────────────
|
||||
HASH_PAYLOADS = [
|
||||
{}, {"k": "v"}, {"b": 2, "a": [1, 2.0]}, ["x", {"z": 1}],
|
||||
{"é": "ümlaut", "deep": {"z": 1, "a": [{"b": 2.0, "a": [True, None]}]}},
|
||||
"raw string payload", 1234, 2.5, 1e30, True, None,
|
||||
{"text": "quote\" inside\\ back\nnewline\ttab"},
|
||||
]
|
||||
for i, payload in enumerate(HASH_PAYLOADS):
|
||||
kw = {} if i == 0 else None # the {} payload rides via args
|
||||
check(f"content_hash:{i}",
|
||||
json.dumps(eb.eb_content_hash(payload if payload != {} or i > 0 else {}))
|
||||
if i or payload else json.dumps(eb.eb_content_hash({})),
|
||||
run_go(binpath, go_env(), {"op": "content_hash", "payload": payload}))
|
||||
|
||||
# ── emit — envelope build + durable append ─────────────────────────
|
||||
EMIT_CASES = [
|
||||
("user.message", "user", {"text": "hi"}),
|
||||
("tool.result", "", None), # bare defaults
|
||||
("rune.started", "system", {}), # {} falsy → hash ""
|
||||
("x.y", "", None), # → partition "system"
|
||||
("memory.created", "sys", {"nested": {"a": [1, 2, "x"]}, "flag": True, "nothing": None}),
|
||||
("user.message", "café", {"text": "quote\" inside\\ back\nnewline\ttab", "é": "ümlaut"}),
|
||||
("capability.failed", "sys", {"float_int": 1.0, "float_frac": 2.5, "big": 1e30}),
|
||||
("task.created", "sys", [1, "two", None, True]),
|
||||
("task.created", "sys", "raw string payload"),
|
||||
("task.created", "sys", 0),
|
||||
]
|
||||
for case in EMIT_CASES:
|
||||
t, src, payload = case
|
||||
check(f"emit:{t}/{str(src)[:20]}",
|
||||
json.dumps(eb.eb_emit(t, source=src, payload=payload)),
|
||||
run_go(binpath, go_env(), {"op": "emit", "event_type": t,
|
||||
"source": src, "payload": payload}))
|
||||
for bad in ({}, 5, None, 2.5, True, []):
|
||||
check(f"emit_valueerror:{type(bad).__name__}",
|
||||
json.dumps(eb.eb_emit("x", source=bad)),
|
||||
run_go(binpath, go_env(), {"op": "emit", "event_type": "x",
|
||||
"source": bad, "payload": None}))
|
||||
|
||||
# ── emit_line — the journal LINE bytes ──────────────────────────────
|
||||
for t, src, payload in (
|
||||
("user.message", "user", {"text": "hi"}),
|
||||
("session.ended", "microkernel", {"k": ["a", {"z": 2.0}]})):
|
||||
line = eb.eb_emit_line(t, source=src, payload=payload)
|
||||
check(f"emit_line:{t}",
|
||||
json.dumps(line),
|
||||
run_go(binpath, go_env(), {"op": "emit_line", "event_type": t,
|
||||
"source": src, "payload": payload}))
|
||||
|
||||
# ── log — truncation + filter matrix ────────────────────────────────
|
||||
LOG_CASES = [None, 2, 0, -1, -3, -1000, 100]
|
||||
for limit in LOG_CASES:
|
||||
check(f"log:{limit}",
|
||||
json.dumps(eb.eb_log(limit=limit, event_type=None)),
|
||||
run_go(binpath, go_env(), {"op": "log", "limit": limit, "event_type": None}))
|
||||
for limit, et in ((3, "user.message"), (3, "nope"), (2, "tool.result"), (0, "user.message")):
|
||||
check(f"log:{limit}:{et}",
|
||||
json.dumps(eb.eb_log(limit=limit, event_type=et)),
|
||||
run_go(binpath, go_env(), {"op": "log", "limit": limit, "event_type": et}))
|
||||
|
||||
# correlated corrupt-line raise (message-only contract)
|
||||
p = eb._store_path()
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text("not json\n")
|
||||
try:
|
||||
eb.log()
|
||||
oracle_corrupt = json.dumps({"error": None})
|
||||
except RuntimeError as e:
|
||||
oracle_corrupt = json.dumps({"error": "RuntimeError", "message": str(e)})
|
||||
check("log_corrupt", oracle_corrupt,
|
||||
run_go(binpath, go_env(), {"op": "replay"}))
|
||||
p.unlink() # the corrupt fixture must not leak into the cross-writer probe
|
||||
|
||||
# cross-writer probe: the ORACLE writes N events; Go replays the SAME file
|
||||
eb_emit = [eb.emit("user.message", source="user", payload={"i": i}) for i in range(3)]
|
||||
go_rows = run_go(binpath, go_env(), {"op": "replay"})
|
||||
check("replay_cross_writer",
|
||||
json.dumps({"rows": eb.log()}),
|
||||
go_rows)
|
||||
if len(json.loads(go_rows)["rows"]) != 3:
|
||||
fails += 1
|
||||
print("CROSS-WRITER probe: Go replayed != 3 rows")
|
||||
|
||||
# ── stats — first-occurrence by_type ────────────────────────────────
|
||||
check("stats", json.dumps(eb.eb_stats()),
|
||||
run_go(binpath, go_env(), {"op": "stats"}))
|
||||
|
||||
# ── handle_hook — prefix + payload shaping ──────────────────────────
|
||||
HOOK_CASES = [
|
||||
("on_startup", None), ("on_message", {"k": "v", "n": 5}),
|
||||
("weird.event", None),
|
||||
("on_response", '{"x": 1, "y": [2, 3]}'),
|
||||
("on_schedule", "not json at all"),
|
||||
("on_session_end", ""), ("on_session_end", 0), ("on_session_end", "0"),
|
||||
("on_session_end", "[1, 2]"),
|
||||
("on_session_end", '{"deep": {"z": 1, "a": ["ü"]}}'),
|
||||
("on_memory_add", {"long": "x" * 250, "uni": "café", "d": {"k": None},
|
||||
"l": [1, "two"], "b": True, "fl": 2.5, "ni": None}),
|
||||
("on_startup", {"empty": {}, "list": [], "str": ""}),
|
||||
("on_startup", {"floats": [1.0, 1200.0, 1e-7, 2.5]}),
|
||||
]
|
||||
for ev, data in HOOK_CASES:
|
||||
name = f"hook:{ev}:{str(data)[:24]}"
|
||||
check(name, json.dumps(eb.eb_handle_hook(ev, data)),
|
||||
run_go(binpath, go_env(), {"op": "handle_hook", "event": ev, "data": data}))
|
||||
|
||||
# ── main — captured CLI stdout ──────────────────────────────────────
|
||||
MAIN_CASES = [
|
||||
(["event_bus.py", "emit", "user.message", "--source", "user",
|
||||
"--payload", '{"text":"hi"}'], False),
|
||||
(["event_bus.py", "hook", "--event", "on_startup", "--data", '{"k":"v"}'], False),
|
||||
(["event_bus.py", "stats"], False),
|
||||
(["event_bus.py", "log", "--limit", "2"], False),
|
||||
(["event_bus.py", "emit", "session.started", "--json"], False),
|
||||
(["event_bus.py", "stats", "--json"], True),
|
||||
(["event_bus.py", "log"], True),
|
||||
(["event_bus.py", "hook"], False),
|
||||
(["event_bus.py", "emit"], False),
|
||||
]
|
||||
for i, (argv, seed) in enumerate(MAIN_CASES):
|
||||
kw = {"seed": True} if seed else {}
|
||||
check(f"main:{i}",
|
||||
json.dumps(eb.eb_main(argv, **kw)),
|
||||
run_go(binpath, go_env(), {"op": "main", "argv": argv, "seed": seed}))
|
||||
|
||||
# ── negative control — a perturbed comparison MUST diverge ─────────
|
||||
probe = run_go(binpath, go_env(), {"op": "stats"})
|
||||
mutated = probe.replace(":2", ":999", 1) if ":2" in probe else probe + "x"
|
||||
if mutated == probe:
|
||||
print("NEGATIVE CONTROL FAILED TO DETECT")
|
||||
return 2
|
||||
print(f"parity: {total - fails}/{total} match (negative control: detected)")
|
||||
return 1 if fails else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,204 @@
|
||||
"""eb_freeze_seams — sitecustomize rig for freezing event_bus truth.
|
||||
|
||||
Auto-imported on interpreter start (PYTHONPATH=scripts/eb_freeze_seams:src)
|
||||
by BOTH the truth generator (scripts/gen_event_bus_truth.py) and the
|
||||
shared harness's --verify mode, so frozen truth is byte-reproducible in any
|
||||
process (dichotomy wave #2479 gap close, contract factory-gap-eb):
|
||||
|
||||
* datetime.datetime is FROZEN to EB_FROZEN_NOW — the oracle stamps the
|
||||
envelope timestamp through rune.orchestration.records._now (a bound
|
||||
import the module-level pin cannot reach). The class patch runs before
|
||||
either module imports.
|
||||
* uuid.uuid4 is a deterministic counter (ee_freeze_seams precedent) —
|
||||
records.new_id mints "evt_" + uuid4().hex[:12]. The counter RESETS per
|
||||
seam call so a case's ids depend only on the mints inside it.
|
||||
* RUNE_MEMORY_DIR is pinned to a hermetic mkdtemp sandbox BEFORE
|
||||
rune.event_bus imports — journal writes never touch the repo, and
|
||||
event_bus.EVENT_LOG (an import-time constant) resolves to the sandbox.
|
||||
* The seam wrappers (eb_store_path, eb_event_types, eb_content_hash,
|
||||
eb_emit, eb_emit_line, eb_log, eb_log_corrupt, eb_stats,
|
||||
eb_handle_hook, eb_main) are bound onto rune.event_bus so the shared
|
||||
harness's getattr(module, fn) freezes the event-bus surface: each one
|
||||
runs the REAL oracle fns and returns JSON-safe snapshots (journal
|
||||
replays, captured journal line bytes, captured CLI stdout, and the
|
||||
loud ValueError message-only contract).
|
||||
* Machine paths never enter truth: absolute roots are normalized to
|
||||
EB_SANDBOX / EB_HOME / REPO_ROOT placeholders (paths quirk law).
|
||||
"""
|
||||
|
||||
import contextlib # noqa: E402
|
||||
import io # noqa: E402
|
||||
import itertools # noqa: E402
|
||||
import os # noqa: E402
|
||||
import sys # noqa: E402
|
||||
import tempfile # noqa: E402
|
||||
import uuid as _uuid # noqa: E402
|
||||
|
||||
_REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
sys.path.insert(0, os.path.join(_REPO, "scripts"))
|
||||
sys.path.insert(0, os.path.join(_REPO, "src"))
|
||||
|
||||
from freeze_rigs import pin_clock # noqa: E402
|
||||
|
||||
# EB_FROZEN_NOW: the pinned instant every clock site reports.
|
||||
FROZEN_NOW = "2026-10-14T11:22:33.123456+00:00"
|
||||
pin_clock(FROZEN_NOW)
|
||||
|
||||
_MINT = itertools.count(1)
|
||||
|
||||
|
||||
class _DetUUID:
|
||||
"""uuid4() stand-in whose .hex (records.new_id) and str() (legacy
|
||||
default_factory) are pure functions of the counter."""
|
||||
|
||||
def __init__(self, n: int):
|
||||
self.hex = f"{n:032x}"
|
||||
self.int = int(self.hex, 16)
|
||||
|
||||
def __str__(self):
|
||||
h = self.hex
|
||||
return f"{h[:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:]}"
|
||||
|
||||
|
||||
def _reset_mint():
|
||||
global _MINT
|
||||
_MINT = itertools.count(1)
|
||||
|
||||
|
||||
_uuid.uuid4 = lambda: _DetUUID(next(_MINT)) # type: ignore[assignment]
|
||||
|
||||
# Hermetic journal sandbox BEFORE rune.event_bus imports — EVENT_LOG is an
|
||||
# import-time constant (os.path.join(memory_dir(), ...)), so the env pin
|
||||
# must precede the import. The sandbox path itself never enters truth.
|
||||
_SANDBOX = tempfile.mkdtemp(prefix="eb-freeze-mem-")
|
||||
os.environ["RUNE_MEMORY_DIR"] = _SANDBOX
|
||||
|
||||
import rune.event_bus as eb # noqa: E402 (after the clock + uuid + env pins)
|
||||
|
||||
# The three seed events (the legacy suite's a/a/b shape): ids ride the
|
||||
# counter (1..3 -> flattened evt_000000000000), types/types order pinned.
|
||||
SEEDS = [
|
||||
("user.message", "user", {"text": "hi"}),
|
||||
("tool.result", "", None),
|
||||
("user.message", "user", {"n": 1}),
|
||||
]
|
||||
|
||||
|
||||
def _seed():
|
||||
for t, src, payload in SEEDS:
|
||||
eb.emit(t, source=src, payload=payload)
|
||||
|
||||
|
||||
def _normalize(p) -> str:
|
||||
s = str(p)
|
||||
s = s.replace(_SANDBOX, "EB_SANDBOX")
|
||||
home = os.environ.get("EB_FREEZE_HOME", "")
|
||||
if home:
|
||||
s = s.replace(home, "EB_HOME")
|
||||
s = s.replace(_REPO, "REPO_ROOT")
|
||||
return s
|
||||
|
||||
|
||||
def _clear_journal():
|
||||
p = eb._store_path()
|
||||
if p.exists():
|
||||
p.unlink()
|
||||
|
||||
|
||||
def eb_store_path():
|
||||
return _normalize(eb._store_path())
|
||||
|
||||
|
||||
def eb_event_types():
|
||||
return dict(eb.EVENT_TYPES)
|
||||
|
||||
|
||||
def eb_content_hash(payload):
|
||||
return eb._content_hash(payload)
|
||||
|
||||
|
||||
def eb_emit(event_type, source="", payload=None):
|
||||
_reset_mint()
|
||||
_clear_journal()
|
||||
try:
|
||||
d = eb.emit(event_type, source=source, payload=payload)
|
||||
except ValueError as e:
|
||||
return {"error": "ValueError", "message": str(e), "journal": eb.log()}
|
||||
return {"emitted": d, "journal": eb.log()}
|
||||
|
||||
|
||||
def eb_emit_line(event_type, source="", payload=None):
|
||||
_reset_mint()
|
||||
_clear_journal()
|
||||
eb.emit(event_type, source=source, payload=payload)
|
||||
return eb._store_path().read_text()
|
||||
|
||||
|
||||
def eb_log(limit=None, event_type=None):
|
||||
_reset_mint()
|
||||
_clear_journal()
|
||||
_seed()
|
||||
return {"rows": eb.log(limit=limit, event_type=event_type)}
|
||||
|
||||
|
||||
def eb_log_corrupt():
|
||||
_reset_mint()
|
||||
_clear_journal()
|
||||
p = eb._store_path()
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text("not json\n")
|
||||
try:
|
||||
eb.log()
|
||||
return {"error": None}
|
||||
except RuntimeError as e:
|
||||
return {"error": "RuntimeError", "message": str(e)}
|
||||
|
||||
|
||||
def eb_stats():
|
||||
_reset_mint()
|
||||
_clear_journal()
|
||||
_seed()
|
||||
total, by_type = eb.stats()
|
||||
return {"total": total, "by_type": by_type}
|
||||
|
||||
|
||||
def eb_handle_hook(event, data=None):
|
||||
_reset_mint()
|
||||
_clear_journal()
|
||||
try:
|
||||
r = eb.handle_hook(event, data)
|
||||
except ValueError as e:
|
||||
return {"hook": {"error": "ValueError", "message": str(e)}}
|
||||
return {"hook": r, "journal": eb.log()}
|
||||
|
||||
|
||||
def eb_main(argv, seed=False):
|
||||
_reset_mint()
|
||||
_clear_journal()
|
||||
if seed:
|
||||
_seed()
|
||||
old = sys.argv
|
||||
buf = io.StringIO()
|
||||
code = 0
|
||||
try:
|
||||
sys.argv = list(argv)
|
||||
with contextlib.redirect_stdout(buf):
|
||||
eb.main()
|
||||
except SystemExit as e:
|
||||
code = e.code if isinstance(e.code, int) else 0
|
||||
finally:
|
||||
sys.argv = old
|
||||
return {"stdout": buf.getvalue(), "exit": code}
|
||||
|
||||
|
||||
eb.eb_store_path = eb_store_path
|
||||
eb.eb_event_types = eb_event_types
|
||||
eb.eb_content_hash = eb_content_hash
|
||||
eb.eb_emit = eb_emit
|
||||
eb.eb_emit_line = eb_emit_line
|
||||
eb.eb_log = eb_log
|
||||
eb.eb_log_corrupt = eb_log_corrupt
|
||||
eb.eb_stats = eb_stats
|
||||
eb.eb_handle_hook = eb_handle_hook
|
||||
eb.eb_main = eb_main
|
||||
eb.EB_SANDBOX = _SANDBOX
|
||||
+124
-19
@@ -1,35 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""gen_event_bus_truth.py — factory batch 9 truth battery (live oracle)."""
|
||||
"""gen_event_bus_truth.py — shared-harness truth battery for the event_bus
|
||||
gap closure (#2479, contract factory-gap-eb).
|
||||
|
||||
import json
|
||||
Uses scripts/freeze_truth.py (the ONE truth law) under the
|
||||
scripts/eb_freeze_seams sitecustomize rig (PYTHONPATH carries BOTH) to
|
||||
freeze src/rune/event_bus.py behavior for the Go port at src/go/event-bus:
|
||||
|
||||
* _content_hash — the orchestration sha256: convention over any payload
|
||||
shape (sort_keys=True is recursive, ", "/": " separators, ensure_ascii).
|
||||
* eb_store_path — the canonical journal path under the hermetic RUNE_MEMORY_DIR
|
||||
sandbox, normalized to EB_SANDBOX (machine paths never enter truth).
|
||||
* eb_event_types — EVENT_TYPES in definition order (json.dumps keeps it).
|
||||
* eb_emit — envelope build + durable append: schema_version-first to_dict,
|
||||
partition_key from source-or-system, payload_hash truthiness ({}, 0 and
|
||||
None hash ""), minted event_id / frozen clock, and the loud ValueError
|
||||
branches (non-str source: dict/int/NoneType/float/bool/list — message
|
||||
carries the Python type name, NOTHING written).
|
||||
* eb_emit_line — the exact journal LINE bytes (json.dumps(to_dict,
|
||||
sort_keys=True) + newline).
|
||||
* eb_log — replay + event_type filter + events[-limit:] truncation
|
||||
(limit 0 → ALL — the -0 slice quirk; -1 → [1:] drops the first).
|
||||
* eb_log_corrupt — the typed corrupt-line raise (message-only contract).
|
||||
* eb_stats — (total, by_type) with by_type in first-occurrence order.
|
||||
* eb_handle_hook — hook.<event> vocabulary prefix, source microkernel,
|
||||
payload shaping (str → json.loads with the raw[:200] fallback; dict →
|
||||
str(v)[:200] CODEPOINT truncation with Python repr forms; falsy data
|
||||
stays {}), and the journal replay after each hook.
|
||||
* eb_main — captured CLI stdout + exit code under the pins (emit/log/stats/
|
||||
hook, --json modes, the usage line for a bare emit).
|
||||
|
||||
Verify (byte-identical law):
|
||||
PYTHONPATH=scripts/eb_freeze_seams:src \\
|
||||
python3 scripts/freeze_truth.py --verify src/go/event-bus/truth_bus.json \\
|
||||
--module rune.event_bus
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO / "src"))
|
||||
sys.path.insert(0, str(REPO / "src" / "rune"))
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
import event_bus as eb # noqa: E402
|
||||
FROZEN_NOW = "2026-10-14T11:22:33.123456+00:00"
|
||||
os.environ["RUNE_FROZEN_NOW"] = FROZEN_NOW
|
||||
|
||||
from freeze_truth import freeze, write_envelope # noqa: E402
|
||||
|
||||
|
||||
def main():
|
||||
payloads = [
|
||||
{},
|
||||
{"k": "v"},
|
||||
{"task_id": "t1", "status": "failed", "count": 3},
|
||||
{"nested": {"a": [1, 2, "x"]}, "flag": True, "nothing": None},
|
||||
{"text": "quote\" inside\\ back\nnewline\ttab"},
|
||||
{"float_int": 1.0, "float_frac": 2.5},
|
||||
]
|
||||
cases = [
|
||||
{"fn": "hash", "name": f"payload_{i}", "input": {"payload": p},
|
||||
"expected": {"hash": eb._content_hash(p)}}
|
||||
for i, p in enumerate(payloads)
|
||||
# ── _content_hash — direct oracle rows ──────────────────────────
|
||||
("_content_hash", ({},), {}),
|
||||
("_content_hash", ({"k": "v"},), {}),
|
||||
("_content_hash", ({"task_id": "t1", "status": "failed", "count": 3},), {}),
|
||||
("_content_hash", ({"nested": {"a": [1, 2, "x"]}, "flag": True, "nothing": None},), {}),
|
||||
("_content_hash", ({"text": "quote\" inside\\ back\nnewline\ttab"},), {}),
|
||||
("_content_hash", ({"é": "ümlaut", "b": {"z": 1, "a": 2.0}},), {}),
|
||||
("_content_hash", ([1, "two", None, True],), {}),
|
||||
("_content_hash", ({"float_int": 1.0, "float_frac": 2.5, "big": 1e30},), {}),
|
||||
("_content_hash", ("raw string payload",), {}),
|
||||
("_content_hash", ({"z": 1, "a": 2},), {}), # sort_keys reorders
|
||||
# ── eb_store_path / eb_event_types ─────────────────────────────
|
||||
("eb_store_path", (), {}),
|
||||
("eb_event_types", (), {}),
|
||||
# ── eb_emit — envelope build + durable append ──────────────────
|
||||
("eb_emit", ("user.message",), {"source": "user", "payload": {"text": "hi"}}),
|
||||
("eb_emit", ("tool.result",), {}), # bare defaults
|
||||
("eb_emit", ("rune.started",), {"source": "system", "payload": {}}), # {} falsy
|
||||
("eb_emit", ("x.y",), {"source": "", "payload": None}), # system fallback
|
||||
("eb_emit", ("memory.created",), {"source": "sys",
|
||||
"payload": {"nested": {"a": [1, 2, "x"]}, "flag": True, "nothing": None}}),
|
||||
("eb_emit", ("user.message",), {"source": "café",
|
||||
"payload": {"text": "quote\" inside\\ back\nnewline\ttab", "é": "ümlaut"}}),
|
||||
("eb_emit", ("capability.failed",), {"source": "sys",
|
||||
"payload": {"float_int": 1.0, "float_frac": 2.5, "big": 1e30}}),
|
||||
("eb_emit", ("task.created",), {"source": "sys", "payload": [1, "two", None, True]}),
|
||||
("eb_emit", ("task.created",), {"source": "sys", "payload": "raw string payload"}),
|
||||
("eb_emit", ("task.created",), {"source": "sys", "payload": 0}), # falsy int
|
||||
# ── eb_emit — loud ValueError + nothing written ────────────────
|
||||
("eb_emit", ("x",), {"source": {}, "payload": None}),
|
||||
("eb_emit", ("x",), {"source": 5, "payload": None}),
|
||||
("eb_emit", ("x",), {"source": None, "payload": None}),
|
||||
("eb_emit", ("x",), {"source": 2.5, "payload": None}),
|
||||
("eb_emit", ("x",), {"source": True, "payload": None}),
|
||||
("eb_emit", ("x",), {"source": [], "payload": None}),
|
||||
# ── eb_emit_line — the journal LINE bytes ───────────────────────
|
||||
("eb_emit_line", ("user.message",), {"source": "user", "payload": {"text": "hi"}}),
|
||||
("eb_emit_line", ("session.ended",), {"source": "microkernel",
|
||||
"payload": {"k": ["a", {"z": 2.0}]}}),
|
||||
# ── eb_log — filter + [-limit:] truncation matrix ──────────────
|
||||
("eb_log", (), {"limit": None, "event_type": None}),
|
||||
("eb_log", (), {"limit": 2, "event_type": None}),
|
||||
("eb_log", (), {"limit": 0, "event_type": None}), # -0 → ALL
|
||||
("eb_log", (), {"limit": -1, "event_type": None}), # [1:] quirk
|
||||
("eb_log", (), {"limit": 100, "event_type": None}),
|
||||
("eb_log", (), {"limit": 3, "event_type": "user.message"}),
|
||||
("eb_log", (), {"limit": 3, "event_type": "nope"}),
|
||||
("eb_log", (), {"limit": 2, "event_type": "tool.result"}),
|
||||
("eb_log_corrupt", (), {}),
|
||||
# ── eb_stats ────────────────────────────────────────────────────
|
||||
("eb_stats", (), {}),
|
||||
# ── eb_handle_hook — prefix + payload shaping ──────────────────
|
||||
("eb_handle_hook", ("on_startup",), {}),
|
||||
("eb_handle_hook", ("on_message",), {"data": {"k": "v", "n": 5}}),
|
||||
("eb_handle_hook", ("weird.event",), {}),
|
||||
("eb_handle_hook", ("on_response",), {"data": '{"x": 1, "y": [2, 3]}'}),
|
||||
("eb_handle_hook", ("on_schedule",), {"data": "not json at all"}),
|
||||
("eb_handle_hook", ("hook.on_startup",), {}),
|
||||
("eb_handle_hook", ("on_session_end",), {"data": ""}),
|
||||
("eb_handle_hook", ("on_session_end",), {"data": 0}),
|
||||
("eb_handle_hook", ("on_session_end",), {"data": "0"}),
|
||||
("eb_handle_hook", ("on_memory_add",),
|
||||
{"data": {"long": "x" * 250, "uni": "café", "d": {"k": None},
|
||||
"l": [1, "two"], "b": True, "fl": 2.5, "ni": None}}),
|
||||
("eb_handle_hook", ("on_session_end",), {"data": "[1, 2]"}),
|
||||
("eb_handle_hook", ("on_session_end",), {"data": '{"deep": {"z": 1, "a": ["ü"]}}'}),
|
||||
# ── eb_main — captured CLI stdout ───────────────────────────────
|
||||
("eb_main", (["event_bus.py", "emit", "user.message", "--source", "user",
|
||||
"--payload", '{"text":"hi"}'],), {}),
|
||||
("eb_main", (["event_bus.py", "hook", "--event", "on_startup",
|
||||
"--data", '{"k":"v"}'],), {}),
|
||||
("eb_main", (["event_bus.py", "stats"],), {}),
|
||||
("eb_main", (["event_bus.py", "log", "--limit", "2"],), {}),
|
||||
("eb_main", (["event_bus.py", "emit", "session.started", "--json"],), {}),
|
||||
("eb_main", (["event_bus.py", "stats", "--json"],), {"seed": True}),
|
||||
("eb_main", (["event_bus.py", "log"],), {"seed": True}),
|
||||
("eb_main", (["event_bus.py", "hook"],), {}),
|
||||
("eb_main", (["event_bus.py", "emit"],), {}),
|
||||
]
|
||||
out = {"oracle": "event_bus._content_hash", "generated": "2026-10-07", "cases": cases}
|
||||
(REPO / "src/go/event-bus" / "truth_bus.json").write_text(json.dumps(out, indent=1) + "\n")
|
||||
print(f"wrote {len(cases)} cases")
|
||||
truth = freeze("rune.event_bus", cases, oracle_module="rune.event_bus")
|
||||
dest = REPO / "src/go/event-bus/truth_bus.json"
|
||||
write_envelope(FROZEN_NOW, truth, dest)
|
||||
print(f"wrote {len(truth)} cases -> {dest}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
main()
|
||||
+3
-29
@@ -53,13 +53,9 @@ func writeEventJSON(sb *strings.Builder, v interface{}) {
|
||||
sb.WriteString(pyQuoteEvent(x))
|
||||
case float64:
|
||||
// PARITY: Python json.dumps renders 1.0 as "1.0" (floats keep
|
||||
// .0; ints stay ints). The oracle fixture pinned this.
|
||||
if x == float64(int64(x)) && x < 1e16 && x > -1e16 {
|
||||
sb.WriteString(itoaE(int64(x)))
|
||||
sb.WriteString(".0")
|
||||
} else {
|
||||
sb.WriteString(jsonNumber(x))
|
||||
}
|
||||
// .0; ints stay ints) and repr-exponent forms match repr(float)
|
||||
// (pyFloatRepr — 2-digit exponents, Infinity/NaN names).
|
||||
sb.WriteString(pyFloatRepr(x))
|
||||
case []interface{}:
|
||||
sb.WriteString("[")
|
||||
for i, e := range x {
|
||||
@@ -104,19 +100,6 @@ func writeEventJSON(sb *strings.Builder, v interface{}) {
|
||||
}
|
||||
}
|
||||
|
||||
func trimFloatE(f float64) string {
|
||||
s := fmtFloatE(f)
|
||||
return s
|
||||
}
|
||||
|
||||
func fmtFloatE(f float64) string {
|
||||
// Python repr semantics: integral → no decimal point
|
||||
if f == float64(int64(f)) && f < 1e16 && f > -1e16 {
|
||||
return itoaE(int64(f))
|
||||
}
|
||||
return strconvFormatE(f)
|
||||
}
|
||||
|
||||
func itoaE(n int64) string {
|
||||
if n == 0 {
|
||||
return "0"
|
||||
@@ -136,15 +119,6 @@ func itoaE(n int64) string {
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func strconvFormatE(f float64) string {
|
||||
return strings.TrimSpace(strings.Replace(jsonNumber(f), "e+", "e+", 1))
|
||||
}
|
||||
|
||||
func jsonNumber(f float64) string {
|
||||
b, _ := json.Marshal(f)
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func pyQuoteEvent(s string) string {
|
||||
var sb strings.Builder
|
||||
sb.WriteByte('"')
|
||||
|
||||
+450
-257
@@ -1,26 +1,26 @@
|
||||
// event-bus — the Go reactive event journal (R2 W4 slice 1).
|
||||
// Replaces src/rune/event_bus.py: same events.jsonl file, same envelope
|
||||
// format, same CLI surface. Go goroutines + sync.Mutex for concurrent
|
||||
// append safety; os.WriteFile for the durable journal.
|
||||
// event-bus — the Go reactive event journal (R2 W4 slice 1; gap-closed for
|
||||
// the dichotomy wave). Replaces src/rune/event_bus.py: same events.jsonl
|
||||
// file, same envelope format, same CLI surface.
|
||||
//
|
||||
// The journal I/O, StorePath and HandleHook live in store.go (they delegate
|
||||
// to the canonical JSONL store shape). envelope.go carries event_envelope's
|
||||
// port; the "op" mode is the parity surface the freeze/parity harnesses
|
||||
// drive under the EE_FROZEN_NOW / EE_UUID_BASE pins.
|
||||
//
|
||||
// Usage:
|
||||
//
|
||||
// event-bus emit <event_type> [--source <s>] [--payload <json>]
|
||||
// event-bus log [--limit N] [--type <t>]
|
||||
// event-bus stats
|
||||
// event-bus emit <event_type> [--source <s>] [--payload <json>] [--json]
|
||||
// event-bus log [--limit N] [--type <t>] [--json]
|
||||
// event-bus stats [--json]
|
||||
// event-bus hook --event <e> --data <json>
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// EventEnvelope — the canonical record (matches rune/orchestration/records.py).
|
||||
@@ -36,248 +36,275 @@ type EventEnvelope struct {
|
||||
Timestamp string `json:"timestamp"`
|
||||
}
|
||||
|
||||
func newID(prefix string) string {
|
||||
now := time.Now().UTC()
|
||||
nonce := fmt.Sprintf("%x", now.UnixNano())
|
||||
return fmt.Sprintf("%s-%s", prefix, nonce[:12])
|
||||
}
|
||||
|
||||
func nowISO() string {
|
||||
return time.Now().UTC().Format("2006-01-02T15:04:05.000000Z")
|
||||
}
|
||||
|
||||
func journalPath() string {
|
||||
home := os.Getenv("RUNE_MEMORY_DIR")
|
||||
if home == "" {
|
||||
home = filepath.Join(os.Getenv("HOME"), ".rune")
|
||||
}
|
||||
return filepath.Join(home, "journal", "events.jsonl")
|
||||
}
|
||||
|
||||
func contentHash(payload map[string]interface{}) string {
|
||||
if payload == nil {
|
||||
return ""
|
||||
}
|
||||
keys := make([]string, 0, len(payload))
|
||||
for k := range payload {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
parts := make([]string, len(keys))
|
||||
for i, k := range keys {
|
||||
v, _ := payload[k]
|
||||
parts[i] = fmt.Sprintf("%q: %v", k, v)
|
||||
}
|
||||
h := sha256.Sum256([]byte("{" + strings.Join(parts, ", ") + "}"))
|
||||
return "sha256:" + hex.EncodeToString(h[:])
|
||||
}
|
||||
|
||||
func emit(eventType, source string, payload map[string]interface{}) EventEnvelope {
|
||||
env := EventEnvelope{
|
||||
EventID: newID("evt"),
|
||||
EventType: eventType,
|
||||
PartitionKey: source,
|
||||
Timestamp: nowISO(),
|
||||
}
|
||||
if source == "" {
|
||||
env.PartitionKey = "system"
|
||||
}
|
||||
if payload != nil {
|
||||
env.PayloadHash = contentHash(payload)
|
||||
}
|
||||
path := journalPath()
|
||||
os.MkdirAll(filepath.Dir(path), 0755)
|
||||
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus: cannot open journal: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer f.Close()
|
||||
data, _ := json.Marshal(env)
|
||||
f.Write(append(data, '\n'))
|
||||
return env
|
||||
}
|
||||
|
||||
func replay() []EventEnvelope {
|
||||
path := journalPath()
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
defer f.Close()
|
||||
var events []EventEnvelope
|
||||
scanner := bufio.NewScanner(f)
|
||||
scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
|
||||
for scanner.Scan() {
|
||||
line := strings.TrimSpace(scanner.Text())
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
var env EventEnvelope
|
||||
if err := json.Unmarshal([]byte(line), &env); err != nil {
|
||||
continue // skip malformed lines
|
||||
}
|
||||
events = append(events, env)
|
||||
}
|
||||
return events
|
||||
}
|
||||
|
||||
func main() {
|
||||
if len(os.Args) < 2 {
|
||||
fmt.Fprintln(os.Stderr, "usage: event-bus <emit|log|stats> [args...]")
|
||||
os.Exit(1)
|
||||
argv := append([]string{"event-bus"}, os.Args[1:]...) // runMain skips argv[0]
|
||||
out, code := runMain(argv)
|
||||
fmt.Print(out)
|
||||
if code != 0 {
|
||||
os.Exit(code)
|
||||
}
|
||||
}
|
||||
|
||||
switch os.Args[1] {
|
||||
// runMain — event_bus.main over argv (op mode replays this with minted
|
||||
// ids/clock pinned); argv[0] is the script name, argparse-style. Returns
|
||||
// stdout bytes and the process exit code.
|
||||
func runMain(argv []string) (string, int) {
|
||||
if len(argv) > 0 {
|
||||
argv = argv[1:] // skip the program name (argparse consumes sys.argv[1:])
|
||||
}
|
||||
var sb strings.Builder
|
||||
say := func(format string, a ...interface{}) {
|
||||
sb.WriteString(fmt.Sprintf(format, a...))
|
||||
sb.WriteString("\n")
|
||||
}
|
||||
if len(argv) < 1 {
|
||||
say("usage: event-bus <emit|log|stats> [args...]")
|
||||
return sb.String(), 1
|
||||
}
|
||||
cmd := argv[0]
|
||||
flags := parseFlags(argv[1:])
|
||||
|
||||
switch cmd {
|
||||
case "emit":
|
||||
if len(os.Args) < 3 {
|
||||
fmt.Fprintln(os.Stderr, "usage: event-bus emit <event_type> [--source <s>] [--payload <json>]")
|
||||
os.Exit(1)
|
||||
if flags.rest == "" {
|
||||
say("usage: event_bus.py emit <event_type>")
|
||||
return sb.String(), 0
|
||||
}
|
||||
args := os.Args[2:] // skip "event-bus" and "emit"
|
||||
eventType := args[0]
|
||||
source := ""
|
||||
var payload map[string]interface{}
|
||||
for i := 1; i < len(args); i++ {
|
||||
if args[i] == "--source" && i+1 < len(args) {
|
||||
source = args[i+1]
|
||||
i++
|
||||
}
|
||||
if args[i] == "--payload" && i+1 < len(args) {
|
||||
json.Unmarshal([]byte(args[i+1]), &payload)
|
||||
i++
|
||||
var payload interface{}
|
||||
if flags.payload != "" {
|
||||
dec := json.NewDecoder(strings.NewReader(flags.payload))
|
||||
dec.UseNumber()
|
||||
v, err := decodeOrderedValue(dec)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus: bad --payload: %v\n", err)
|
||||
return sb.String(), 1
|
||||
}
|
||||
payload = v
|
||||
}
|
||||
env, err := Emit(flags.rest, flags.source, payload)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus: %v\n", err)
|
||||
return sb.String(), 1
|
||||
}
|
||||
if flags.jsonOut {
|
||||
say("%s", pyDumpsIndent(env, 0))
|
||||
} else {
|
||||
say("Emitted %s (%s)", opStr(env, "event_type"), opStr(env, "event_id"))
|
||||
}
|
||||
env := emit(eventType, source, payload)
|
||||
out, _ := json.MarshalIndent(env, "", " ")
|
||||
fmt.Println(string(out))
|
||||
|
||||
case "log":
|
||||
limit := 20
|
||||
eventType := ""
|
||||
args := os.Args[2:] // skip "event-bus" and "log"
|
||||
for i := 0; i < len(args); i++ {
|
||||
if args[i] == "--limit" && i+1 < len(args) {
|
||||
fmt.Sscanf(args[i+1], "%d", &limit)
|
||||
i++
|
||||
}
|
||||
if args[i] == "--type" && i+1 < len(args) {
|
||||
eventType = args[i+1]
|
||||
i++
|
||||
}
|
||||
events, err := LogRows(json.Number(fmt.Sprintf("%d", flags.limit)), flagOrNil(flags.evType))
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus: %v\n", err)
|
||||
return sb.String(), 1
|
||||
}
|
||||
events := replay()
|
||||
if eventType != "" {
|
||||
var filtered []EventEnvelope
|
||||
if flags.jsonOut {
|
||||
arr := make([]interface{}, len(events))
|
||||
for i, e := range events {
|
||||
arr[i] = e
|
||||
}
|
||||
say("%s", pyDumpsIndent(arr, 0))
|
||||
} else {
|
||||
for _, e := range events {
|
||||
if e.EventType == eventType {
|
||||
filtered = append(filtered, e)
|
||||
ts := "?"
|
||||
if v, ok := e.vals["timestamp"]; ok {
|
||||
ts = truncateRunes(v.(string), 19)
|
||||
}
|
||||
et := "?"
|
||||
if v, ok := e.vals["event_type"]; ok {
|
||||
et = pyStrValue(v)
|
||||
}
|
||||
pk := "?"
|
||||
if v, ok := e.vals["partition_key"]; ok {
|
||||
pk = pyStrValue(v)
|
||||
}
|
||||
say(" %s [%-25s] %s", ts, et, pk)
|
||||
}
|
||||
events = filtered
|
||||
}
|
||||
if limit > 0 && len(events) > limit {
|
||||
events = events[len(events)-limit:]
|
||||
}
|
||||
for _, e := range events {
|
||||
ts := strings.SplitN(e.Timestamp, ".", 2)[0]
|
||||
fmt.Printf(" %s [%-25s] %s\n", ts, e.EventType, e.PartitionKey)
|
||||
}
|
||||
|
||||
case "stats":
|
||||
events := replay()
|
||||
byType := make(map[string]int)
|
||||
for _, e := range events {
|
||||
byType[e.EventType]++
|
||||
}
|
||||
var fileSize int64
|
||||
if fi, err := os.Stat(journalPath()); err == nil {
|
||||
fileSize = fi.Size()
|
||||
}
|
||||
out, _ := json.MarshalIndent(map[string]interface{}{
|
||||
"total": len(events),
|
||||
"types": byType,
|
||||
"file_size": fileSize,
|
||||
}, "", " ")
|
||||
fmt.Println(string(out))
|
||||
|
||||
case "create-envelope":
|
||||
args := os.Args[2:]
|
||||
source, channel, text := "", "", ""
|
||||
var payload map[string]interface{}
|
||||
for i := 0; i < len(args); i++ {
|
||||
switch args[i] {
|
||||
case "--source":
|
||||
if i+1 < len(args) {
|
||||
source = args[i+1]
|
||||
i++
|
||||
}
|
||||
case "--channel":
|
||||
if i+1 < len(args) {
|
||||
channel = args[i+1]
|
||||
i++
|
||||
}
|
||||
case "--text":
|
||||
if i+1 < len(args) {
|
||||
text = args[i+1]
|
||||
i++
|
||||
}
|
||||
case "--payload":
|
||||
if i+1 < len(args) {
|
||||
json.Unmarshal([]byte(args[i+1]), &payload)
|
||||
i++
|
||||
total, byType := statsCounts()
|
||||
if flags.jsonOut {
|
||||
fileSize := journalFileSize()
|
||||
out := ordKV("total", json.Number(itoaE(int64(total))), "types", byType,
|
||||
"file_size", json.Number(itoaE(fileSize)))
|
||||
say("%s", pyDumpsIndent(out, 0))
|
||||
} else {
|
||||
say("Events: %d", total)
|
||||
say("File size: %s bytes", commaInt(int64(journalFileSize())))
|
||||
if byType != nil && len(byType.keys) > 0 {
|
||||
for _, it := range stableTopTypes(byType) {
|
||||
say(" %-30s %s", opStr(it, "key"), opStr(it, "count"))
|
||||
}
|
||||
}
|
||||
}
|
||||
if payload == nil {
|
||||
payload = map[string]interface{}{}
|
||||
|
||||
case "hook":
|
||||
event := flags.event
|
||||
if event == "" {
|
||||
event = "unknown"
|
||||
}
|
||||
if text != "" {
|
||||
payload["text"] = text
|
||||
var data interface{}
|
||||
if flags.data != "" {
|
||||
dec := json.NewDecoder(strings.NewReader(flags.data))
|
||||
dec.UseNumber()
|
||||
v, err := decodeOrderedValue(dec)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus: bad --data: %v\n", err)
|
||||
return sb.String(), 1
|
||||
}
|
||||
data = v
|
||||
}
|
||||
env := createEnvelope(source, channel, payload, nil, nil, nil)
|
||||
result := HandleHook(event, data)
|
||||
say("%s", pyDumpsValue(result))
|
||||
|
||||
case "create-envelope":
|
||||
var payload interface{}
|
||||
if flags.payload != "" {
|
||||
dec := json.NewDecoder(strings.NewReader(flags.payload))
|
||||
dec.UseNumber()
|
||||
v, err := decodeOrderedValue(dec)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus: bad --payload: %v\n", err)
|
||||
return sb.String(), 1
|
||||
}
|
||||
payload = v
|
||||
}
|
||||
if flags.text != "" {
|
||||
o, ok := payload.(*ordObj)
|
||||
if !ok {
|
||||
o = newOrdObj()
|
||||
payload = o
|
||||
}
|
||||
o.set("text", flags.text)
|
||||
}
|
||||
env := createEnvelope(flags.source, flags.channel, payload, nil, nil, nil)
|
||||
out, _ := json.MarshalIndent(env.ToDict(), "", " ")
|
||||
fmt.Println(string(out))
|
||||
say("%s", string(out))
|
||||
|
||||
case "op":
|
||||
// op — JSON-in/JSON-out parity surface (dichotomy gap close):
|
||||
// the freeze/parity harnesses drive every ported fn through
|
||||
// here under the EE_FROZEN_NOW / EE_UUID_BASE pins.
|
||||
dec := json.NewDecoder(os.Stdin)
|
||||
dec.UseNumber()
|
||||
v, err := decodeOrderedValue(dec)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus op: bad request: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
req, ok := v.(*ordObj)
|
||||
if !ok {
|
||||
fmt.Fprintf(os.Stderr, "event-bus op: request must be an object\n")
|
||||
os.Exit(1)
|
||||
}
|
||||
pyJSON, err := runOp(req)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus op: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
var semantic interface{}
|
||||
dec2 := json.NewDecoder(strings.NewReader(pyJSON))
|
||||
dec2.UseNumber()
|
||||
dec2.Decode(&semantic)
|
||||
out, _ := json.Marshal(map[string]interface{}{
|
||||
"json": pyJSON,
|
||||
"result": semantic,
|
||||
})
|
||||
fmt.Println(string(out))
|
||||
return opMode(), 0
|
||||
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "unknown command: %s\n", os.Args[1])
|
||||
os.Exit(1)
|
||||
fmt.Fprintf(os.Stderr, "unknown command: %s\n", cmd)
|
||||
return sb.String(), 1
|
||||
}
|
||||
return sb.String(), 0
|
||||
}
|
||||
|
||||
func flagOrNil(s string) interface{} {
|
||||
if s == "" {
|
||||
return nil
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// mainFlags — the argparse surface (--source/--payload/--limit/--type/
|
||||
// --event/--data/--json + the create-envelope extras).
|
||||
type mainFlags struct {
|
||||
rest, source, payload, evType, event, data, channel, text string
|
||||
limit int
|
||||
jsonOut bool
|
||||
}
|
||||
|
||||
func parseFlags(args []string) mainFlags {
|
||||
f := mainFlags{limit: 20}
|
||||
if len(args) > 0 && !strings.HasPrefix(args[0], "--") &&
|
||||
(args[0] != "" || len(args) == 1) {
|
||||
f.rest = args[0]
|
||||
args = args[1:]
|
||||
}
|
||||
for i := 0; i < len(args); i++ {
|
||||
val := func() string {
|
||||
if i+1 < len(args) {
|
||||
i++
|
||||
return args[i]
|
||||
}
|
||||
return ""
|
||||
}
|
||||
switch args[i] {
|
||||
case "--source":
|
||||
f.source = val()
|
||||
case "--payload":
|
||||
f.payload = val()
|
||||
case "--limit":
|
||||
n, _ := strconv.Atoi(val())
|
||||
f.limit = n
|
||||
case "--type":
|
||||
f.evType = val()
|
||||
case "--event":
|
||||
f.event = val()
|
||||
case "--data":
|
||||
f.data = val()
|
||||
case "--channel":
|
||||
f.channel = val()
|
||||
case "--text":
|
||||
f.text = val()
|
||||
case "--json":
|
||||
f.jsonOut = true
|
||||
}
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// statsCounts — CLI stats helper: (total, by_type ordered).
|
||||
func statsCounts() (int, *ordObj) {
|
||||
rows, err := ReplayOrdered(StorePath())
|
||||
if err != nil {
|
||||
return 0, newOrdObj()
|
||||
}
|
||||
byType := newOrdObj()
|
||||
for _, r := range rows {
|
||||
t, _ := r.vals["event_type"].(string)
|
||||
if t == "" {
|
||||
t = "unknown"
|
||||
}
|
||||
n, _ := byType.vals[t].(json.Number)
|
||||
byType.set(t, json.Number(itoaE(int64Of(n)+1)))
|
||||
}
|
||||
return len(rows), byType
|
||||
}
|
||||
|
||||
func journalFileSize() int64 {
|
||||
fi, err := os.Stat(StorePath())
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return fi.Size()
|
||||
}
|
||||
|
||||
// ── op mode — the parity dispatch (envelope.go carries the port) ──────────
|
||||
|
||||
func opMode() string {
|
||||
dec := json.NewDecoder(os.Stdin)
|
||||
dec.UseNumber()
|
||||
v, err := decodeOrderedValue(dec)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus op: bad request: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
req, ok := v.(*ordObj)
|
||||
if !ok {
|
||||
fmt.Fprintf(os.Stderr, "event-bus op: request must be an object\n")
|
||||
os.Exit(1)
|
||||
}
|
||||
pyJSON, err := runOp(req)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus op: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
var semantic interface{}
|
||||
dec2 := json.NewDecoder(strings.NewReader(pyJSON))
|
||||
dec2.UseNumber()
|
||||
dec2.Decode(&semantic)
|
||||
out, _ := json.Marshal(map[string]interface{}{
|
||||
"json": pyJSON,
|
||||
"result": semantic,
|
||||
})
|
||||
fmt.Println(string(out))
|
||||
return ""
|
||||
}
|
||||
|
||||
// decodeOrderedValue / decodeOrderedFrom — token walk preserving object
|
||||
// key order (*ordObj) and number literals (json.Number via UseNumber).
|
||||
func decodeOrderedValue(dec *json.Decoder) (interface{}, error) {
|
||||
@@ -349,35 +376,9 @@ func optStrPtr(o *ordObj, key string) *string {
|
||||
return &s
|
||||
}
|
||||
|
||||
// legacyStateKeys — the ee_to_dict / ee_to_canonical_dict_unbridged state
|
||||
// dicts arrive in dataclass field order; values map onto the struct.
|
||||
func legacyFromState(state interface{}) *LegacyEventEnvelope {
|
||||
m := payloadAsMap(state)
|
||||
if m == nil {
|
||||
return &LegacyEventEnvelope{}
|
||||
}
|
||||
taskID := (*string)(nil)
|
||||
if v, ok := m["task_id"].(string); ok {
|
||||
taskID = &v
|
||||
}
|
||||
var canonical *EventEnvelope
|
||||
return &LegacyEventEnvelope{
|
||||
EventID: m["event_id"].(string),
|
||||
Timestamp: m["timestamp"].(string),
|
||||
Source: m["source"].(string),
|
||||
Channel: m["channel"].(string),
|
||||
AuthorityClass: m["authority_class"].(string),
|
||||
TaskID: taskID,
|
||||
Payload: m["payload"],
|
||||
Provenance: m["provenance"],
|
||||
Urgency: m["urgency"].(string),
|
||||
Canonical: canonical,
|
||||
}
|
||||
}
|
||||
|
||||
// runOp — one parity op; returns the PYTHON-json byte form (separators,
|
||||
// ensure_ascii, insertion order). The uuid counter resets per op,
|
||||
// mirroring the ee_freeze_seams rig.
|
||||
// mirroring the eb_freeze_seams seam reset law.
|
||||
func runOp(req *ordObj) (string, error) {
|
||||
resetEEUUID()
|
||||
switch opStr(req, "op") {
|
||||
@@ -425,6 +426,49 @@ func runOp(req *ordObj) (string, error) {
|
||||
return pyDumpsValue(orderedTable(eeChannelKeys, func(k string) interface{} { return eeChannels[k] })), nil
|
||||
case "authority_classes":
|
||||
return pyDumpsValue(orderedTable(eeAuthorityKeys, func(k string) interface{} { return eeAuthorityClasses[k] })), nil
|
||||
|
||||
// ── event_bus gap-close surface (store.go) ──────────────────────────
|
||||
case "store_path":
|
||||
return pyDumpsValue(storePathNormalized()), nil
|
||||
case "event_types":
|
||||
return pyDumpsValue(orderedTable(ebEventTypeKeys, func(k string) interface{} { return ebEventTypes[k] })), nil
|
||||
case "content_hash":
|
||||
payload, _ := opGet(req, "payload")
|
||||
return pyDumpsValue(ContentHashAny(payload)), nil
|
||||
case "emit":
|
||||
ebOpClearJournal()
|
||||
payload, _ := opGet(req, "payload")
|
||||
return emitOp(opStr(req, "event_type"), mustArg(req, "source"), payload)
|
||||
case "emit_line":
|
||||
ebOpClearJournal()
|
||||
payload, _ := opGet(req, "payload")
|
||||
return emitLineOp(opStr(req, "event_type"), mustArg(req, "source"), payload)
|
||||
case "log":
|
||||
ebOpClearJournal()
|
||||
seedJournalOp()
|
||||
limit, _ := opGet(req, "limit")
|
||||
evType, _ := opGet(req, "event_type")
|
||||
return logOp(limit, evType)
|
||||
case "stats":
|
||||
ebOpClearJournal()
|
||||
seedJournalOp()
|
||||
return statsOp()
|
||||
case "handle_hook":
|
||||
ebOpClearJournal()
|
||||
return handleHookOp(opStr(req, "event"), mustArg(req, "data"))
|
||||
case "main":
|
||||
ebOpClearJournal()
|
||||
if truthy, _ := req.vals["seed"].(bool); truthy {
|
||||
seedJournalOp()
|
||||
}
|
||||
return runMainOp(req)
|
||||
case "replay":
|
||||
// raw store query — NO clear, NO seed (the cross-writer probe)
|
||||
rows, err := ReplayOrdered(StorePath())
|
||||
if err != nil {
|
||||
return pyDumpsValue(ordKV("error", "RuntimeError", "message", err.Error())), nil
|
||||
}
|
||||
return pyDumpsValue(ordKV("rows", rowsOrNil(rows))), nil
|
||||
}
|
||||
return "", fmt.Errorf("unknown op: %q", opStr(req, "op"))
|
||||
}
|
||||
@@ -434,6 +478,161 @@ func mustArg(req *ordObj, key string) interface{} {
|
||||
return v
|
||||
}
|
||||
|
||||
// legacyStateKeys — the ee_to_dict / ee_to_canonical_dict_unbridged state
|
||||
// dicts arrive in dataclass field order; values map onto the struct.
|
||||
func legacyFromState(state interface{}) *LegacyEventEnvelope {
|
||||
m := payloadAsMap(state)
|
||||
if m == nil {
|
||||
return &LegacyEventEnvelope{}
|
||||
}
|
||||
taskID := (*string)(nil)
|
||||
if v, ok := m["task_id"].(string); ok {
|
||||
taskID = &v
|
||||
}
|
||||
var canonical *EventEnvelope
|
||||
return &LegacyEventEnvelope{
|
||||
EventID: m["event_id"].(string),
|
||||
Timestamp: m["timestamp"].(string),
|
||||
Source: m["source"].(string),
|
||||
Channel: m["channel"].(string),
|
||||
AuthorityClass: m["authority_class"].(string),
|
||||
TaskID: taskID,
|
||||
Payload: m["payload"],
|
||||
Provenance: m["provenance"],
|
||||
Urgency: m["urgency"].(string),
|
||||
Canonical: canonical,
|
||||
}
|
||||
}
|
||||
|
||||
// storePathNormalized — the freeze/parity normalization law: absolute
|
||||
// machine roots never appear in truth rows (EB_SANDBOX / EB_HOME /
|
||||
// REPO_ROOT). Order matters: repo/worktree roots first, then the
|
||||
// hardcoded rune home the unset-env default derives.
|
||||
func storePathNormalized() string {
|
||||
p := StorePath()
|
||||
if sandbox := os.Getenv("EB_SANDBOX_ROOT"); sandbox != "" {
|
||||
p = strings.ReplaceAll(p, sandbox, "EB_SANDBOX")
|
||||
}
|
||||
if home := os.Getenv("EB_HOME_ROOT"); home != "" {
|
||||
p = strings.ReplaceAll(p, home, "EB_HOME")
|
||||
}
|
||||
if root := os.Getenv("EB_REPO_ROOT"); root != "" {
|
||||
p = strings.ReplaceAll(p, root, "REPO_ROOT")
|
||||
}
|
||||
p = strings.ReplaceAll(p, "/mnt/workdrive/rune", "REPO_ROOT")
|
||||
return p
|
||||
}
|
||||
|
||||
// emitOp — the eb_emit seam form: {emitted, journal} (or the loud
|
||||
// ValueError shape {error, message, journal: []} — nothing written).
|
||||
func emitOp(eventType string, source interface{}, payload interface{}) (string, error) {
|
||||
env, err := Emit(eventType, source, payload)
|
||||
if err != nil {
|
||||
return pyDumpsValue(ordKV("error", "ValueError", "message", err.Error(),
|
||||
"journal", []interface{}{})), nil
|
||||
}
|
||||
rows, rerr := ReplayOrdered(StorePath())
|
||||
if rerr != nil {
|
||||
return "", rerr
|
||||
}
|
||||
return pyDumpsValue(ordKV("emitted", env, "journal", rowsOrNil(rows))), nil
|
||||
}
|
||||
|
||||
// emitLineOp — the eb_emit_line seam: emit then capture the journal FILE
|
||||
// bytes (the json.dumps(to_dict, sort_keys=True) line contract).
|
||||
func emitLineOp(eventType string, source interface{}, payload interface{}) (string, error) {
|
||||
if _, err := Emit(eventType, source, payload); err != nil {
|
||||
return pyDumpsValue(ordKV("error", "ValueError", "message", err.Error(),
|
||||
"line", "")), nil
|
||||
}
|
||||
b, rerr := os.ReadFile(StorePath())
|
||||
if rerr != nil {
|
||||
return "", rerr
|
||||
}
|
||||
return pyDumpsValue(string(b)), nil
|
||||
}
|
||||
|
||||
func logOp(limit, evType interface{}) (string, error) {
|
||||
rows, err := LogRows(limit, evType)
|
||||
if err != nil {
|
||||
return pyDumpsValue(ordKV("error", "RuntimeError", "message", err.Error())), nil
|
||||
}
|
||||
return pyDumpsValue(ordKV("rows", rowsOrNil(rows))), nil
|
||||
}
|
||||
|
||||
func statsOp() (string, error) {
|
||||
out, err := Stats()
|
||||
if err != nil {
|
||||
return pyDumpsValue(ordKV("error", "RuntimeError", "message", err.Error())), nil
|
||||
}
|
||||
return pyDumpsValue(out), nil
|
||||
}
|
||||
|
||||
func handleHookOp(event string, data interface{}) (string, error) {
|
||||
result := HandleHook(event, data)
|
||||
if result.vals["error"] != nil {
|
||||
return pyDumpsValue(result), nil
|
||||
}
|
||||
rows, rerr := ReplayOrdered(StorePath())
|
||||
if rerr != nil {
|
||||
return pyDumpsValue(ordKV("hook", result, "error", "RuntimeError", "message", rerr.Error())), nil
|
||||
}
|
||||
return pyDumpsValue(ordKV("hook", result, "journal", rowsOrNil(rows))), nil
|
||||
}
|
||||
|
||||
// runMainOp — {"op":"main","argv":[...]}: the captured-stdout CLI seam
|
||||
// (exit code + bytes), ids/clock already pinned for the op.
|
||||
func runMainOp(req *ordObj) (string, error) {
|
||||
argvV, _ := opGet(req, "argv")
|
||||
list, _ := argvV.([]interface{})
|
||||
argv := make([]string, len(list))
|
||||
for i, a := range list {
|
||||
argv[i], _ = a.(string)
|
||||
}
|
||||
stdout, code := runMain(argv)
|
||||
if code != 0 {
|
||||
// argparse-style failure — surface as exit + captured stderr-empty
|
||||
return pyDumpsValue(ordKV("stdout", stdout, "exit", json.Number(itoaE(int64(code))))), nil
|
||||
}
|
||||
return pyDumpsValue(ordKV("stdout", stdout, "exit", json.Number("0"))), nil
|
||||
}
|
||||
|
||||
func rowsOrNil(rows []*ordObj) interface{} {
|
||||
if rows == nil {
|
||||
return []interface{}{}
|
||||
}
|
||||
arr := make([]interface{}, len(rows))
|
||||
for i, r := range rows {
|
||||
arr[i] = r
|
||||
}
|
||||
return arr
|
||||
}
|
||||
|
||||
// ebOpClearJournal — the rig's per-case clear: every store-touching op
|
||||
// starts with an empty journal (the seam case is self-contained).
|
||||
func ebOpClearJournal() {
|
||||
_ = os.Remove(StorePath())
|
||||
}
|
||||
|
||||
// seedJournalOp — the rig's SEEDS through the real Emit.
|
||||
var opSeedSpecs = []struct {
|
||||
event, source string
|
||||
payload interface{}
|
||||
}{
|
||||
{"user.message", "user", ordKV("text", "hi")},
|
||||
{"tool.result", "", nil},
|
||||
{"user.message", "user", ordKV("n", json.Number("1"))},
|
||||
}
|
||||
|
||||
func seedJournalOp() {
|
||||
for _, s := range opSeedSpecs {
|
||||
if _, err := Emit(s.event, s.source, s.payload); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "event-bus op: seed emit: %v\n", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// eeChannelKeys / eeAuthorityKeys — the tables iterate in the ORACLE's
|
||||
// definition order (json.dumps insertion-order law).
|
||||
var eeChannelKeys = []string{"conversation", "work", "system", "tool", "background", "sensor"}
|
||||
@@ -455,9 +654,3 @@ func orderedTable(keys []string, get func(string) interface{}) *ordObj {
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
func init() {
|
||||
// batch 54: the AddEvent kernel is exercised via the existing truth battery
|
||||
// (truth_bus_test.go already covers ContentHashEvent byte-parity)
|
||||
_ = AddEvent // keep the compiler happy — used by the truth test
|
||||
}
|
||||
@@ -0,0 +1,737 @@
|
||||
package main
|
||||
|
||||
// store.go — event_bus.py gap close (dichotomy wave, contract factory-gap-eb):
|
||||
// the two audit-named gaps land here name-preserving — StorePath (_store_path)
|
||||
// and HandleHook (handle_hook) — plus the journal-store machinery they ride
|
||||
// (the canonical DurableEventStore write/replay shape the Python oracle
|
||||
// delegates to in rune.orchestration.event_store).
|
||||
//
|
||||
// Parity laws carried over from the oracle:
|
||||
// * EventEnvelope.to_dict is the dataclass asdict — schema_version FIRST
|
||||
// (base-class field order), then the envelope fields.
|
||||
// * The JSONL store line is json.dumps(to_dict, sort_keys=True) — keys
|
||||
// SORTED on every nesting level, ", "/": " separators, ensure_ascii.
|
||||
// * source must be a str (X3 discharge): non-str raises the loud
|
||||
// ValueError whose message carries the Python type name.
|
||||
// * str(v)[:200] slices by CODEPOINTS (the sleep_time quirk).
|
||||
// * the clock is always a seam (EE_FROZEN_NOW) and ids ride the
|
||||
// deterministic uuid counter reset per op (ee_freeze_seams law).
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ── _store_path — the canonical journal path ──────────────────────────────
|
||||
|
||||
// pyExpand — os.path.expandvars then expanduser (memory_env.expand).
|
||||
func pyExpand(p string) string {
|
||||
p = os.ExpandEnv(p)
|
||||
if strings.HasPrefix(p, "~") {
|
||||
home := os.Getenv("HOME")
|
||||
if home != "" {
|
||||
p = home + p[1:]
|
||||
}
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// ebRuneHome — memory_env.rune_home: RUNE_HOME or the hardcoded default.
|
||||
func ebRuneHome() string {
|
||||
if v, ok := os.LookupEnv("RUNE_HOME"); ok {
|
||||
if v == "" {
|
||||
return ""
|
||||
}
|
||||
return pyExpand(v)
|
||||
}
|
||||
return "/mnt/workdrive/rune"
|
||||
}
|
||||
|
||||
// ebMemoryDir — memory_env.memory_dir: RUNE_MEMORY_DIR (SET-but-empty wins —
|
||||
// Python os.environ.get returns "" and expand("") is "") else <rune_home>/memory.
|
||||
func ebMemoryDir() string {
|
||||
if v, ok := os.LookupEnv("RUNE_MEMORY_DIR"); ok {
|
||||
return pyExpand(v)
|
||||
}
|
||||
return pyExpand(filepath.Join(ebRuneHome(), "memory"))
|
||||
}
|
||||
|
||||
// StorePath — event_bus._store_path: the canonical journal file
|
||||
// (the migration keeps the SAME journal file as the legacy writer).
|
||||
func StorePath() string {
|
||||
return filepath.Join(ebMemoryDir(), "journal", "events.jsonl")
|
||||
}
|
||||
|
||||
// ── EVENT_TYPES — the typed event vocabulary (definition order) ───────────
|
||||
|
||||
var ebEventTypeKeys = []string{
|
||||
// Lifecycle
|
||||
"rune.started", "session.started", "session.ended", "rune.shutdown",
|
||||
// Conversation
|
||||
"user.message", "response.generated", "owner.correction",
|
||||
"commitment.created", "commitment.closed",
|
||||
// Memory
|
||||
"memory.created", "memory.recalled", "memory.conflict",
|
||||
"memory.superseded", "memory.abstraction_candidate",
|
||||
// Capability
|
||||
"capability.requested", "capability.selected", "capability.failed",
|
||||
"capability.unavailable", "plugin.timeout",
|
||||
// Task
|
||||
"task.created", "task.completed", "task.failed", "task.stale",
|
||||
// Growth
|
||||
"lesson.candidate", "lesson.promoted", "skill.candidate",
|
||||
"self_model.evidence",
|
||||
// System
|
||||
"model.unavailable", "storage.degraded", "context.pressure",
|
||||
// Hook lifecycle
|
||||
"hook.on_startup", "hook.on_message", "hook.on_response",
|
||||
"hook.on_session_end", "hook.on_schedule", "hook.on_memory_add",
|
||||
}
|
||||
|
||||
var ebEventTypes = map[string]string{
|
||||
"rune.started": "Rune session started",
|
||||
"session.started": "New conversation session began",
|
||||
"session.ended": "Conversation session ended",
|
||||
"rune.shutdown": "Rune shutting down",
|
||||
"user.message": "User sent a message",
|
||||
"response.generated": "Response generated",
|
||||
"owner.correction": "Owner corrected Rune",
|
||||
"commitment.created": "A promise/commitment was made",
|
||||
"commitment.closed": "A commitment was fulfilled or abandoned",
|
||||
"memory.created": "New memory entry added",
|
||||
"memory.recalled": "Memory retrieved for use",
|
||||
"memory.conflict": "Contradictory memories detected",
|
||||
"memory.superseded": "Memory entry superseded",
|
||||
"memory.abstraction_candidate": "Repeated pattern detected",
|
||||
"capability.requested": "Capability needed",
|
||||
"capability.selected": "Capability chosen for use",
|
||||
"capability.failed": "Capability execution failed",
|
||||
"capability.unavailable": "Needed capability not found",
|
||||
"plugin.timeout": "Plugin execution timed out",
|
||||
"task.created": "New task created",
|
||||
"task.completed": "Task finished",
|
||||
"task.failed": "Task failed",
|
||||
"task.stale": "Task hasn't been touched in a while",
|
||||
"lesson.candidate": "Potential lesson identified",
|
||||
"lesson.promoted": "Lesson promoted to durable knowledge",
|
||||
"skill.candidate": "Potential skill identified",
|
||||
"self_model.evidence": "Self-model evidence collected",
|
||||
"model.unavailable": "Model provider unavailable",
|
||||
"storage.degraded": "Storage system degraded",
|
||||
"context.pressure": "Context window under pressure",
|
||||
"hook.on_startup": "Startup hook fired",
|
||||
"hook.on_message": "Message hook fired",
|
||||
"hook.on_response": "Response hook fired",
|
||||
"hook.on_session_end": "Session end hook fired",
|
||||
"hook.on_schedule": "Scheduled hook fired",
|
||||
"hook.on_memory_add": "Memory add hook fired",
|
||||
}
|
||||
|
||||
// ── Python scalar semantics ───────────────────────────────────────────────
|
||||
|
||||
// pyTruthy — Python bool(v) for a decoded JSON value: {} and [] and "0"
|
||||
// (the INT literal) and 0.0 are falsy, "0" the string is truthy.
|
||||
func pyTruthy(v interface{}) bool {
|
||||
switch x := v.(type) {
|
||||
case nil:
|
||||
return false
|
||||
case bool:
|
||||
return x
|
||||
case string:
|
||||
return x != ""
|
||||
case json.Number:
|
||||
if isIntLiteral(string(x)) {
|
||||
n, err := strconv.ParseInt(string(x), 10, 64)
|
||||
if err != nil {
|
||||
return true // huge ints are truthy
|
||||
}
|
||||
return n != 0
|
||||
}
|
||||
f, err := strconv.ParseFloat(string(x), 64)
|
||||
if err != nil {
|
||||
return true
|
||||
}
|
||||
return f != 0
|
||||
case []interface{}:
|
||||
return len(x) > 0
|
||||
case *ordObj:
|
||||
return len(x.keys) > 0
|
||||
default:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
func isIntLiteral(s string) bool {
|
||||
if s == "" {
|
||||
return false
|
||||
}
|
||||
i := 0
|
||||
if s[0] == '-' {
|
||||
if len(s) == 1 {
|
||||
return false
|
||||
}
|
||||
i = 1
|
||||
}
|
||||
for ; i < len(s); i++ {
|
||||
c := s[i]
|
||||
if c < '0' || c > '9' {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// pyTypeName — type(x).__name__ for the ValueError message (and repr).
|
||||
func pyTypeName(v interface{}) string {
|
||||
switch x := v.(type) {
|
||||
case nil:
|
||||
return "NoneType"
|
||||
case bool:
|
||||
return "bool"
|
||||
case string:
|
||||
return "str"
|
||||
case json.Number:
|
||||
if isIntLiteral(string(x)) {
|
||||
return "int"
|
||||
}
|
||||
return "float"
|
||||
case []interface{}:
|
||||
return "list"
|
||||
case *ordObj:
|
||||
return "dict"
|
||||
default:
|
||||
return fmt.Sprintf("%T", x)
|
||||
}
|
||||
}
|
||||
|
||||
// truncateRunes — Python str[:n] slices CODEPOINTS, never bytes.
|
||||
func truncateRunes(s string, n int) string {
|
||||
runes := []rune(s)
|
||||
if len(runes) <= n {
|
||||
return s
|
||||
}
|
||||
return string(runes[:n])
|
||||
}
|
||||
|
||||
// pyReprStr — Python repr(str): single quotes, \\ \' \n \r \t escapes,
|
||||
// \xHH for ASCII controls, printable unicode kept verbatim.
|
||||
func pyReprStr(s string) string {
|
||||
var sb strings.Builder
|
||||
sb.WriteByte('\'')
|
||||
for _, r := range []rune(s) {
|
||||
switch r {
|
||||
case '\'':
|
||||
sb.WriteString("\\'")
|
||||
case '\\':
|
||||
sb.WriteString("\\\\")
|
||||
case '\n':
|
||||
sb.WriteString("\\n")
|
||||
case '\r':
|
||||
sb.WriteString("\\r")
|
||||
case '\t':
|
||||
sb.WriteString("\\t")
|
||||
default:
|
||||
if r < 0x20 || r == 0x7f {
|
||||
sb.WriteString(fmt.Sprintf("\\x%02x", r))
|
||||
} else {
|
||||
sb.WriteRune(r)
|
||||
}
|
||||
}
|
||||
}
|
||||
sb.WriteByte('\'')
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// pyStrValue — Python str(v) for a decoded JSON value: a bare str is its
|
||||
// raw self; EVERYTHING else is the repr form (containers don't define
|
||||
// __str__ in the oracle's data shapes).
|
||||
func pyStrValue(v interface{}) string {
|
||||
if s, ok := v.(string); ok {
|
||||
return s
|
||||
}
|
||||
return pyReprValue(v)
|
||||
}
|
||||
|
||||
// pyReprValue — Python repr(v): strings single-quoted + escaped, containers
|
||||
// rendered with repr'd elements and repr'd keys, insertion order preserved.
|
||||
func pyReprValue(v interface{}) string {
|
||||
switch x := v.(type) {
|
||||
case string:
|
||||
return pyReprStr(x)
|
||||
case bool:
|
||||
if x {
|
||||
return "True"
|
||||
}
|
||||
return "False"
|
||||
case nil:
|
||||
return "None"
|
||||
case json.Number:
|
||||
return pyNumberRepr(string(x))
|
||||
case []interface{}:
|
||||
parts := make([]string, len(x))
|
||||
for i, e := range x {
|
||||
parts[i] = pyReprValue(e)
|
||||
}
|
||||
return "[" + strings.Join(parts, ", ") + "]"
|
||||
case *ordObj:
|
||||
parts := make([]string, 0, len(x.keys))
|
||||
for _, k := range x.keys {
|
||||
parts = append(parts, pyReprValue(k)+": "+pyReprValue(x.vals[k]))
|
||||
}
|
||||
return "{" + strings.Join(parts, ", ") + "}"
|
||||
default:
|
||||
return fmt.Sprintf("%v", x)
|
||||
}
|
||||
}
|
||||
|
||||
// pyNumberRepr — repr of a Python int/float literal: ints stay as-is;
|
||||
// floats render Python-repr form ("1200.0" integral, "2.5", "1e-07").
|
||||
func pyNumberRepr(lit string) string {
|
||||
if isIntLiteral(lit) {
|
||||
return lit
|
||||
}
|
||||
f, err := strconv.ParseFloat(lit, 64)
|
||||
if err != nil {
|
||||
return lit
|
||||
}
|
||||
return pyFloatRepr(f)
|
||||
}
|
||||
|
||||
// pyFloatRepr — Python repr(float): integral values <1e16 keep ".0"
|
||||
// ("1200.0"), positional notation for decimal exponents in [-4, 16),
|
||||
// scientific with a 2-digit exponent beyond (Go's 'g' shortest matches
|
||||
// Python's shortest digits; encoding/json's 1-digit exponents do NOT),
|
||||
// and inf/NaN render the json.dumps names (Infinity / NaN).
|
||||
func pyFloatRepr(f float64) string {
|
||||
if math.IsNaN(f) {
|
||||
return "NaN"
|
||||
}
|
||||
if math.IsInf(f, 1) {
|
||||
return "Infinity"
|
||||
}
|
||||
if math.IsInf(f, -1) {
|
||||
return "-Infinity"
|
||||
}
|
||||
if math.Signbit(f) && f == 0 {
|
||||
return "-0.0"
|
||||
}
|
||||
if f == float64(int64(f)) && math.Abs(f) < 1e16 {
|
||||
return itoaE(int64(f)) + ".0"
|
||||
}
|
||||
exp := math.Floor(math.Log10(math.Abs(f)))
|
||||
if exp >= -4 && exp < 16 {
|
||||
s := strconv.FormatFloat(f, 'f', -1, 64)
|
||||
if !strings.Contains(s, ".") {
|
||||
s += ".0"
|
||||
}
|
||||
return s
|
||||
}
|
||||
return strconv.FormatFloat(f, 'g', -1, 64)
|
||||
}
|
||||
|
||||
// ── handle_hook — log hook events to the journal ──────────────────────────
|
||||
|
||||
// HandleHook — event_bus.handle_hook: "hook." prefix when the event is in
|
||||
// EVENT_TYPES, source "microkernel", payload shaping (str → json.loads with
|
||||
// the raw[:200] fallback; dict → str(v)[:200]; anything else falsy stays {}).
|
||||
// Returns {status, event, handler, logged}.
|
||||
func HandleHook(event string, data interface{}) *ordObj {
|
||||
eventType := event
|
||||
if _, ok := ebEventTypes["hook."+event]; ok {
|
||||
eventType = "hook." + event
|
||||
}
|
||||
source := "microkernel"
|
||||
var payload interface{} = newOrdObj()
|
||||
if pyTruthy(data) {
|
||||
if s, ok := data.(string); ok {
|
||||
parsed, err := pyLoads(s)
|
||||
if err != nil {
|
||||
payload = ordKV("raw", truncateRunes(s, 200))
|
||||
} else {
|
||||
payload = parsed
|
||||
}
|
||||
} else if d, ok := data.(*ordObj); ok {
|
||||
shaped := newOrdObj()
|
||||
for _, k := range d.keys {
|
||||
shaped.set(k, truncateRunes(pyStrValue(d.vals[k]), 200))
|
||||
}
|
||||
payload = shaped
|
||||
}
|
||||
// neither str nor dict → payload stays {}
|
||||
}
|
||||
emitted, err := Emit(eventType, source, payload)
|
||||
if err != nil {
|
||||
// emit's loud ValueError — propagate the message contract
|
||||
ev := ordKV("error", "ValueError", "message", err.Error())
|
||||
return ev
|
||||
}
|
||||
return ordKV("status", "ok",
|
||||
"event", event,
|
||||
"handler", "event_bus",
|
||||
"logged", opStr(emitted, "event_id"))
|
||||
}
|
||||
|
||||
// ── emit — the canonical append (X3 loud source contract) ─────────────────
|
||||
|
||||
// Emit — event_bus.emit: build the EventEnvelope (partition_key from source
|
||||
// or "system", content-addressed payload_hash), append + flush through the
|
||||
// DurableEventStore JSONL store, return the to_dict form. A non-str source
|
||||
// fails loud with the oracle's ValueError message.
|
||||
func Emit(eventType string, source interface{}, payload interface{}) (*ordObj, error) {
|
||||
if _, ok := source.(string); !ok {
|
||||
return nil, fmt.Errorf(
|
||||
"source must be a string, got '%s'; pass the payload via payload=...",
|
||||
pyTypeName(source))
|
||||
}
|
||||
partitionKey := source.(string)
|
||||
if partitionKey == "" {
|
||||
partitionKey = "system"
|
||||
}
|
||||
payloadHash := ""
|
||||
if pyTruthy(payload) {
|
||||
payloadHash = ContentHashAny(payload)
|
||||
}
|
||||
env := ordKV(
|
||||
"schema_version", 1,
|
||||
"event_type", eventType,
|
||||
"partition_key", partitionKey,
|
||||
"event_id", eeMintEventID(),
|
||||
"parent_id", nil,
|
||||
"actor_id", "",
|
||||
"device_id", "",
|
||||
"authorization", "",
|
||||
"payload_hash", payloadHash,
|
||||
"timestamp", eeNowISO(),
|
||||
)
|
||||
if err := appendStoreLine(StorePath(), pyDumpsSortedValue(env)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return env, nil
|
||||
}
|
||||
|
||||
// appendStoreLine — DurableEventStore.append+flush: mkdir parents, one
|
||||
// append line + "\n", fsync (durable right after emit).
|
||||
func appendStoreLine(path, line string) error {
|
||||
if dir := filepath.Dir(path); dir != "" {
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
if _, err := f.WriteString(line + "\n"); err != nil {
|
||||
return err
|
||||
}
|
||||
return f.Sync()
|
||||
}
|
||||
|
||||
// ── json.loads / repr of decoded payload (str-data path) ─────────────────
|
||||
|
||||
// pyLoads — Python json.loads over a str: insertion-ordered objects
|
||||
// (*ordObj), number literals as json.Number (int vs float preserved),
|
||||
// JSONDecodeError → error.
|
||||
func pyLoads(s string) (interface{}, error) {
|
||||
dec := json.NewDecoder(strings.NewReader(s))
|
||||
dec.UseNumber()
|
||||
v, err := decodeOrderedValue(dec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Python json.loads rejects trailing garbage after one document.
|
||||
if _, err := dec.Token(); err == nil {
|
||||
return nil, fmt.Errorf("Extra data")
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// ── replay / log / stats (canonical store queries) ────────────────────────
|
||||
|
||||
// ReplayOrdered — DurableEventStore.replay: 1-based offsets, blank lines
|
||||
// skipped, corrupt JSON raises (typed CorruptEventStoreError, message
|
||||
// "corrupt event at offset N: <python json error>").
|
||||
func ReplayOrdered(path string) ([]*ordObj, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
var rows []*ordObj
|
||||
offset := 0
|
||||
for _, raw := range strings.Split(string(data), "\n") {
|
||||
offset++
|
||||
line := strings.TrimSpace(raw)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
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)
|
||||
}
|
||||
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)
|
||||
}
|
||||
rows = append(rows, o)
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
// LogRows — event_bus.log: replay rows in append order, event_type filter,
|
||||
// then events[-limit:] truncation (limit 0 → ALL — the -0 slice quirk —
|
||||
// negative limit slices from the FRONT: [1:] for limit=-1).
|
||||
func LogRows(limit interface{}, eventType interface{}) ([]*ordObj, error) {
|
||||
rows, err := ReplayOrdered(StorePath())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if ev, ok := eventType.(string); ok {
|
||||
var filtered []*ordObj
|
||||
for _, r := range rows {
|
||||
if t, _ := r.vals["event_type"].(string); t == ev {
|
||||
filtered = append(filtered, r)
|
||||
}
|
||||
}
|
||||
rows = filtered
|
||||
}
|
||||
if limit == nil {
|
||||
return rows, nil
|
||||
}
|
||||
n, err := jsonNumberInt(limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
idx := -n
|
||||
if idx < 0 {
|
||||
idx += len(rows)
|
||||
}
|
||||
if idx < 0 {
|
||||
idx = 0
|
||||
}
|
||||
if idx > len(rows) {
|
||||
idx = len(rows)
|
||||
}
|
||||
return rows[idx:], nil
|
||||
}
|
||||
|
||||
func jsonNumberInt(v interface{}) (int, error) {
|
||||
switch x := v.(type) {
|
||||
case json.Number:
|
||||
return strconv.Atoi(string(x))
|
||||
case float64:
|
||||
return int(x), nil
|
||||
case int:
|
||||
return x, nil
|
||||
}
|
||||
return 0, fmt.Errorf("limit must be an integer, got %v", pyTypeName(v))
|
||||
}
|
||||
|
||||
// Stats — event_bus.stats: (total, by_type) with by_type in first-occurrence
|
||||
// (replay) order — json.dumps preserves it, Go maps would sort it away.
|
||||
func Stats() (*ordObj, error) {
|
||||
rows, err := ReplayOrdered(StorePath())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
byType := newOrdObj()
|
||||
for _, r := range rows {
|
||||
t, _ := r.vals["event_type"].(string)
|
||||
if t == "" {
|
||||
t = "unknown"
|
||||
}
|
||||
n, _ := byType.vals[t].(json.Number)
|
||||
byType.set(t, json.Number(itoaE(int64Of(n)+1)))
|
||||
}
|
||||
out := ordKV("total", json.Number(itoaE(int64(len(rows)))), "by_type", byType)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func int64Of(n json.Number) int64 {
|
||||
v, err := strconv.ParseInt(string(n), 10, 64)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// ── content hash over ANY payload shape ───────────────────────────────────
|
||||
|
||||
// ContentHashAny — event_bus._content_hash for a decoded payload of any
|
||||
// shape: json.dumps(payload, sort_keys=True) — ordObj keys SORTED at every
|
||||
// level, ", "/": " separators, ensure_ascii quoting.
|
||||
func ContentHashAny(payload interface{}) string {
|
||||
return "sha256:" + sha256Hex(renderSorted(payload))
|
||||
}
|
||||
|
||||
func sha256Hex(s string) string {
|
||||
sum := sha256.Sum256([]byte(s))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// toSortedValue — deep-convert ordObj → map with ordObj VALS converted too,
|
||||
// so writeEventJSON's sorted map rendering applies at every nesting level
|
||||
// (sort_keys=True is recursive) without disturbing json.Number literals.
|
||||
func toSortedValue(v interface{}) interface{} {
|
||||
switch x := v.(type) {
|
||||
case *ordObj:
|
||||
m := make(map[string]interface{}, len(x.keys))
|
||||
for _, k := range x.keys {
|
||||
m[k] = toSortedValue(x.vals[k])
|
||||
}
|
||||
return m
|
||||
case []interface{}:
|
||||
out := make([]interface{}, len(x))
|
||||
for i, e := range x {
|
||||
out[i] = toSortedValue(e)
|
||||
}
|
||||
return out
|
||||
default:
|
||||
return v
|
||||
}
|
||||
}
|
||||
|
||||
func renderSorted(v interface{}) string {
|
||||
return pyDumpsValue(toSortedValue(v))
|
||||
}
|
||||
|
||||
// pyDumpsSortedValue — json.dumps(v, sort_keys=True) byte form.
|
||||
func pyDumpsSortedValue(v interface{}) string {
|
||||
return renderSorted(v)
|
||||
}
|
||||
|
||||
// pyDumpsIndent — json.dumps(v, indent=2): item separator ",\n"+indent,
|
||||
// key separator ": ", EMPTY containers stay inline ({} / []).
|
||||
func pyDumpsIndent(v interface{}, depth int) string {
|
||||
var sb strings.Builder
|
||||
writeIndentJSON(&sb, v, depth)
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func writeIndentJSON(sb *strings.Builder, v interface{}, depth int) {
|
||||
inner := strings.Repeat(" ", 2*(depth+1))
|
||||
outer := strings.Repeat(" ", 2*depth)
|
||||
switch x := v.(type) {
|
||||
case nil, bool, string, json.Number, float64:
|
||||
writeEventJSON(sb, v)
|
||||
case *ordObj:
|
||||
if len(x.keys) == 0 {
|
||||
sb.WriteString("{}")
|
||||
return
|
||||
}
|
||||
sb.WriteString("{\n")
|
||||
for i, k := range x.keys {
|
||||
if i > 0 {
|
||||
sb.WriteString(",\n")
|
||||
}
|
||||
sb.WriteString(inner)
|
||||
sb.WriteString(pyQuoteEvent(k))
|
||||
sb.WriteString(": ")
|
||||
writeIndentJSON(sb, x.vals[k], depth+1)
|
||||
}
|
||||
sb.WriteString("\n")
|
||||
sb.WriteString(outer)
|
||||
sb.WriteString("}")
|
||||
case map[string]interface{}:
|
||||
keys := make([]string, 0, len(x))
|
||||
for k := range x {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
if len(keys) == 0 {
|
||||
sb.WriteString("{}")
|
||||
return
|
||||
}
|
||||
sb.WriteString("{\n")
|
||||
for i, k := range keys {
|
||||
if i > 0 {
|
||||
sb.WriteString(",\n")
|
||||
}
|
||||
sb.WriteString(inner)
|
||||
sb.WriteString(pyQuoteEvent(k))
|
||||
sb.WriteString(": ")
|
||||
writeIndentJSON(sb, x[k], depth+1)
|
||||
}
|
||||
sb.WriteString("\n")
|
||||
sb.WriteString(outer)
|
||||
sb.WriteString("}")
|
||||
case []interface{}:
|
||||
if len(x) == 0 {
|
||||
sb.WriteString("[]")
|
||||
return
|
||||
}
|
||||
sb.WriteString("[\n")
|
||||
for i, e := range x {
|
||||
if i > 0 {
|
||||
sb.WriteString(",\n")
|
||||
}
|
||||
sb.WriteString(inner)
|
||||
writeIndentJSON(sb, e, depth+1)
|
||||
}
|
||||
sb.WriteString("\n")
|
||||
sb.WriteString(outer)
|
||||
sb.WriteString("]")
|
||||
default:
|
||||
writeEventJSON(sb, v)
|
||||
}
|
||||
}
|
||||
|
||||
// ── stats CLI formatting (Python str.format parity) ───────────────────────
|
||||
|
||||
// commaInt — Python f"{n:,}" thousands separators.
|
||||
func commaInt(n int64) string {
|
||||
raw := itoaE(n)
|
||||
neg := strings.HasPrefix(raw, "-")
|
||||
if neg {
|
||||
raw = raw[1:]
|
||||
}
|
||||
var sb strings.Builder
|
||||
for i := 0; i < len(raw); i++ {
|
||||
if i > 0 && (len(raw)-i)%3 == 0 {
|
||||
sb.WriteByte(',')
|
||||
}
|
||||
sb.WriteByte(raw[i])
|
||||
}
|
||||
out := sb.String()
|
||||
if neg {
|
||||
out = "-" + out
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// stableTopTypes — sorted(types.items(), key=lambda x: -x[1])[:10]:
|
||||
// count desc, INSERTION order preserved among ties (Python sort stability).
|
||||
func stableTopTypes(byType *ordObj) []*ordObj {
|
||||
type kv struct {
|
||||
k string
|
||||
v int
|
||||
}
|
||||
items := make([]kv, 0, len(byType.keys))
|
||||
for _, k := range byType.keys {
|
||||
items = append(items, kv{k, int(int64Of(byType.vals[k].(json.Number)))})
|
||||
}
|
||||
sort.SliceStable(items, func(i, j int) bool { return items[i].v > items[j].v })
|
||||
if len(items) > 10 {
|
||||
items = items[:10]
|
||||
}
|
||||
out := make([]*ordObj, len(items))
|
||||
for i, it := range items {
|
||||
out[i] = ordKV("key", it.k, "count", json.Number(itoaE(int64(it.v))))
|
||||
}
|
||||
return out
|
||||
}
|
||||
+1523
-84
File diff suppressed because it is too large.
Load diff
@@ -1,36 +1,409 @@
|
||||
package main
|
||||
|
||||
// truth_bus_test.go — factory batch 9 parity gate.
|
||||
// truth_bus_test.go — the event_bus parity gate (dichotomy gap close,
|
||||
// factory-gap-eb). Replays every case from truth_bus.json (the shared
|
||||
// freeze_truth.py envelope: frozen clock + deterministic uuid counter +
|
||||
// hermetic journal sandbox, 61 Python-oracle cases WITH inputs, generated
|
||||
// by scripts/gen_event_bus_truth.py under scripts/eb_freeze_seams)
|
||||
// VERBATIM through the ported Go functions: StorePath (_store_path),
|
||||
// EVENT_TYPES, HandleHook (handle_hook: "hook."-vocabulary prefix, str→
|
||||
// json.loads with the raw[:200] fallback, dict→str(v)[:200] codepoint
|
||||
// truncation, falsy data stays {}), Emit (schema_version-first to_dict,
|
||||
// source-or-system partition, payload_hash truthiness, the loud ValueError
|
||||
// branches with NOTHING written), LogRows (filter + [-limit:] truncation
|
||||
// incl. the limit 0 → ALL quirk and the -1 → [1:] front-drop), Stats
|
||||
// (first-occurrence by_type), the journal LINE bytes, the typed
|
||||
// corrupt-store raise, and the captured CLI stdout. A divergence fails the
|
||||
// port, not the oracle.
|
||||
|
||||
import (
|
||||
_ "embed"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestBusParity(t *testing.T) {
|
||||
b, err := os.ReadFile("truth_bus.json")
|
||||
//go:embed truth_bus.json
|
||||
var frozenBusTruthJSON string
|
||||
|
||||
type busCase struct {
|
||||
Fn string `json:"fn"`
|
||||
Args []any `json:"args"`
|
||||
Kwargs map[string]any `json:"kwargs"`
|
||||
Expected json.RawMessage `json:"expected"`
|
||||
}
|
||||
|
||||
// runBusCase — dispatch one frozen case; returns the PYTHON-json byte form
|
||||
// of the actual (same renderer the op surface uses).
|
||||
func runBusCase(t *testing.T, c busCase) string {
|
||||
t.Helper()
|
||||
resetEEUUID()
|
||||
clearJournal(t)
|
||||
switch c.Fn {
|
||||
case "_content_hash":
|
||||
return pyDumpsValue(ContentHashAny(batteryPayload(t, c.Args[0])))
|
||||
case "eb_store_path":
|
||||
return pyDumpsValue(storePathNormalized())
|
||||
case "eb_event_types":
|
||||
return pyDumpsValue(orderedTable(ebEventTypeKeys, func(k string) interface{} { return ebEventTypes[k] }))
|
||||
case "eb_emit":
|
||||
var srcVal any = "" // oracle default source=""
|
||||
if raw, ok := c.Kwargs["source"]; ok {
|
||||
srcVal = batteryDecodeRaw(t, raw)
|
||||
}
|
||||
var payloadVal any // oracle default payload=None
|
||||
if raw, ok := c.Kwargs["payload"]; ok && raw != nil {
|
||||
payloadVal = batteryDecodeRaw(t, raw)
|
||||
}
|
||||
return pyDumpsValue(runEmitSeam(t,
|
||||
batteryArgStr(c, "event_type"), srcVal, payloadVal))
|
||||
case "eb_emit_line":
|
||||
_, err := Emit(batteryArgStr(c, "event_type"),
|
||||
kwargString(t, c, "source", ""),
|
||||
batteryKwargVal(t, c, "payload"))
|
||||
if err != nil {
|
||||
t.Fatalf("eb_emit_line: %v", err)
|
||||
}
|
||||
b, err := os.ReadFile(StorePath())
|
||||
if err != nil {
|
||||
t.Fatalf("read journal: %v", err)
|
||||
}
|
||||
return pyDumpsValue(string(b))
|
||||
case "eb_log":
|
||||
seedJournal(t)
|
||||
rows, err := LogRows(batteryKwargVal(t, c, "limit"),
|
||||
batteryKwargVal(t, c, "event_type"))
|
||||
if err == nil {
|
||||
return pyDumpsValue(ordKV("rows", rowsSlice(rows)))
|
||||
}
|
||||
return pyDumpsValue(ordKV("error", "RuntimeError", "message", err.Error()))
|
||||
case "eb_log_corrupt":
|
||||
if err := os.MkdirAll(filepath.Dir(StorePath()), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(StorePath(), []byte("not json\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err := LogRows(nil, nil)
|
||||
if err == nil {
|
||||
return pyDumpsValue(ordKV("error", nil))
|
||||
}
|
||||
return pyDumpsValue(ordKV("error", "RuntimeError", "message", err.Error()))
|
||||
case "eb_stats":
|
||||
seedJournal(t)
|
||||
out, err := Stats()
|
||||
if err != nil {
|
||||
t.Fatalf("eb_stats: %v", err)
|
||||
}
|
||||
return pyDumpsValue(out)
|
||||
case "eb_handle_hook":
|
||||
result := HandleHook(batteryArgStr(c, "event"),
|
||||
batteryKwargVal(t, c, "data"))
|
||||
if result.vals["error"] != nil {
|
||||
return pyDumpsValue(ordKV("hook", result))
|
||||
}
|
||||
rows, err := ReplayOrdered(StorePath())
|
||||
if err != nil {
|
||||
t.Fatalf("eb_handle_hook journal: %v", err)
|
||||
}
|
||||
return pyDumpsValue(ordKV("hook", result, "journal", rowsSlice(rows)))
|
||||
case "eb_main":
|
||||
argvV, _ := c.Args[0].([]interface{})
|
||||
argv := make([]string, len(argvV))
|
||||
for i, a := range argvV {
|
||||
argv[i], _ = a.(string)
|
||||
}
|
||||
if truthyKwarg(t, c, "seed") {
|
||||
for _, s := range seedSpecs {
|
||||
if _, err := Emit(s.event, s.source, batteryPayload(t, s.payload)); err != nil {
|
||||
t.Fatalf("seed emit: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
stdout, code := runMain(argv)
|
||||
return pyDumpsValue(ordKV("stdout", stdout, "exit", json.Number(itoaE(int64(code)))))
|
||||
}
|
||||
t.Fatalf("unknown frozen fn %q", c.Fn)
|
||||
return ""
|
||||
}
|
||||
|
||||
type seedSpec struct {
|
||||
event, source string
|
||||
payload any
|
||||
}
|
||||
|
||||
// seedSpecs — the rig's SEEDS (a/a/b shape: two user.message, tool.result).
|
||||
var seedSpecs = []seedSpec{
|
||||
{"user.message", "user", map[string]interface{}{"text": "hi"}},
|
||||
{"tool.result", "", nil},
|
||||
{"user.message", "user", map[string]interface{}{"n": json.Number("1")}},
|
||||
}
|
||||
|
||||
// runEmitSeam — the eb_emit seam: {emitted, journal} or the loud ValueError
|
||||
// shape {error, message, journal: []} (NOTHING written).
|
||||
func runEmitSeam(t *testing.T, eventType string, source, payload any) *ordObj {
|
||||
t.Helper()
|
||||
env, err := Emit(eventType, source, payload)
|
||||
if err != nil {
|
||||
t.Fatalf("truth_bus.json missing: %v", err)
|
||||
return ordKV("error", "ValueError", "message", err.Error(), "journal", []interface{}{})
|
||||
}
|
||||
dec := json.NewDecoder(strings.NewReader(string(b)))
|
||||
dec.UseNumber()
|
||||
var doc struct {
|
||||
Cases []struct {
|
||||
Name string `json:"name"`
|
||||
Input struct {
|
||||
Payload map[string]interface{} `json:"payload"`
|
||||
} `json:"input"`
|
||||
Expected struct {
|
||||
Hash string `json:"hash"`
|
||||
} `json:"expected"`
|
||||
} `json:"cases"`
|
||||
rows, rerr := ReplayOrdered(StorePath())
|
||||
if rerr != nil {
|
||||
t.Fatalf("replay: %v", rerr)
|
||||
}
|
||||
dec.Decode(&doc)
|
||||
for _, c := range doc.Cases {
|
||||
if got := ContentHashEvent(c.Input.Payload); got != c.Expected.Hash {
|
||||
t.Errorf("%s: hash\n got %q\nwant %q", c.Name, got, c.Expected.Hash)
|
||||
return ordKV("emitted", env, "journal", rowsSlice(rows))
|
||||
}
|
||||
|
||||
func rowsSlice(rows []*ordObj) []interface{} {
|
||||
out := make([]interface{}, len(rows))
|
||||
for i, r := range rows {
|
||||
out[i] = r
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// seedJournal — the rig's SEEDS (via the real Emit): a/a/b shape.
|
||||
func seedJournal(t *testing.T) {
|
||||
t.Helper()
|
||||
for _, s := range seedSpecs {
|
||||
if _, err := Emit(s.event, s.source, batteryAnyPayload(t, s.payload)); err != nil {
|
||||
t.Fatalf("seed emit: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func batteryAnyPayload(t *testing.T, v any) any {
|
||||
t.Helper()
|
||||
if v == nil {
|
||||
return nil
|
||||
}
|
||||
return batteryDecodeRaw(t, v)
|
||||
}
|
||||
|
||||
func clearJournal(t *testing.T) {
|
||||
t.Helper()
|
||||
_ = os.Remove(StorePath())
|
||||
}
|
||||
|
||||
func batteryArgStr(c busCase, what string) string {
|
||||
if len(c.Args) == 0 {
|
||||
return ""
|
||||
}
|
||||
s, _ := c.Args[0].(string)
|
||||
return s
|
||||
}
|
||||
|
||||
func batteryKwargVal(t *testing.T, c busCase, key string) any {
|
||||
t.Helper()
|
||||
v, ok := c.Kwargs[key]
|
||||
if !ok || v == nil {
|
||||
return nil
|
||||
}
|
||||
return batteryPayload(t, v)
|
||||
}
|
||||
|
||||
func kwargString(t *testing.T, c busCase, key, def string) string {
|
||||
v, ok := c.Kwargs[key]
|
||||
if !ok || v == nil {
|
||||
return def
|
||||
}
|
||||
s, ok := v.(string)
|
||||
if !ok {
|
||||
t.Fatalf("kwarg %s must be a string, got %v (%T)", key, v, v)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func truthyKwarg(t *testing.T, c busCase, key string) bool {
|
||||
v, ok := c.Kwargs[key]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
return pyTruthy(batteryDecodeRaw(t, v))
|
||||
}
|
||||
|
||||
func batteryDecodeRaw(t *testing.T, v any) any {
|
||||
t.Helper()
|
||||
b, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
t.Fatalf("decode kwarg: %v", err)
|
||||
}
|
||||
dec := json.NewDecoder(strings.NewReader(string(b)))
|
||||
dec.UseNumber()
|
||||
out, err := decodeOrderedValue(dec)
|
||||
if err != nil {
|
||||
t.Fatalf("ordered decode %s: %v", b, err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func canonDecBus(t *testing.T, s string) any {
|
||||
var out any
|
||||
dec := json.NewDecoder(strings.NewReader(s))
|
||||
dec.UseNumber()
|
||||
if err := dec.Decode(&out); err != nil {
|
||||
t.Fatalf("decode actual %q: %v", s, err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func canonRawBus(t *testing.T, raw json.RawMessage) any {
|
||||
t.Helper()
|
||||
var out any
|
||||
dec := json.NewDecoder(strings.NewReader(string(raw)))
|
||||
dec.UseNumber()
|
||||
if err := dec.Decode(&out); err != nil {
|
||||
t.Fatalf("decode expected %q: %v", raw, err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestFrozenBusTruthReplay(t *testing.T) {
|
||||
var doc struct {
|
||||
FrozenNow string `json:"frozen_now"`
|
||||
Cases []busCase `json:"cases"`
|
||||
}
|
||||
dec := json.NewDecoder(strings.NewReader(frozenBusTruthJSON))
|
||||
dec.UseNumber()
|
||||
if err := dec.Decode(&doc); err != nil {
|
||||
t.Fatalf("decode truth_bus.json: %v", err)
|
||||
}
|
||||
t.Setenv("EE_FROZEN_NOW", doc.FrozenNow)
|
||||
sandbox := t.TempDir()
|
||||
t.Setenv("RUNE_MEMORY_DIR", sandbox)
|
||||
t.Setenv("EB_SANDBOX_ROOT", sandbox)
|
||||
t.Setenv("EB_REPO_ROOT", "") // keep REPO_ROOT normalization off in tests
|
||||
|
||||
for i, c := range doc.Cases {
|
||||
got := runBusCase(t, c)
|
||||
want := canonRawBus(t, c.Expected)
|
||||
if !reflect.DeepEqual(canonDecBus(t, got), want) {
|
||||
t.Errorf("case %d fn=%s:\n got %v\nwant %v", i, c.Fn,
|
||||
canonDecBus(t, got), want)
|
||||
}
|
||||
}
|
||||
if len(doc.Cases) != 61 {
|
||||
t.Fatalf("expected 61 frozen cases, found %d", len(doc.Cases))
|
||||
}
|
||||
}
|
||||
|
||||
// TestStorePathDerivation — the env-derivation matrix the frozen corpus
|
||||
// pins through ONE import-time shape: RUNE_MEMORY_DIR (set-but-empty wins),
|
||||
// RUNE_HOME override, ~/ expansion, unset → the hardcoded rune home.
|
||||
func TestStorePathDerivation(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
env map[string]string
|
||||
unset []string
|
||||
want string
|
||||
}{
|
||||
{"memory_dir_set", map[string]string{"RUNE_MEMORY_DIR": "/sb/x"}, nil, "/sb/x/journal/events.jsonl"},
|
||||
{"memory_dir_empty", map[string]string{"RUNE_MEMORY_DIR": ""}, nil, "journal/events.jsonl"},
|
||||
{"rune_home", map[string]string{"RUNE_HOME": "/sb/home"}, []string{"RUNE_MEMORY_DIR"}, "/sb/home/memory/journal/events.jsonl"},
|
||||
{"tilde", map[string]string{"RUNE_MEMORY_DIR": "~/x", "HOME": "/sb/home"}, nil, "/sb/home/x/journal/events.jsonl"},
|
||||
{"default", nil, []string{"RUNE_MEMORY_DIR", "RUNE_HOME"}, "/mnt/workdrive/rune/memory/journal/events.jsonl"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
for _, k := range c.unset {
|
||||
os.Unsetenv(k)
|
||||
}
|
||||
for k, v := range c.env {
|
||||
t.Setenv(k, v)
|
||||
}
|
||||
defer func() { os.Unsetenv("RUNE_MEMORY_DIR"); os.Unsetenv("RUNE_HOME") }()
|
||||
if got := StorePath(); got != c.want {
|
||||
t.Errorf("StorePath() = %q, want %q", got, c.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestCommaInt — the f"{n:,}" thousands separator (stats CLI file_size).
|
||||
func TestCommaInt(t *testing.T) {
|
||||
cases := map[int64]string{0: "0", 5: "5", 999: "999", 1000: "1,000", 1234567: "1,234,567", -1200: "-1,200"}
|
||||
for n, want := range cases {
|
||||
if got := commaInt(n); got != want {
|
||||
t.Errorf("commaInt(%d) = %q, want %q", n, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestTruncateRunes — str[:200] slices CODEPOINTS, never bytes (the
|
||||
// sleep_time quirk; é counts as ONE character).
|
||||
func TestTruncateRunes(t *testing.T) {
|
||||
s := strings.Repeat("é", 210)[:0] + strings.Repeat("é", 210)
|
||||
got := truncateRunes(s, 200)
|
||||
if len([]rune(got)) != 200 {
|
||||
t.Fatalf("rune truncation broke: %d runes", len([]rune(got)))
|
||||
}
|
||||
if truncateRunes("hello", 3) != "hel" {
|
||||
t.Fatal("ascii truncation broke")
|
||||
}
|
||||
}
|
||||
|
||||
// TestPyStrValueRepr — the dict-payload str(v) representation contract:
|
||||
// a bare str at the TOP is its raw self, but container contents render in
|
||||
// repr form (single quotes) — str({'k': 'café'}) == "{'k': 'café'}".
|
||||
func TestPyStrValueRepr(t *testing.T) {
|
||||
cases := map[string]struct {
|
||||
in any
|
||||
want string
|
||||
}{
|
||||
"str": {"café", "'café'"},
|
||||
"bool": {true, "True"},
|
||||
"none": {nil, "None"},
|
||||
"int": {json.Number("42"), "42"},
|
||||
"float": {json.Number("2.5"), "2.5"},
|
||||
"float1": {json.Number("1200.0"), "1200.0"},
|
||||
"list": {[]any{json.Number("1"), "two", nil}, "[1, 'two', None]"},
|
||||
"dq": {"a\"b", "'a\"b'"},
|
||||
}
|
||||
for name, c := range cases {
|
||||
o := newOrdObj()
|
||||
o.set("k", c.in)
|
||||
got := pyStrValue(o)
|
||||
want := "{'k': " + c.want + "}"
|
||||
if got := pyStrValue(o); got != want {
|
||||
t.Errorf("%s: pyStrValue = %q, want %q", name, got, want)
|
||||
}
|
||||
_ = got
|
||||
}
|
||||
// a bare str value keeps its raw self (str("café") == "café")
|
||||
if got := pyStrValue("café"); got != "café" {
|
||||
t.Errorf("bare str: %q", got)
|
||||
}
|
||||
// newline/quote escaping in repr form
|
||||
o := newOrdObj()
|
||||
o.set("k", "a'b\nz")
|
||||
if got, want := pyStrValue(o), "{'k': 'a\\'b\\nz'}"; got != want {
|
||||
t.Errorf("repr escapes: %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLongValueTruncation — dict payload values slice at 200 codepoints
|
||||
// BEFORE hashing (the 250-x string pins the 200-char shape); the shaping
|
||||
// happens in HandleHook, not in the hash.
|
||||
func TestLongValueTruncation(t *testing.T) {
|
||||
t.Setenv("EE_FROZEN_NOW", "2026-10-14T11:22:33.123456+00:00")
|
||||
sandbox := t.TempDir()
|
||||
t.Setenv("RUNE_MEMORY_DIR", sandbox)
|
||||
t.Setenv("EB_SANDBOX_ROOT", sandbox)
|
||||
resetEEUUID()
|
||||
data := ordKV("long", strings.Repeat("x", 250))
|
||||
result := HandleHook("on_message", data)
|
||||
logged := opStr(result, "logged")
|
||||
rows, err := ReplayOrdered(StorePath())
|
||||
if err != nil || len(rows) != 1 {
|
||||
t.Fatalf("journal: %v %d", err, len(rows))
|
||||
}
|
||||
gotHash, _ := rows[0].vals["payload_hash"].(string)
|
||||
wantHash := ContentHashAny(ordKV("long", strings.Repeat("x", 200)))
|
||||
if gotHash != wantHash {
|
||||
t.Fatalf("hash %s != truncated-content hash %s (logged %s)",
|
||||
gotHash, wantHash, logged)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
exit=0
|
||||
@@ -0,0 +1 @@
|
||||
exit=0
|
||||
@@ -0,0 +1 @@
|
||||
exit=0
|
||||
@@ -0,0 +1 @@
|
||||
exit=0
|
||||
@@ -0,0 +1 @@
|
||||
exit=0
|
||||
@@ -0,0 +1,53 @@
|
||||
# event_bus gap-close verification summary (dichotomy wave #2479, contract factory-gap-eb)
|
||||
|
||||
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).
|
||||
|
||||
## Evidence index (cross-checked by Read before attestation)
|
||||
|
||||
| Gate | Command | Evidence (exit) |
|
||||
|---|---|---|
|
||||
| Truth freeze ×1 | PYTHONPATH=scripts/eb_freeze_seams:src python3 scripts/gen_event_bus_truth.py | (wrote 61 cases) verification/eb-audit.log — generator output also in-session |
|
||||
| freeze --verify pass 1 | PYTHONPATH=scripts/eb_freeze_seams:src python3 scripts/freeze_truth.py --verify src/go/event-bus/truth_bus.json --module rune.event_bus | verification/eb-freeze-verify-1.log — "OK: 61 cases byte-identical" (exit 0) |
|
||||
| freeze --verify pass 2 | same | verification/eb-freeze-verify-2.log — "OK: 61 cases byte-identical" (exit 0) |
|
||||
| go build ./... | (cd src/go/event-bus && go build ./...) | verification/eb-gobuild.log (exit 0, silent) |
|
||||
| go vet | (cd src/go/event-bus && go vet .) | verification/eb-govet.log (exit 0, silent) |
|
||||
| gofmt | gofmt -l . | verification/eb-gofmt.log (empty + exit 0) |
|
||||
| battery + race | go test -race . (replays truth_bus.json 61 cases VERBATIM + envelope 83 + units) | verification/eb-gotest.log — "ok rune/event-bus 1.095s" (exit 0) |
|
||||
| live parity | PYTHONPATH=scripts/eb_freeze_seams:src python3 scripts/check_event_bus_parity.py | verification/eb-parity.log — "parity: 73/73 match (negative control: detected)" (exit 0) |
|
||||
| Python suite live pin | PYTHONPATH=src uv-venv -m pytest tests/test_legacy_event_bus.py -q | verification/eb-pytest-live.log / .exit (4 passed, exit 0) |
|
||||
| Python suite pinned | PYTHONPATH=src RUNE_EVENT_BUS_CORE=python ... pytest | verification/eb-pytest-python-pin.log / .exit (4 passed, exit 0) |
|
||||
| audit regen | python3 scripts/dichotomy_audit.py --write | verification/eb-audit.log — FULL 55% (105/192); event_bus 7/7 FULL |
|
||||
|
||||
## Parity divergences fixed as DEFECTS (evidence: frozen rows + parity sweep)
|
||||
|
||||
* old-era emit path hashed payload with `fmt.Sprintf("%q: %v")` (Go repr, not
|
||||
json.dumps) — replaced by ContentHashAny (writeEventJSON sort_keys form).
|
||||
* old-era minted ids "evt-<wallclock-nanos>" and timestamps
|
||||
"...Z" — replaced with records.new_id shape "evt_<hex12>" via the
|
||||
deterministic counter seam and _now ISO "+00:00" clock seam.
|
||||
* old-era journalPath default $HOME/.rune — the oracle derives
|
||||
<RUNE_MEMORY_DIR|RUNE_HOME/memory>/journal/events.jsonl (memory_env chain
|
||||
with set-but-empty env + ~ / $VAR expansion).
|
||||
* replay() silently SKIPPED corrupt lines — the store raises the typed
|
||||
CorruptEventStoreError("corrupt event at offset N: ...") (message-only
|
||||
contract, pinned by the corrupt row).
|
||||
* CLI log timestamp used byte SplitN — now CODEPOINT [:19] (sleep_time quirk).
|
||||
* parent_id/emit to_dict omitted `parent_id: null` via omitempty tags — the
|
||||
to_dict render is field-ordered (schema_version first) with explicit null.
|
||||
|
||||
## Divergences documented (not bugs)
|
||||
|
||||
* Corrupt-line error suffix is Python's JSONDecodeError rendering
|
||||
("Expecting value: line 1 column 1 (char 0)") — the Go store raises the
|
||||
same message for the corpus-pinned leading-garbage shape; arbitrary
|
||||
malformed-JSON variants are not message-parity-covered (documented limit).
|
||||
* repr(float) for scientific notation pins 2-digit exponents + positional
|
||||
window [-4,16) (pyFloatRepr) — matching CPython repr(float) behavior.
|
||||
|
||||
## Commands not run / notes
|
||||
|
||||
* The 61 frozen cases + 73 live-parity pairs include the limit 0 → ALL and
|
||||
-N → front-drop slice quirks, raw[:200] codepoint truncation, str/repr
|
||||
value rendering, ensure_ascii hashing, and the captured CLI stdout.
|
||||
* SHAs captured AFTER the final successful push (rebase rewrites them).
|
||||
Reference in new issue
Block a user