feat(context_broker): close PARTIAL gaps — dichotomy wave
This commit is contained in:
Nick committed
1 parent
a829834f8d
commit
8119914fc3
6 files changed
+4368
-60
No files matched your search
@@ -0,0 +1,180 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Live parity check: the Go context-broker binary vs the LIVE Python oracle
|
||||
(RUNE_COGNITION_CORE=python pinned AND delegation auto), fuzz + edge cases.
|
||||
Complements the frozen battery (truth.json) with generated inputs."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import string
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "src"))
|
||||
|
||||
BIN = Path("/tmp/cb-live-parity")
|
||||
|
||||
|
||||
def build():
|
||||
subprocess.run(["go", "build", "-o", str(BIN), "."],
|
||||
cwd=ROOT / "src/go/context-broker", check=True)
|
||||
|
||||
|
||||
def go_eval(payload: dict) -> dict:
|
||||
proc = subprocess.run([str(BIN), "eval"], input=json.dumps(payload),
|
||||
capture_output=True, text=True)
|
||||
if proc.returncode != 0:
|
||||
raise RuntimeError(f"go eval failed: {proc.stderr}")
|
||||
return json.loads(proc.stdout)
|
||||
|
||||
|
||||
fails = 0
|
||||
checked = 0
|
||||
|
||||
|
||||
def check(name, got, want):
|
||||
global fails, checked
|
||||
checked += 1
|
||||
if got != want:
|
||||
fails += 1
|
||||
print(f"DIVERGE {name}: go={got!r} py={want!r}")
|
||||
|
||||
|
||||
def est_case(text):
|
||||
check(f"est[{text!r:.40}]", go_eval({"op": "est", "text": text}),
|
||||
py._py_default_estimator(text))
|
||||
|
||||
|
||||
def budget_case(w):
|
||||
check(f"budget[{w}]", go_eval({"op": "budget", "window": w}),
|
||||
py._py_default_budget(w))
|
||||
|
||||
|
||||
def membud_case(w):
|
||||
check(f"membud[{w}]", go_eval({"op": "membud", "window": w}),
|
||||
py._py_default_memory_budget(w))
|
||||
|
||||
|
||||
def memoff_case(v):
|
||||
check(f"memoff[{v!r}]", go_eval({"op": "memoff", "value": v}),
|
||||
py._py_memory_disabled_value(v))
|
||||
|
||||
|
||||
def through_case(messages, through):
|
||||
want = py._through_id(json.loads(json.dumps(messages)), through)
|
||||
got = go_eval({"op": "through", "messages": messages, "through": through})
|
||||
check(f"through[{through}/{len(messages)}]", got, want)
|
||||
|
||||
|
||||
rng = random.Random(2479)
|
||||
|
||||
os.environ["RUNE_COGNITION_CORE"] = "python"
|
||||
from rune import context_broker as py # noqa: E402
|
||||
|
||||
# fuzz: estimator
|
||||
for _ in range(200):
|
||||
n = rng.randint(0, 300)
|
||||
est_case("".join(rng.choice(string.printable[:94]) for _ in range(n)))
|
||||
est_case("héllo wörld ünïcode ✓ 日本語テキスト")
|
||||
est_case("🎉 emoji surrogate pair")
|
||||
|
||||
# fuzz: budgets (incl. boundaries around max(2048, w/4) crossover)
|
||||
for w in [0, 1, 2559, 2560, 2561, 8191, 8192, 8193, 10000, 65536, 131072,
|
||||
200000, 10**9] + [rng.randint(0, 200000) for _ in range(100)]:
|
||||
budget_case(w)
|
||||
membud_case(w)
|
||||
|
||||
# fuzz: memory-off raw values
|
||||
for v in ["", "0", "00", "false", "FALSE", "False", "no", "No", "off", "OFF",
|
||||
"off ", " off", "1", "true", "YES", "on", "anything", "\t", "\n",
|
||||
"0.0", "nil", "None", "null"]:
|
||||
memoff_case(v)
|
||||
|
||||
# fuzz: through_id over random message shapes
|
||||
for _ in range(100):
|
||||
msgs = []
|
||||
for _ in range(rng.randint(1, 8)):
|
||||
m = {"role": rng.choice(["user", "assistant", "system"]),
|
||||
"content": "".join(rng.choice(string.printable[:70]) for _ in range(rng.randint(0, 40)))}
|
||||
if rng.random() < 0.3:
|
||||
m["extra"] = {"n": rng.randint(0, 5), "flag": rng.random() < 0.5}
|
||||
msgs.append(m)
|
||||
through_case(msgs, rng.randint(-1, len(msgs) + 1))
|
||||
|
||||
# unicode through_id (ensure_ascii path)
|
||||
uni = [{"role": "user", "content": "日本語 テキスト 🎉 emoji"},
|
||||
{"role": "assistant", "content": "héllo wörld"}]
|
||||
through_case(uni, 1)
|
||||
through_case(uni, 0)
|
||||
|
||||
# fuzz: admission kernel (priority desc, recency tie, budget, memory cap)
|
||||
NAMES = ["system", "user", "history", "memory", "persona_memories", "tools", "notes"]
|
||||
for case_i in range(150):
|
||||
entries = []
|
||||
for j in range(rng.randint(1, 7)):
|
||||
name = rng.choice(NAMES)
|
||||
entries.append({
|
||||
"name": name,
|
||||
"priority": rng.randint(0, 100),
|
||||
"tokens": rng.randint(1, 900),
|
||||
"source": f"src-{j}",
|
||||
"evictable": name not in ("system", "user"),
|
||||
})
|
||||
budget = rng.choice([500, 1000, 2000, 4000])
|
||||
mem_budget = rng.choice([0, 300, 1200])
|
||||
got = go_eval({"op": "admit", "entries": entries, "budget": budget,
|
||||
"memory_budget": mem_budget})
|
||||
bk = py.ContextBroker()
|
||||
py_entries = [(e["name"], py.ContributionBlock(e["name"], e["priority"],
|
||||
"c" * e["tokens"], e["tokens"],
|
||||
e["source"], evictable=e["evictable"]), j)
|
||||
for j, e in enumerate(entries)]
|
||||
try:
|
||||
_adm, adm = bk._admit(py_entries, budget, memory_budget=mem_budget)
|
||||
want = {"admitted": [bl.name for bl in _adm],
|
||||
"blocks": adm["blocks"], "admitted_total": adm["admitted_total"],
|
||||
"dropped": [d["name"] for d in adm["dropped"]],
|
||||
"memory_budget": adm["memory_budget"], "memory_used": adm["memory_used"]}
|
||||
check(f"admit[{case_i}]", got, json.loads(json.dumps(want, sort_keys=True)))
|
||||
except py.ContextOverflowError as exc:
|
||||
# Go wraps the oracle's message with the wrapped sentinel; the oracle
|
||||
# text must be a verbatim prefix of the Go detail.
|
||||
ok = (isinstance(got, dict) and got.get("error") == "ContextOverflowError"
|
||||
and got.get("detail", "").startswith(str(exc)))
|
||||
check(f"admit[{case_i}]", "ContextOverflowError" if ok else got,
|
||||
"ContextOverflowError")
|
||||
|
||||
# fuzz: render (system folding, history pass-through, user capture)
|
||||
for case_i in range(100):
|
||||
blocks = []
|
||||
for j in range(rng.randint(1, 6)):
|
||||
name = rng.choice(NAMES)
|
||||
if rng.random() < 0.5:
|
||||
content = "".join(rng.choice(string.ascii_letters) for _ in range(rng.randint(1, 30)))
|
||||
else:
|
||||
content = [{"role": rng.choice(["user", "assistant"]),
|
||||
"content": "".join(rng.choice(string.ascii_letters) for _ in range(rng.randint(1, 20)))}
|
||||
for _ in range(rng.randint(1, 3))]
|
||||
blocks.append({"name": name, "priority": rng.randint(0, 100),
|
||||
"tokens": 0, "source": f"src-{j}", "content": content})
|
||||
got = go_eval({"op": "render", "blocks": blocks})
|
||||
py_blocks = []
|
||||
for b in blocks:
|
||||
content = b["content"]
|
||||
if isinstance(content, list):
|
||||
content = [dict(m) for m in content]
|
||||
py_blocks.append(py.ContributionBlock(b["name"], b["priority"], content,
|
||||
b["tokens"], b["source"]))
|
||||
caps = py.ContextBroker()._render(py_blocks)
|
||||
want = {"token_count": caps.token_count, "messages": caps.messages,
|
||||
"system_preamble": caps.system_preamble, "user_content": caps.user_content,
|
||||
"blocks": [bl.name for bl in caps.blocks]}
|
||||
check(f"render[{case_i}]", got, json.loads(json.dumps(want)))
|
||||
|
||||
print(f"live parity: {checked - fails}/{checked} cases match "
|
||||
f"(RUNE_COGNITION_CORE=python pinned)")
|
||||
sys.exit(1 if fails else 0)
|
||||
@@ -0,0 +1,383 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Freeze the oracle truth for the GO gap-close port of
|
||||
src/rune/context_broker.py (dichotomy wave #2479, GO, factory-gap-ctxbroker).
|
||||
|
||||
Every NEW function ported by the gap-close, every branch, against the LIVE
|
||||
oracle pinned to the Python cores (RUNE_COGNITION_CORE=python — the _py_*
|
||||
oracles ARE the parity floor being frozen). Deep-copied inputs: the oracle
|
||||
mutates message lists in place (compact_history archive path), so each row
|
||||
re-freezes the input verbatim.
|
||||
|
||||
Ops frozen (fn = the Go-side dispatcher key):
|
||||
_py_default_estimator est (text)
|
||||
_cc_core cc_core (env RUNE_COGNITION_CORE) -> bool available
|
||||
_py_default_budget budget (window)
|
||||
_py_default_memory_budget membud (window, incl. None/str/bool coercion)
|
||||
_py_memory_disabled_value memoff (raw string)
|
||||
reset_discovered_windows / register_window / default_window
|
||||
windows (sequence of ops -> final registry view)
|
||||
summarize_history summarize (transport-failure branch: ValueError)
|
||||
compact_history compact (messages, window, summarizer, session)
|
||||
rolling_summary_state rstate (key -> state | null)
|
||||
_through_id through (messages, through_count)
|
||||
store_rolling_summary / evict_rolling rstore (op sequence -> state + return)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "src"))
|
||||
|
||||
os.environ["RUNE_COGNITION_CORE"] = "python" # pin the Python parity floor
|
||||
|
||||
from rune import context_broker as cb # noqa: E402
|
||||
|
||||
rows: list[dict] = []
|
||||
|
||||
|
||||
def row(name, fn, inp, out):
|
||||
rows.append({"name": name, "fn": fn, "input": inp, "want": out})
|
||||
|
||||
|
||||
# ── _py_default_estimator: every branch ────────────────────────────────────
|
||||
row("est:empty", "est", {"text": ""}, cb._py_default_estimator(""))
|
||||
row("est:none-coerced-empty", "est", {"text": None}, cb._py_default_estimator(None))
|
||||
row("est:short", "est", {"text": "hello"}, cb._py_default_estimator("hello"))
|
||||
row("est:exact-8", "est", {"text": "12345678"}, cb._py_default_estimator("12345678"))
|
||||
row("est:9-floor", "est", {"text": "123456789"}, cb._py_default_estimator("123456789"))
|
||||
row("est:unicode-rune-len", "est", {"text": "héllo wörld"}, cb._py_default_estimator("héllo wörld"))
|
||||
|
||||
# ── _cc_core: python-pin + auto-without-build ───────────────────────────────
|
||||
from rune import cc_loader # noqa: E402
|
||||
_LOADER_AVAIL = bool(cc_loader.available())
|
||||
row("cc_core:python-pin-returns-none", "cc_core",
|
||||
{"env": {"RUNE_COGNITION_CORE": "python"}, "loader_available": _LOADER_AVAIL},
|
||||
None if cb._cc_core() is None else "CORE")
|
||||
del os.environ["RUNE_COGNITION_CORE"]
|
||||
row("cc_core:auto-fallback", "cc_core", {"env": {}, "loader_available": _LOADER_AVAIL},
|
||||
None if cb._cc_core() is None else "CORE")
|
||||
|
||||
# ── _py_default_budget: every branch ────────────────────────────────────────
|
||||
row("budget:0-floor", "budget", {"window": 0}, cb._py_default_budget(0))
|
||||
row("budget:1-floor", "budget", {"window": 1}, cb._py_default_budget(1))
|
||||
row("budget:100-floor", "budget", {"window": 100}, cb._py_default_budget(100))
|
||||
row("budget:4096", "budget", {"window": 4096}, cb._py_default_budget(4096))
|
||||
row("budget:8192", "budget", {"window": 8192}, cb._py_default_budget(8192))
|
||||
row("budget:131072", "budget", {"window": 131072}, cb._py_default_budget(131072))
|
||||
row("budget:quarter-exact-8192", "budget", {"window": 8192}, cb._py_default_budget(8192))
|
||||
|
||||
# ── _py_default_memory_budget: every branch incl. coercion ─────────────────
|
||||
row("membud:0-cap", "membud", {"window": 0}, cb._py_default_memory_budget(0))
|
||||
row("membud:none-cap", "membud", {"window": None}, cb._py_default_memory_budget(None))
|
||||
row("membud:neg-cap", "membud", {"window": -5}, cb._py_default_memory_budget(-5))
|
||||
row("membud:8-eq-1", "membud", {"window": 8}, cb._py_default_memory_budget(8))
|
||||
row("membud:2400", "membud", {"window": 2400}, cb._py_default_memory_budget(2400))
|
||||
row("membud:9600-cap", "membud", {"window": 9600}, cb._py_default_memory_budget(9600))
|
||||
row("membud:96000-cap", "membud", {"window": 96000}, cb._py_default_memory_budget(96000))
|
||||
|
||||
# bool/float coerce through the public fn's int(window) path when a Rust core
|
||||
# is present; on the Python floor they hit _py_default_memory_budget raw.
|
||||
row("membud:bool-true", "membud", {"window": True}, cb._py_default_memory_budget(True))
|
||||
row("membud:float", "membud", {"window": 2400.9}, cb._py_default_memory_budget(2400.9))
|
||||
|
||||
# ── _py_memory_disabled_value: every branch ────────────────────────────────
|
||||
for v in ("", "0", "false", "no", "off", "1", "true", "yes", "on", "OFF",
|
||||
" False ", "anything", " ", "off\n"):
|
||||
row(f"memoff:{v!r}", "memoff", {"value": v}, cb._py_memory_disabled_value(v))
|
||||
|
||||
# ── window registry: reset/register/default with NO live discovery ─────────
|
||||
# default_window's live-discovery import is exercised in the fallback branch
|
||||
# only when model_id is falsy (registry miss + import of local_tool_runtime
|
||||
# would attempt a real HTTP call otherwise — the Go seam stubs discovery;
|
||||
# here we freeze: falsy id, registry hit, registry miss w/ capped discovery).
|
||||
cb.reset_discovered_windows()
|
||||
row("windows:reset-then-missing-none", "windows",
|
||||
{"ops": [{"op": "default_window", "model_id": None}]},
|
||||
{"last": cb.default_window(None)})
|
||||
row("windows:none-empty-str", "windows",
|
||||
{"ops": [{"op": "default_window", "model_id": ""}]},
|
||||
{"last": cb.default_window("")})
|
||||
row("windows:register-hit", "windows",
|
||||
{"ops": [{"op": "register_window", "model_id": "m1", "window": 32768},
|
||||
{"op": "default_window", "model_id": "m1"}]},
|
||||
{"last": (cb.reset_discovered_windows(), cb.register_window("m1", 32768),
|
||||
cb.default_window("m1"))[2]})
|
||||
row("windows:register-overwrite", "windows",
|
||||
{"ops": [{"op": "register_window", "model_id": "m1", "window": 8192},
|
||||
{"op": "register_window", "model_id": "m1", "window": 65536},
|
||||
{"op": "default_window", "model_id": "m1"}]},
|
||||
{"last": (cb.reset_discovered_windows(), cb.register_window("m1", 8192),
|
||||
cb.register_window("m1", 65536), cb.default_window("m1"))[3]})
|
||||
row("windows:reset-clears", "windows",
|
||||
{"ops": [{"op": "register_window", "model_id": "m1", "window": 8192},
|
||||
{"op": "reset_discovered_windows"},
|
||||
{"op": "default_window", "model_id": "m1"}]},
|
||||
{"last": (cb.reset_discovered_windows(), cb.register_window("m1", 8192),
|
||||
cb.reset_discovered_windows(), cb.default_window("m1"))[3]})
|
||||
row("windows:register-str-window-coerced", "windows",
|
||||
{"ops": [{"op": "register_window", "model_id": "m1", "window": "28672"},
|
||||
{"op": "default_window", "model_id": "m1"}]},
|
||||
{"last": (cb.reset_discovered_windows(), cb.register_window("m1", "28672"),
|
||||
cb.default_window("m1"))[2]})
|
||||
# discovery-failure branch: unknown model -> the live oracle tries the
|
||||
# transport; in this offline env that raises inside context_length ->
|
||||
# record_failure + fallback 8192 (never cached). Freeze the returned value.
|
||||
cb.reset_discovered_windows()
|
||||
row("windows:miss-discover-fallback-8192", "windows",
|
||||
{"ops": [{"op": "default_window", "model_id": "definitely-not-a-model:offline"}]},
|
||||
{"last": cb.default_window("definitely-not-a-model:offline"),
|
||||
"note": "live oracle: T1 context_discovery + FALLBACK_WINDOW, not cached"})
|
||||
|
||||
# ── summarize_history: transport-failure branch (offline) ──────────────────
|
||||
try:
|
||||
cb.summarize_history([{"role": "user", "content": "hi"}], model="offline-no-such-model")
|
||||
row("summarize:unexpected-success", "summarize", {"messages": []}, "UNEXPECTED-SUCCESS")
|
||||
except ValueError as exc:
|
||||
row("summarize:transport-fail-valueerror", "summarize",
|
||||
{"messages": [{"role": "user", "content": "hi"}]},
|
||||
{"error": "ValueError", "detail": str(exc)[:120]})
|
||||
except Exception as exc: # noqa: BLE001 — freeze whatever the live env does
|
||||
row("summarize:transport-fail-other", "summarize",
|
||||
{"messages": [{"role": "user", "content": "hi"}]},
|
||||
{"error": type(exc).__name__, "detail": str(exc)[:120]})
|
||||
|
||||
# ── _through_id: every branch ───────────────────────────────────────────────
|
||||
msgs = [{"role": "user", "content": "first"},
|
||||
{"role": "assistant", "content": "reply"},
|
||||
{"role": "user", "content": "second"}]
|
||||
row("through:empty-list", "through", {"messages": [], "through": 0}, cb._through_id([], 0))
|
||||
row("through:neg", "through", {"messages": msgs, "through": -1}, cb._through_id(msgs, -1))
|
||||
row("through:at-end", "through", {"messages": msgs, "through": 3}, cb._through_id(msgs, 3))
|
||||
row("through:beyond-end", "through", {"messages": msgs, "through": 99}, cb._through_id(msgs, 99))
|
||||
row("through:anchor-0", "through", {"messages": msgs, "through": 0}, cb._through_id(msgs, 0))
|
||||
row("through:anchor-1", "through", {"messages": msgs, "through": 1}, cb._through_id(msgs, 1))
|
||||
row("through:content-hash-stable", "through",
|
||||
{"messages": msgs, "through": 2},
|
||||
(cb._through_id(msgs, 2), cb._through_id(copy.deepcopy(msgs), 2),
|
||||
cb._through_id([{"content": "second", "role": "user"}], 0))[0],
|
||||
)
|
||||
# key-order independence (sort_keys=True)
|
||||
row("through:key-order-independent", "through",
|
||||
{"messages": [{"role": "user", "content": "x"}, {"content": "x", "role": "user"}],
|
||||
"through": 1},
|
||||
cb._through_id([{"role": "user", "content": "x"}, {"content": "x", "role": "user"}], 1))
|
||||
|
||||
# ── rolling summary store: state/store/evict (self-contained rows) ──────────
|
||||
cb._ROLLING.clear()
|
||||
row("rstate:miss", "rstate", {"ops": [{"op": "read", "source": "operator", "sid": "s1"}]},
|
||||
cb.rolling_summary_state(("operator", "s1")))
|
||||
cb.store_rolling_summary(("operator", "s1"), "SUM", 7, "abcdef0123456789")
|
||||
row("rstate:hit-copy", "rstate",
|
||||
{"ops": [{"op": "store", "source": "operator", "sid": "s1", "summary": "SUM",
|
||||
"through": 7, "tid": "abcdef0123456789"},
|
||||
{"op": "read", "source": "operator", "sid": "s1"}]},
|
||||
cb.rolling_summary_state(("operator", "s1")))
|
||||
row("rstore:store-then-through-id", "rstore",
|
||||
{"msgs": msgs},
|
||||
(cb.store_rolling_summary(("operator", "s2"), "S2", 1, cb._through_id(msgs, 1)),
|
||||
cb.rolling_summary_state(("operator", "s2")))[1])
|
||||
|
||||
# ── compact_history: every branch (injected summarizer = deterministic) ────
|
||||
def fresh(seed: str, n_pairs: int) -> list[dict]:
|
||||
out = []
|
||||
for i in range(n_pairs):
|
||||
out.append({"role": "user", "content": f"{seed} user {i} " + "x" * (i * 40)})
|
||||
out.append({"role": "assistant", "content": f"{seed} assistant {i} " + "y" * (i * 40)})
|
||||
return out
|
||||
|
||||
|
||||
def sumr(messages):
|
||||
return "SUMMARY<" + "|".join(m.get("content", "")[:10] for m in messages) + ">"
|
||||
|
||||
|
||||
def compact_row(name, messages, window, *, summarizer="sum", session="", prior=None,
|
||||
archive=False):
|
||||
"""Freeze one compact_history call; self-contained (prior state inline)."""
|
||||
cb._ROLLING.clear()
|
||||
if prior:
|
||||
cb.store_rolling_summary(("operator", session) if session else ("operator",),
|
||||
prior["summary"], prior["through_count"],
|
||||
prior.get("through_id", ""))
|
||||
kw = dict(summarizer=sumr if summarizer != "boom" else boom)
|
||||
arch = None
|
||||
if archive:
|
||||
# deterministic RELATIVE fixture path (chdir'd to repo root): the
|
||||
# label is embedded in the summary block, so it must be portable
|
||||
os.chdir(ROOT)
|
||||
arch = Path("src/go/context-broker/.fixtures/archive.jsonl")
|
||||
arch.parent.mkdir(parents=True, exist_ok=True)
|
||||
inp = {"messages": copy.deepcopy(messages), "window": window,
|
||||
"summarizer": summarizer}
|
||||
if session:
|
||||
inp["session"] = session
|
||||
if prior:
|
||||
inp["prior"] = prior
|
||||
if archive:
|
||||
inp["archive"] = "FIXTURE"
|
||||
inp["archive_path"] = str(arch)
|
||||
compacted, ev = cb.compact_history(copy.deepcopy(messages), window,
|
||||
archive_path=arch,
|
||||
summarizer=sumr if summarizer != "boom" else boom,
|
||||
session_id=session)
|
||||
out = {"event": ev}
|
||||
if messages is not None:
|
||||
out["compacted"] = compacted
|
||||
if archive:
|
||||
body = arch.read_text() if arch.exists() else ""
|
||||
out["archive_lines"] = body.count("\n")
|
||||
out["archived_msg_count"] = len(json.loads(body.splitlines()[0])["messages"]) if body else 0
|
||||
out["archived_msgs"] = json.loads(body.splitlines()[0])["messages"] if body else []
|
||||
arch.unlink(missing_ok=True)
|
||||
row(name, "compact", inp, out)
|
||||
|
||||
|
||||
def boom(messages):
|
||||
raise RuntimeError("summarizer down")
|
||||
|
||||
|
||||
# no messages -> early return, event untriggered
|
||||
compact_row("compact:empty-messages", [], 8192)
|
||||
|
||||
# <= KEEP_TURNS user turns -> early return
|
||||
compact_row("compact:at-keep-turns-no-trigger", fresh("a", 5), 8192)
|
||||
compact_row("compact:below-keep-turns-no-trigger", fresh("a", 4), 8192)
|
||||
|
||||
# > keep turns, no session, summarizer success
|
||||
compact_row("compact:trigger-no-session", fresh("b", 7), 8192)
|
||||
|
||||
# with session: rolling persistence across TWO compactions (C1: only new)
|
||||
cb._ROLLING.clear()
|
||||
m3 = fresh("c", 8)
|
||||
c1, ev1 = cb.compact_history(copy.deepcopy(m3), 8192, summarizer=sumr,
|
||||
session_id="s3")
|
||||
prior3 = cb.rolling_summary_state(("operator", "s3"))
|
||||
row("compact:session-first", "compact",
|
||||
{"messages": m3, "window": 8192, "session": "s3", "summarizer": "sum"},
|
||||
{"event": ev1, "compacted": c1})
|
||||
# second compaction: window rotated under the stored count (#141 rematch)
|
||||
m3b = m3[:2] + m3[-10:]
|
||||
compact_row("compact:session-rotation-rematch", m3b, 8192, session="s3",
|
||||
prior=prior3)
|
||||
# prior through beyond len(older) — history shrank, re-summarize from scratch
|
||||
compact_row("compact:session-through-beyond-shrink", fresh("d", 7), 8192,
|
||||
session="s4", prior={"summary": "PRIOR-SUM", "through_count": 999,
|
||||
"through_id": ""})
|
||||
# nothing new beyond the rolled summary: reuse it verbatim
|
||||
cb._ROLLING.clear()
|
||||
m9 = fresh("g", 7)
|
||||
_, ev9_pre = cb.compact_history(copy.deepcopy(m9), 8192, summarizer=sumr,
|
||||
session_id="s9")
|
||||
prior9 = cb.rolling_summary_state(("operator", "s9"))
|
||||
# call again with IDENTICAL messages: older[6:] is empty -> reuse verbatim
|
||||
compact_row("compact:session-nothing-new-reuse", m9, 8192, session="s9",
|
||||
prior=prior9)
|
||||
|
||||
# summarizer failure -> C2 degrade to tail
|
||||
compact_row("compact:summarizer-fail-degrade-tail", fresh("e", 7), 8192,
|
||||
summarizer="boom")
|
||||
|
||||
# summarizer failure with tiny window (tail budget floor: always keep >=1)
|
||||
m6 = [{"role": "user", "content": "u" * 9000}, {"role": "assistant", "content": "a" * 9000},
|
||||
{"role": "user", "content": "u2" * 9000}, {"role": "assistant", "content": "a2" * 9000},
|
||||
{"role": "user", "content": "u3" * 9000}, {"role": "assistant", "content": "a3" * 9000},
|
||||
{"role": "user", "content": "u4" * 9000}]
|
||||
compact_row("compact:fail-tiny-window-tail-first-only", m6, 10, summarizer="boom")
|
||||
|
||||
# archive write: success path — event.archived True, label = the path itself
|
||||
compact_row("compact:archive-success", fresh("f", 6), 8192, session="s7",
|
||||
archive=True)
|
||||
|
||||
# default_estimator used inside event tokens (python pin): tokens_before math
|
||||
compact_row("compact:token-math-estimator",
|
||||
[{"role": "user", "content": "12345678"}, {"role": "assistant", "content": "1234"}],
|
||||
8192)
|
||||
|
||||
# ── ContextBroker kernel: to_dict / collect / admit / render ────────────────
|
||||
from rune.context_broker import ContributionBlock, ContextBroker, ContextOverflowError # noqa: E402
|
||||
|
||||
# to_dict
|
||||
blk = ContributionBlock("memory", 5, "hello world", 2, "recall", evictable=True)
|
||||
row("to_dict:block", "to_dict", {}, blk.to_dict())
|
||||
blk2 = ContributionBlock("system", 100, "sys", 0, "persona", evictable=False)
|
||||
row("to_dict:mandatory", "to_dict", {}, blk2.to_dict())
|
||||
|
||||
# collect: happy, empty-content skip, failure → T1 degrade, malformed → T1
|
||||
def make_broker():
|
||||
b = ContextBroker()
|
||||
b.register_contributor("ok", lambda task, session: ContributionBlock("tools", 9, "tooltext", 2, "t"), 1)
|
||||
b.register_contributor("empty", lambda task, session: ContributionBlock("x", 1, "", 0, "s"), 1)
|
||||
b.register_contributor("boom", lambda task, session: (_ for _ in ()).throw(RuntimeError("cb down")), 1)
|
||||
b.register_contributor("malformed", lambda task, session: ContributionBlock("y", 1, 12345, 0, "s"), 1)
|
||||
b.register_contributor("list-ok", lambda task, session: ContributionBlock("history", 3, [{"role": "user", "content": "hi"}], 1, "h"), 2)
|
||||
b.register_contributor("none", lambda task, session: None, 1)
|
||||
return b
|
||||
|
||||
|
||||
b = make_broker()
|
||||
degraded: list[str] = []
|
||||
collected = b._collect({}, {}, degraded)
|
||||
row("collect:degraded-and-ordered", "collect", {},
|
||||
{"names": [name for name, _blk, _b in collected],
|
||||
"degraded": degraded,
|
||||
"failed_events": sorted(getattr(b, "_last_failures", []))})
|
||||
|
||||
# admit: priority desc, recency tie, memory sub-budget, overflow
|
||||
def blocks(*specs):
|
||||
out = []
|
||||
for name, pri, tokens, evictable in specs:
|
||||
out.append(ContributionBlock(name, pri, "c" * tokens, tokens, f"src-{name}", evictable=evictable))
|
||||
return out
|
||||
|
||||
|
||||
bk = ContextBroker()
|
||||
admitted, adm = bk._admit([(b["name"], bl, 0) for b, bl in zip(
|
||||
[{"name": "system"}, {"name": "memory"}, {"name": "history"}, {"name": "user"}],
|
||||
blocks(("system", 100, 500, False), ("memory", 5, 300, True),
|
||||
("history", 7, 900, True), ("user", 100, 100, False)))], 1000)
|
||||
row("admit:priority-recency-budget", "admit", {},
|
||||
{"admitted": [bl.name for bl in admitted],
|
||||
"blocks": adm["blocks"], "admitted_total": adm["admitted_total"],
|
||||
"dropped": [d["name"] for d in adm["dropped"]],
|
||||
"memory_budget": adm["memory_budget"], "memory_used": adm["memory_used"]})
|
||||
|
||||
# memory sub-budget: two memory blocks, second exceeds sub-budget
|
||||
admitted2, adm2 = bk._admit([("m1", blocks(("m1", 9, 700, True))[0], 0),
|
||||
("m2", blocks(("m2", 8, 700, True))[0], 1)], 4000, memory_budget=1200)
|
||||
row("admit:memory-subbudget-drop", "admit", {},
|
||||
{"admitted": [bl.name for bl in admitted2],
|
||||
"dropped": [d["name"] for d in adm2["dropped"]],
|
||||
"memory_used": adm2["memory_used"],
|
||||
"reasons": [d["reason"] for d in adm2["dropped"]]})
|
||||
|
||||
# mandatory overflow → raise
|
||||
try:
|
||||
bk._admit([("user", blocks(("user", 100, 5000, False))[0], 0)], 1000)
|
||||
row("admit:mandatory-overflow-raises", "admit", {}, "NO-RAISE")
|
||||
except ContextOverflowError as exc:
|
||||
row("admit:mandatory-overflow-raises", "admit", {},
|
||||
{"error": "ContextOverflowError", "detail": str(exc)})
|
||||
|
||||
# render: system folding, history pass-through, user capture
|
||||
capsule = bk._render(blocks(("persona", 100, 7, False), ("tools", 90, 5, False),
|
||||
("history", 70, 4, False), ("user", 100, 2, False)))
|
||||
row("render:order-and-fold", "render", {},
|
||||
{"token_count": capsule.token_count,
|
||||
"messages": capsule.messages,
|
||||
"system_preamble": capsule.system_preamble,
|
||||
"user_content": capsule.user_content,
|
||||
"blocks": [bl.name for bl in capsule.blocks]})
|
||||
|
||||
print(f"froze {len(rows)} rows total")
|
||||
|
||||
doc = {
|
||||
"frozen_ts": "2026-10-12T00:00:00Z",
|
||||
"generator": "scripts/gen_context_broker_truth.py — live oracle src/rune/context_broker.py, RUNE_COGNITION_CORE=python pinned",
|
||||
"rows": rows,
|
||||
}
|
||||
out = ROOT / "src/go/context-broker/truth.json"
|
||||
out.write_text(json.dumps(doc, indent=1, sort_keys=True) + "\n")
|
||||
@@ -0,0 +1,420 @@
|
||||
package main
|
||||
|
||||
// broker.go — the ContextBroker orchestration core (admission, collection,
|
||||
// rendering) from src/rune/context_broker.py, gap-close dichotomy wave.
|
||||
// The Python class stays the runtime authority; this is the deterministic
|
||||
// kernel with the contributor callbacks and failure recorder injectable.
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// MemoryBlockNames — blocks that count as "memory" for the absence proof.
|
||||
var MemoryBlockNames = map[string]bool{
|
||||
"memory": true, "persona_memories": true, "consolidated": true,
|
||||
}
|
||||
|
||||
// ErrContextOverflow — one block alone exceeds the input budget (fail loud).
|
||||
var ErrContextOverflow = errors.New("context overflow")
|
||||
|
||||
// ContributionBlock — a typed, measurable piece of context. Content is
|
||||
// either a string (system/user text) or []Message (history messages).
|
||||
type ContributionBlock struct {
|
||||
Name string
|
||||
Priority int
|
||||
Content interface{}
|
||||
Tokens int
|
||||
Source string
|
||||
Evictable bool
|
||||
}
|
||||
|
||||
// ToDict — serialize for pending_state embedding (P1-02 #25).
|
||||
func (b *ContributionBlock) ToDict() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
"name": b.Name,
|
||||
"priority": b.Priority,
|
||||
"content": b.Content,
|
||||
"tokens": b.Tokens,
|
||||
"source": b.Source,
|
||||
"evictable": b.Evictable,
|
||||
}
|
||||
}
|
||||
|
||||
// ContributorFunc — callback(task, session) → block or nil.
|
||||
type ContributorFunc func(task, session map[string]interface{}) *ContributionBlock
|
||||
|
||||
// FailureRecorder is the T1 record_failure seam (oracle: subsystem_health).
|
||||
type FailureRecorder func(name string, err error)
|
||||
|
||||
// Contributor — a named registry entry with its priority bucket.
|
||||
type Contributor struct {
|
||||
Name string
|
||||
Callback ContributorFunc
|
||||
Bucket int
|
||||
}
|
||||
|
||||
// ContextBroker — admission-controlled context assembler (Go kernel).
|
||||
type ContextBroker struct {
|
||||
mu sync.Mutex
|
||||
contributors []Contributor
|
||||
traces []map[string]interface{}
|
||||
Estimator func(string) int
|
||||
BudgetResolver func(int) int
|
||||
Fail FailureRecorder
|
||||
}
|
||||
|
||||
// NewContextBroker wires the defaults.
|
||||
func NewContextBroker() *ContextBroker {
|
||||
return &ContextBroker{
|
||||
Estimator: DefaultEstimator,
|
||||
BudgetResolver: DefaultBudget,
|
||||
Fail: func(string, error) {},
|
||||
}
|
||||
}
|
||||
|
||||
// RegisterContributor names a contributor.
|
||||
func (cb *ContextBroker) RegisterContributor(name string, callback ContributorFunc, bucket int) {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
cb.contributors = append(cb.contributors, Contributor{Name: name, Callback: callback, Bucket: bucket})
|
||||
}
|
||||
|
||||
// TraceLog returns a shallow copy of every assemble trace (thread-safe).
|
||||
func (cb *ContextBroker) TraceLog() []map[string]interface{} {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
out := make([]map[string]interface{}, len(cb.traces))
|
||||
copy(out, cb.traces)
|
||||
return out
|
||||
}
|
||||
|
||||
// Collect runs contributors in registration order; failures are T1 events
|
||||
// (one pass, lock held — side effects run exactly once per turn). Skips nil
|
||||
// blocks and empty string content; malformed content (neither string nor a
|
||||
// list of message dicts) is a T1 failure too (I5).
|
||||
func (cb *ContextBroker) Collect(task, session map[string]interface{}) (collected []struct {
|
||||
Name string
|
||||
Block *ContributionBlock
|
||||
Bucket int
|
||||
}, degraded []string) {
|
||||
cb.mu.Lock()
|
||||
contributors := make([]Contributor, len(cb.contributors))
|
||||
copy(contributors, cb.contributors)
|
||||
cb.mu.Unlock()
|
||||
|
||||
degradedSet := map[string]bool{}
|
||||
for _, c := range contributors {
|
||||
block, err := func() (block *ContributionBlock, err error) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
if e, ok := r.(error); ok {
|
||||
err = e
|
||||
} else {
|
||||
err = fmt.Errorf("%v", r)
|
||||
}
|
||||
}
|
||||
}()
|
||||
return c.Callback(task, session), nil
|
||||
}()
|
||||
if err != nil {
|
||||
cb.Fail(c.Name, err)
|
||||
if !degradedSet[c.Name] {
|
||||
degradedSet[c.Name] = true
|
||||
degraded = append(degraded, c.Name)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if block == nil {
|
||||
continue
|
||||
}
|
||||
if s, isStr := block.Content.(string); isStr && s == "" {
|
||||
continue
|
||||
}
|
||||
if !contentOK(block.Content) {
|
||||
err := fmt.Errorf("contributor %q returned malformed content: %T", c.Name, block.Content)
|
||||
cb.Fail(c.Name, err)
|
||||
if !degradedSet[c.Name] {
|
||||
degradedSet[c.Name] = true
|
||||
degraded = append(degraded, c.Name)
|
||||
}
|
||||
continue
|
||||
}
|
||||
collected = append(collected, struct {
|
||||
Name string
|
||||
Block *ContributionBlock
|
||||
Bucket int
|
||||
}{c.Name, block, c.Bucket})
|
||||
}
|
||||
return collected, degraded
|
||||
}
|
||||
|
||||
func contentOK(content interface{}) bool {
|
||||
if s, isStr := content.(string); isStr {
|
||||
return s != "" || true
|
||||
}
|
||||
list, isList := content.([]interface{})
|
||||
if isList {
|
||||
for _, item := range list {
|
||||
m, isMap := item.(map[string]interface{})
|
||||
if !isMap {
|
||||
return false
|
||||
}
|
||||
if _, has := m["content"]; !has {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
if _, isMsgs := content.([]Message); isMsgs {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// AdmissionEntry — one collected block paired with its registration name.
|
||||
type AdmissionEntry struct {
|
||||
Name string
|
||||
Block *ContributionBlock
|
||||
}
|
||||
|
||||
// TraceRecord — per-block admission trace row.
|
||||
type TraceRecord struct {
|
||||
Name string `json:"name"`
|
||||
Priority int `json:"priority"`
|
||||
Tokens int `json:"tokens"`
|
||||
Source string `json:"source"`
|
||||
Evictable bool `json:"evictable"`
|
||||
Admitted bool `json:"admitted"`
|
||||
Reason string `json:"reason"`
|
||||
}
|
||||
|
||||
// AdmissionResult — admitted blocks + the full trace dict.
|
||||
type AdmissionResult struct {
|
||||
Admitted []*ContributionBlock
|
||||
Blocks []TraceRecord
|
||||
AdmittedTotal int
|
||||
Dropped []TraceRecord
|
||||
MemoryBudget int
|
||||
MemoryUsed int
|
||||
}
|
||||
|
||||
// Admit — deterministic admission: priority desc, tie: registration recency.
|
||||
// Non-evictable blocks always admit first (overflow → ErrContextOverflow);
|
||||
// evictable blocks admit while the budget lasts; memory-scope blocks are
|
||||
// additionally bounded by the memory sub-budget (P1-03 #26).
|
||||
func Admit(entries []AdmissionEntry, budget int, memoryBudget int) (*AdmissionResult, error) {
|
||||
if memoryBudget < 0 {
|
||||
memoryBudget = DefaultMemoryBudget(0)
|
||||
}
|
||||
type keyed struct {
|
||||
idx int
|
||||
entry AdmissionEntry
|
||||
}
|
||||
items := make([]keyed, len(entries))
|
||||
for i, e := range entries {
|
||||
items[i] = keyed{i, e}
|
||||
}
|
||||
sort.SliceStable(items, func(a, b int) bool {
|
||||
ba, bb := items[a].entry.Block, items[b].entry.Block
|
||||
ka, kb := 0, 0
|
||||
if ba.Evictable {
|
||||
ka = 1
|
||||
}
|
||||
if bb.Evictable {
|
||||
kb = 1
|
||||
}
|
||||
if ka != kb {
|
||||
return ka < kb
|
||||
}
|
||||
if ba.Priority != bb.Priority {
|
||||
return ba.Priority > bb.Priority
|
||||
}
|
||||
return items[a].idx < items[b].idx
|
||||
})
|
||||
|
||||
res := &AdmissionResult{MemoryBudget: memoryBudget}
|
||||
used, memoryUsed := 0, 0
|
||||
for _, it := range items {
|
||||
block := it.entry.Block
|
||||
record := TraceRecord{
|
||||
Name: block.Name, Priority: block.Priority, Tokens: block.Tokens,
|
||||
Source: block.Source, Evictable: block.Evictable,
|
||||
Admitted: true,
|
||||
}
|
||||
if !block.Evictable {
|
||||
if block.Tokens > budget {
|
||||
return nil, fmt.Errorf("mandatory block '%s' needs %d tokens but the input budget is %d — refusing to silently truncate: %w",
|
||||
block.Name, block.Tokens, budget, ErrContextOverflow)
|
||||
}
|
||||
res.Admitted = append(res.Admitted, block)
|
||||
used += block.Tokens
|
||||
res.Blocks = append(res.Blocks, record)
|
||||
continue
|
||||
}
|
||||
inMemoryScope := MemoryBlockNames[block.Name]
|
||||
if inMemoryScope && used+block.Tokens <= budget && memoryUsed+block.Tokens > memoryBudget {
|
||||
record.Admitted = false
|
||||
record.Reason = fmt.Sprintf("memory budget: %d+%d>%d (memory sub-budget exhausted; lowest-priority memory dropped)",
|
||||
memoryUsed, block.Tokens, memoryBudget)
|
||||
res.Blocks = append(res.Blocks, record)
|
||||
continue
|
||||
}
|
||||
if used+block.Tokens <= budget {
|
||||
res.Admitted = append(res.Admitted, block)
|
||||
used += block.Tokens
|
||||
if inMemoryScope {
|
||||
memoryUsed += block.Tokens
|
||||
}
|
||||
res.Blocks = append(res.Blocks, record)
|
||||
} else {
|
||||
record.Admitted = false
|
||||
record.Reason = fmt.Sprintf("budget: %d+%d>%d (lowest-priority evictable dropped)",
|
||||
used, block.Tokens, budget)
|
||||
res.Blocks = append(res.Blocks, record)
|
||||
}
|
||||
}
|
||||
for _, record := range res.Blocks {
|
||||
if !record.Admitted {
|
||||
res.Dropped = append(res.Dropped, record)
|
||||
}
|
||||
}
|
||||
res.AdmittedTotal = used
|
||||
res.MemoryUsed = memoryUsed
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// Capsule — the rendered attention capsule.
|
||||
type Capsule struct {
|
||||
TokenCount int
|
||||
Blocks []*ContributionBlock
|
||||
Messages []map[string]interface{}
|
||||
SystemPreamble string
|
||||
UserContent string
|
||||
}
|
||||
|
||||
// Render — admitted blocks into the typed capsule (fixed message order:
|
||||
// system -> tools -> memory/notes -> history -> user via priority sort).
|
||||
func Render(admitted []*ContributionBlock) *Capsule {
|
||||
type keyed struct {
|
||||
idx int
|
||||
block *ContributionBlock
|
||||
}
|
||||
items := make([]keyed, len(admitted))
|
||||
for i, b := range admitted {
|
||||
items[i] = keyed{i, b}
|
||||
}
|
||||
sort.SliceStable(items, func(a, b int) bool {
|
||||
if items[a].block.Priority != items[b].block.Priority {
|
||||
return items[a].block.Priority > items[b].block.Priority
|
||||
}
|
||||
return items[a].idx < items[b].idx
|
||||
})
|
||||
|
||||
var systemParts []string
|
||||
var history []map[string]interface{}
|
||||
var userBlock *ContributionBlock
|
||||
var capsuleBlocks []*ContributionBlock
|
||||
totalTokens := 0
|
||||
|
||||
for _, it := range items {
|
||||
block := it.block
|
||||
totalTokens += block.Tokens
|
||||
capsuleBlocks = append(capsuleBlocks, block)
|
||||
if s, isStr := block.Content.(string); isStr {
|
||||
if block.Name == "user" {
|
||||
userBlock = block
|
||||
} else {
|
||||
systemParts = append(systemParts, s)
|
||||
}
|
||||
} else {
|
||||
switch list := block.Content.(type) {
|
||||
case []Message:
|
||||
for _, m := range list {
|
||||
history = append(history, deepCopyMap(m))
|
||||
}
|
||||
case []interface{}:
|
||||
for _, item := range list {
|
||||
if m, ok := item.(map[string]interface{}); ok {
|
||||
history = append(history, deepCopyMap(m))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
systemPrompt := ""
|
||||
for i, p := range systemParts {
|
||||
if i > 0 {
|
||||
systemPrompt += "\n"
|
||||
}
|
||||
systemPrompt += p
|
||||
}
|
||||
rendered := make([]map[string]interface{}, 0, len(history)+1)
|
||||
if systemPrompt != "" {
|
||||
rendered = append(rendered, map[string]interface{}{"role": "system", "content": systemPrompt})
|
||||
}
|
||||
rendered = append(rendered, history...)
|
||||
|
||||
userContent := ""
|
||||
if userBlock != nil {
|
||||
userContent, _ = userBlock.Content.(string)
|
||||
}
|
||||
return &Capsule{
|
||||
TokenCount: totalTokens,
|
||||
Blocks: capsuleBlocks,
|
||||
Messages: rendered,
|
||||
SystemPreamble: systemPrompt,
|
||||
UserContent: userContent,
|
||||
}
|
||||
}
|
||||
|
||||
func deepCopyMap(m map[string]interface{}) map[string]interface{} {
|
||||
out := make(map[string]interface{}, len(m))
|
||||
for k, v := range m {
|
||||
if vm, ok := v.(map[string]interface{}); ok {
|
||||
out[k] = deepCopyMap(vm)
|
||||
} else if vl, ok := v.([]interface{}); ok {
|
||||
out[k] = deepCopyList(vl)
|
||||
} else {
|
||||
out[k] = v
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func deepCopyList(l []interface{}) []interface{} {
|
||||
out := make([]interface{}, len(l))
|
||||
for i, v := range l {
|
||||
if vm, ok := v.(map[string]interface{}); ok {
|
||||
out[i] = deepCopyMap(vm)
|
||||
} else if vl, ok := v.([]interface{}); ok {
|
||||
out[i] = deepCopyList(vl)
|
||||
} else {
|
||||
out[i] = v
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func blockNames(blocks []*ContributionBlock) []string {
|
||||
out := make([]string, 0, len(blocks))
|
||||
for _, b := range blocks {
|
||||
out = append(out, b.Name)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// BuildMemorySelector — the memory-selection authority composition
|
||||
// (_build_memory_selector, P1-03 #26). Fail-closed: any construction error
|
||||
// returns nil and the memory contributor abstains — never a fallback
|
||||
// injection. The selector itself stays Python (rune.memory_selector).
|
||||
func BuildMemorySelector(ctor func() (interface{}, error)) interface{} {
|
||||
sel, err := ctor()
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return sel
|
||||
}
|
||||
+708
-13
@@ -1,20 +1,61 @@
|
||||
// Package main — context-broker: the GO port of src/rune/context_broker.py's
|
||||
// budget/estimator kernels (Oracle-Battery Parity Pattern, #2479 GO batch 38).
|
||||
// budget/estimator kernels (Oracle-Battery Parity Pattern, #2479 GO batch 38;
|
||||
// gap-close dichotomy wave — factory-gap-ctxbroker).
|
||||
// The ContextBroker itself stays Python (it's the orchestration authority);
|
||||
// the budget math is the deterministic core.
|
||||
// the budget math is the deterministic core. The gap-close ports the missing
|
||||
// functions name-preserving: the _py_* parity-floor oracles, the discovered-
|
||||
// window registry, the rolling-summary store, _through_id, log_calibration,
|
||||
// and the compact_history/summarize_history state machines with IO behind
|
||||
// injectable seams (Estimator/Summarizer/Discoverer funcs, Archiver writer).
|
||||
package main
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"os"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// defaultEstimator — chars/4 (code-point parity with Python len()).
|
||||
func defaultEstimator(text string) int { return len([]rune(text)) / 4 }
|
||||
// ── estimator / budget kernels ──────────────────────────────────────────────
|
||||
|
||||
// defaultBudget — window - max(2048, window/4) - window/20, floored at 1.
|
||||
func defaultBudget(window int) int {
|
||||
// DefaultEstimator — chars/4 (code-point parity with Python len()).
|
||||
func DefaultEstimator(text string) int { return len([]rune(text)) / 4 }
|
||||
|
||||
// PyDefaultEstimator — the Python parity-floor oracle (_py_default_estimator).
|
||||
// Python: len(text or "") // 4 — None coerces to "".
|
||||
func PyDefaultEstimator(text *string) int {
|
||||
if text == nil {
|
||||
return 0
|
||||
}
|
||||
return len([]rune(*text)) / 4
|
||||
}
|
||||
|
||||
// CcCore — the cognition-core availability seam (_cc_core). The Go port IS
|
||||
// the core, so the Python-pin question collapses: RUNE_COGNITION_CORE=python
|
||||
// pins the Python floor (the caller compares against it), anything else with
|
||||
// the Go core linked is available. Env read is injectable via CcCoreEnv.
|
||||
var CcCoreEnv = os.LookupEnv
|
||||
|
||||
// CcCore reports whether the native core would be used under the given env.
|
||||
// Mirrors the oracle's branch shape: pin -> None; loader probe failure -> None.
|
||||
func CcCore(lookup func(string) (string, bool), loaderAvailable bool) *bool {
|
||||
if v, ok := lookup("RUNE_COGNITION_CORE"); ok && v == "python" {
|
||||
return nil
|
||||
}
|
||||
if !loaderAvailable {
|
||||
return nil
|
||||
}
|
||||
avail := true
|
||||
return &avail
|
||||
}
|
||||
|
||||
// DefaultBudget — window - max(2048, window/4) - window/20, floored at 1.
|
||||
func DefaultBudget(window int) int {
|
||||
reserve := window / 4
|
||||
if reserve < 2048 {
|
||||
reserve = 2048
|
||||
@@ -26,10 +67,13 @@ func defaultBudget(window int) int {
|
||||
return r
|
||||
}
|
||||
|
||||
// PyDefaultBudget — the Python parity-floor oracle (_py_default_budget).
|
||||
func PyDefaultBudget(window int) int { return DefaultBudget(window) }
|
||||
|
||||
const MemoryBudgetCap = 1200
|
||||
|
||||
// defaultMemoryBudget — min(cap, window/8); window<=0 → CAP.
|
||||
func defaultMemoryBudget(window int) int {
|
||||
// DefaultMemoryBudget — min(cap, window/8); window<=0 → CAP.
|
||||
func DefaultMemoryBudget(window int) int {
|
||||
if window <= 0 {
|
||||
return MemoryBudgetCap
|
||||
}
|
||||
@@ -40,6 +84,536 @@ func defaultMemoryBudget(window int) int {
|
||||
return r
|
||||
}
|
||||
|
||||
// PyDefaultMemoryBudget — the Python parity-floor oracle
|
||||
// (_py_default_memory_budget). Python: `if not window or window <= 0`.
|
||||
func PyDefaultMemoryBudget(window int) int { return DefaultMemoryBudget(window) }
|
||||
|
||||
// PyMemoryDisabledValue — the Python parity-floor oracle over the RAW string
|
||||
// (_py_memory_disabled_value): truthy unless in the falsy set.
|
||||
func PyMemoryDisabledValue(value string) bool {
|
||||
switch value {
|
||||
case "", "0", "false", "no", "off":
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// MemoryDisabled — the full gate (memory_disabled): Python cross-check ANDed
|
||||
// with the native verdict when the core is live.
|
||||
func MemoryDisabled(raw string, coreDisabled bool) bool {
|
||||
gate := PyMemoryDisabledValue(raw)
|
||||
return coreDisabled && gate
|
||||
}
|
||||
|
||||
// ── discovered-window registry (reset_discovered_windows/register_window/
|
||||
// default_window) — process-local registry, live discovery injectable ────
|
||||
|
||||
const FallbackWindow = 8192
|
||||
|
||||
var (
|
||||
discoveredMu sync.Mutex
|
||||
discovered = map[string]int{}
|
||||
)
|
||||
|
||||
// DiscoverWindow is the injectable seam for live context-length discovery
|
||||
// (the oracle imports rune.local_tool_runtime.context_length). nil = no
|
||||
// discovery available (offline), which freezes the fallback branch.
|
||||
var DiscoverWindow func(modelID string) int
|
||||
|
||||
// ResetDiscoveredWindows clears the discovered-window registry (I3).
|
||||
func ResetDiscoveredWindows() {
|
||||
discoveredMu.Lock()
|
||||
defer discoveredMu.Unlock()
|
||||
discovered = map[string]int{}
|
||||
}
|
||||
|
||||
// RegisterWindow records a discovered context length for modelID (Task 2).
|
||||
func RegisterWindow(modelID string, window int) {
|
||||
discoveredMu.Lock()
|
||||
defer discoveredMu.Unlock()
|
||||
discovered[modelID] = window
|
||||
}
|
||||
|
||||
// DefaultWindow resolves the model window via the registry, then live
|
||||
// discovery, then the conservative fallback. The fallback (8192) never
|
||||
// enters the cache (C3: a transient discovery failure must not poison the
|
||||
// window for the process lifetime).
|
||||
func DefaultWindow(modelID string) int {
|
||||
discoveredMu.Lock()
|
||||
d := DiscoverWindow
|
||||
discoveredMu.Unlock()
|
||||
return resolveWindow(modelID, d)
|
||||
}
|
||||
|
||||
// resolveWindow — the discovery-injectable core of DefaultWindow; the
|
||||
// discover func is captured per call so concurrent resolvers never swap a
|
||||
// shared global.
|
||||
func resolveWindow(modelID string, discover func(string) int) int {
|
||||
if modelID == "" {
|
||||
return FallbackWindow
|
||||
}
|
||||
discoveredMu.Lock()
|
||||
w, ok := discovered[modelID]
|
||||
discoveredMu.Unlock()
|
||||
if ok {
|
||||
return w
|
||||
}
|
||||
if discover == nil {
|
||||
return FallbackWindow
|
||||
}
|
||||
window := discover(modelID)
|
||||
if window != 0 && window != FallbackWindow {
|
||||
RegisterWindow(modelID, window)
|
||||
}
|
||||
return window
|
||||
}
|
||||
|
||||
// ── rolling-summary store (C1/#141) ─────────────────────────────────────────
|
||||
|
||||
// RollingState — the persisted rolling summary for one session key.
|
||||
type RollingState struct {
|
||||
Summary string `json:"summary"`
|
||||
ThroughCount int `json:"through_count"`
|
||||
ThroughID string `json:"through_id"`
|
||||
}
|
||||
|
||||
var (
|
||||
rollingMu sync.Mutex
|
||||
rolling = map[string]*RollingState{} // key: source + "\x00" + session_id
|
||||
)
|
||||
|
||||
func rollingKey(source, sessionID string) string { return source + "\x00" + sessionID }
|
||||
|
||||
// RollingSummaryState returns the persisted rolling summary for a session
|
||||
// (deep-copied), or nil.
|
||||
func RollingSummaryState(source, sessionID string) *RollingState {
|
||||
rollingMu.Lock()
|
||||
defer rollingMu.Unlock()
|
||||
st, ok := rolling[rollingKey(source, sessionID)]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
cp := *st
|
||||
return &cp
|
||||
}
|
||||
|
||||
// StoreRollingSummary persists a rolling summary keyed by (source, sessionID).
|
||||
func StoreRollingSummary(source, sessionID, summaryText string, throughCount int, throughID string) {
|
||||
rollingMu.Lock()
|
||||
defer rollingMu.Unlock()
|
||||
rolling[rollingKey(source, sessionID)] = &RollingState{
|
||||
Summary: summaryText,
|
||||
ThroughCount: throughCount,
|
||||
ThroughID: throughID,
|
||||
}
|
||||
}
|
||||
|
||||
// EvictRolling — LRU (insertion-order) eviction for dead sessions; returns
|
||||
// the evicted count.
|
||||
func EvictRolling(maxSessions int) int {
|
||||
rollingMu.Lock()
|
||||
defer rollingMu.Unlock()
|
||||
if len(rolling) <= maxSessions {
|
||||
return 0
|
||||
}
|
||||
keys := make([]string, 0, len(rolling))
|
||||
for k := range rolling {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys) // deterministic order (Python relies on dict insertion order)
|
||||
drop := len(rolling) - maxSessions
|
||||
for _, k := range keys[:drop] {
|
||||
delete(rolling, k)
|
||||
}
|
||||
return drop
|
||||
}
|
||||
|
||||
// ThroughID — identity of the coverage boundary: first 16 hex chars of the
|
||||
// sha256 of the first message BEYOND coverage (content, not position —
|
||||
// rotation-proof, #141). Empty when the boundary is outside the list.
|
||||
func ThroughID(messages []map[string]interface{}, throughCount int) string {
|
||||
if throughCount < 0 || throughCount >= len(messages) {
|
||||
return ""
|
||||
}
|
||||
anchor := messages[throughCount]
|
||||
sum := sha256.Sum256([]byte(pyDumpsSorted(anchor)))
|
||||
return fmt.Sprintf("%x", sum[:])[:16]
|
||||
}
|
||||
|
||||
// pyDumpsSorted renders json.dumps(anchor, sort_keys=True) byte-for-byte:
|
||||
// separators ", " / ": ", ensure_ascii escaping, Python literal literals.
|
||||
func pyDumpsSorted(v interface{}) string {
|
||||
var b strings.Builder
|
||||
writePyJSON(&b, v)
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func writePyJSON(b *strings.Builder, v interface{}) {
|
||||
switch val := v.(type) {
|
||||
case nil:
|
||||
b.WriteString("null")
|
||||
case bool:
|
||||
if val {
|
||||
b.WriteString("true")
|
||||
} else {
|
||||
b.WriteString("false")
|
||||
}
|
||||
case float64:
|
||||
b.WriteString(pyFloat(val))
|
||||
case int:
|
||||
b.WriteString(fmt.Sprintf("%d", val))
|
||||
case string:
|
||||
writePyString(b, val)
|
||||
case []interface{}:
|
||||
b.WriteString("[")
|
||||
for i, item := range val {
|
||||
if i > 0 {
|
||||
b.WriteString(", ")
|
||||
}
|
||||
writePyJSON(b, item)
|
||||
}
|
||||
b.WriteString("]")
|
||||
case map[string]interface{}:
|
||||
keys := make([]string, 0, len(val))
|
||||
for k := range val {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
b.WriteString("{")
|
||||
for i, k := range keys {
|
||||
if i > 0 {
|
||||
b.WriteString(", ")
|
||||
}
|
||||
writePyString(b, k)
|
||||
b.WriteString(": ")
|
||||
writePyJSON(b, val[k])
|
||||
}
|
||||
b.WriteString("}")
|
||||
default:
|
||||
blob, _ := json.Marshal(val)
|
||||
b.Write(blob)
|
||||
}
|
||||
}
|
||||
|
||||
func pyFloat(f float64) string {
|
||||
if f == float64(int64(f)) && math.Abs(f) < 1e15 {
|
||||
// integral numbers via JSON can only originate as Python ints —
|
||||
// json.dumps renders int(3) as "3", never "3.0"
|
||||
return fmt.Sprintf("%d", int64(f))
|
||||
}
|
||||
return strconv.FormatFloat(f, 'g', -1, 64)
|
||||
}
|
||||
|
||||
func writePyString(b *strings.Builder, s string) {
|
||||
b.WriteByte('"')
|
||||
for _, r := range s {
|
||||
switch r {
|
||||
case '"':
|
||||
b.WriteString("\\\"")
|
||||
case '\\':
|
||||
b.WriteString("\\\\")
|
||||
case '\n':
|
||||
b.WriteString("\\n")
|
||||
case '\r':
|
||||
b.WriteString("\\r")
|
||||
case '\t':
|
||||
b.WriteString("\\t")
|
||||
case '\b':
|
||||
b.WriteString("\\b")
|
||||
case '\f':
|
||||
b.WriteString("\\f")
|
||||
default:
|
||||
if r < 0x20 || r > 0x7e {
|
||||
if r > 0xffff {
|
||||
r2 := r - 0x10000
|
||||
hi := 0xd800 + (r2 >> 10)
|
||||
lo := 0xdc00 + (r2 & 0x3ff)
|
||||
fmt.Fprintf(b, "\\u%04x\\u%04x", hi, lo)
|
||||
} else {
|
||||
fmt.Fprintf(b, "\\u%04x", r)
|
||||
}
|
||||
} else {
|
||||
b.WriteRune(r)
|
||||
}
|
||||
}
|
||||
}
|
||||
b.WriteByte('"')
|
||||
}
|
||||
|
||||
// LogCalibration — compare the broker estimate with the served
|
||||
// prompt_eval_count. LOG ONLY (drift never feeds admission). Returns the
|
||||
// signed drift; 0 when nothing was served.
|
||||
func LogCalibration(measuredTokens int, promptEvalCount *int, model string) int {
|
||||
if promptEvalCount == nil || *promptEvalCount == 0 {
|
||||
return 0
|
||||
}
|
||||
return *promptEvalCount - measuredTokens
|
||||
}
|
||||
|
||||
// ── compaction state machine (compact_history) ──────────────────────────────
|
||||
|
||||
// Compaction thresholds (mirrors the oracle constants).
|
||||
const (
|
||||
CompactionThreshold = 0.8
|
||||
CompactionKeepTurns = 5
|
||||
CompactionNumPredict = 256
|
||||
)
|
||||
|
||||
// Message — one conversation turn (loosely typed like the Python dict).
|
||||
type Message = map[string]interface{}
|
||||
|
||||
// Archiver is the injectable seam for the raw-history archive append
|
||||
// (oracle: <rune_home>/state/history_archive.jsonl). Returns an error to
|
||||
// exercise the T2 archive-failure branch.
|
||||
type Archiver func(record map[string]interface{}) error
|
||||
|
||||
// Summarizer is the injectable seam for the rolling-summary call
|
||||
// (oracle: summarize_history via OllamaChatTransport). An error is a T2
|
||||
// storage failure: compaction is skipped, the caller degrades.
|
||||
type Summarizer func(messages []Message) (string, error)
|
||||
|
||||
// CompactEvent — the compaction event dict returned alongside the compacted
|
||||
// messages.
|
||||
type CompactEvent struct {
|
||||
Triggered bool `json:"triggered"`
|
||||
TokensBefore int `json:"tokens_before"`
|
||||
TokensAfter int `json:"tokens_after"`
|
||||
TurnsSummarized int `json:"turns_summarized"`
|
||||
Archived bool `json:"archived"`
|
||||
Failed bool `json:"failed"`
|
||||
Source string `json:"source"`
|
||||
Persisted bool `json:"persisted"`
|
||||
DegradedToTail *int `json:"degraded_to_tail,omitempty"`
|
||||
Summary string `json:"summary,omitempty"`
|
||||
}
|
||||
|
||||
// CompactHistory — archive-then-compress state machine, logic 1:1 with the
|
||||
// oracle (rolling persistence, #141 identity rematch, C2 tail degrade).
|
||||
// source is fixed "operator" by the assemble path; sessionID "" disables
|
||||
// persistence. archivePath is the injected archive location (the oracle's
|
||||
// Path parameter): the label is embedded in the summary block; "" renders
|
||||
// the oracle's 'external storage' fallback.
|
||||
func CompactHistory(messages []Message, window int, archivePath string, archiver Archiver, summarizer Summarizer, source, sessionID string) ([]Message, *CompactEvent) {
|
||||
tokensBefore := 0
|
||||
for _, m := range messages {
|
||||
tokensBefore += DefaultEstimator(messageContent(m))
|
||||
}
|
||||
event := &CompactEvent{
|
||||
TokensBefore: tokensBefore,
|
||||
TokensAfter: tokensBefore,
|
||||
Source: source,
|
||||
}
|
||||
if len(messages) == 0 {
|
||||
return messages, event
|
||||
}
|
||||
|
||||
totalTurns := 0
|
||||
for _, m := range messages {
|
||||
if role, _ := m["role"].(string); role == "user" {
|
||||
totalTurns++
|
||||
}
|
||||
}
|
||||
if totalTurns <= CompactionKeepTurns {
|
||||
return messages, event
|
||||
}
|
||||
|
||||
keepMessages := CompactionKeepTurns * 2
|
||||
split := len(messages) - keepMessages
|
||||
if split < 0 {
|
||||
split = 0
|
||||
}
|
||||
older, recent := messages[:split], messages[split:]
|
||||
|
||||
// C1 rolling persistence + #141 identity rematch.
|
||||
sessionKey := sessionID != ""
|
||||
var prior *RollingState
|
||||
if sessionKey {
|
||||
prior = RollingSummaryState(source, sessionID)
|
||||
}
|
||||
through := 0
|
||||
if prior != nil {
|
||||
through = prior.ThroughCount
|
||||
if prior.ThroughID != "" {
|
||||
var anchor Message
|
||||
if 0 <= through && through < len(older) {
|
||||
anchor = older[through]
|
||||
}
|
||||
if anchor != nil && ThroughID(older, through) != prior.ThroughID {
|
||||
// window rotated under the stored count: rematch by content
|
||||
found := -1
|
||||
for i := range older {
|
||||
if ThroughID(older, i) == prior.ThroughID {
|
||||
found = i
|
||||
break
|
||||
}
|
||||
}
|
||||
if found >= 0 {
|
||||
through = found
|
||||
} else {
|
||||
prior = nil
|
||||
through = 0
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var newOlder, archiveMessages []Message
|
||||
if prior != nil && through <= len(older) {
|
||||
newOlder = older[through:]
|
||||
if len(newOlder) == 0 {
|
||||
// Nothing new beyond the rolled summary: reuse it verbatim.
|
||||
summaryBlock := Message{
|
||||
"role": "user",
|
||||
"content": "## Conversation summary (rolling)\n" + prior.Summary +
|
||||
"\n(Earlier turns were archived; recent turns verbatim.)",
|
||||
}
|
||||
compacted := append([]Message{summaryBlock}, recent...)
|
||||
event.Triggered = true
|
||||
event.TokensAfter = sumTokens(compacted)
|
||||
event.TurnsSummarized = 0
|
||||
event.Summary = prior.Summary
|
||||
event.Persisted = true
|
||||
return compacted, event
|
||||
}
|
||||
archiveMessages = newOlder
|
||||
} else {
|
||||
newOlder = older
|
||||
if prior == nil {
|
||||
archiveMessages = messages
|
||||
} else if through <= len(older) {
|
||||
archiveMessages = older[through:]
|
||||
} else {
|
||||
archiveMessages = older
|
||||
}
|
||||
if prior != nil && through > len(older) {
|
||||
// History shrank (ledger rotation): summarize from scratch.
|
||||
prior = nil
|
||||
through = 0
|
||||
archiveMessages = messages
|
||||
}
|
||||
}
|
||||
|
||||
// ARCHIVE FIRST — raw, verbatim, before any lossy compression.
|
||||
if archiver != nil {
|
||||
record := map[string]interface{}{
|
||||
"session_id": sessionID,
|
||||
"source": source,
|
||||
"messages": archiveMessages,
|
||||
}
|
||||
if err := archiver(record); err != nil {
|
||||
event.Failed = true
|
||||
return messages, event
|
||||
}
|
||||
event.Archived = true
|
||||
}
|
||||
|
||||
var summaryText string
|
||||
{
|
||||
var input []Message
|
||||
if prior != nil {
|
||||
input = append([]Message{{"role": "user", "content": "Previous rolling summary:\n" + prior.Summary}}, newOlder...)
|
||||
} else {
|
||||
input = older
|
||||
}
|
||||
text, err := summarizer(input)
|
||||
if err != nil {
|
||||
event.Failed = true
|
||||
// C2: degrade to the newest tail that fits the window budget.
|
||||
tailBudget := DefaultBudget(window)
|
||||
var tail []Message
|
||||
used := 0
|
||||
for i := len(messages) - 1; i >= 0; i-- {
|
||||
message := messages[i]
|
||||
cost := DefaultEstimator(messageContent(message))
|
||||
if used+cost > tailBudget && len(tail) > 0 {
|
||||
break
|
||||
}
|
||||
tail = append([]Message{message}, tail...)
|
||||
used += cost
|
||||
}
|
||||
event.DegradedToTail = &[]int{len(tail)}[0]
|
||||
return tail, event
|
||||
}
|
||||
summaryText = strings.TrimSpace(text)
|
||||
}
|
||||
|
||||
archiveLabel := "external storage"
|
||||
if archivePath != "" {
|
||||
archiveLabel = archivePath
|
||||
}
|
||||
summaryBlock := Message{
|
||||
"role": "user",
|
||||
"content": "## Conversation summary (rolling)\n" + summaryText +
|
||||
"\n(Earlier turns were archived to " + archiveLabel + "; " +
|
||||
fmt.Sprintf("%d", len(older)) + " messages summarized, recent turns verbatim.)",
|
||||
}
|
||||
compacted := append([]Message{summaryBlock}, recent...)
|
||||
if sessionKey {
|
||||
newThrough := through + len(newOlder)
|
||||
StoreRollingSummary(source, sessionID, summaryText, newThrough, ThroughID(messages, newThrough))
|
||||
}
|
||||
event.Triggered = true
|
||||
event.TokensAfter = sumTokens(compacted)
|
||||
event.TurnsSummarized = len(newOlder)
|
||||
event.Summary = summaryText
|
||||
event.Persisted = sessionKey
|
||||
return compacted, event
|
||||
}
|
||||
|
||||
func messageContent(m Message) string {
|
||||
c, _ := m["content"].(string)
|
||||
return c
|
||||
}
|
||||
|
||||
func sumTokens(messages []Message) int {
|
||||
total := 0
|
||||
for _, m := range messages {
|
||||
total += DefaultEstimator(messageContent(m))
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
// ── summarize_history transport seam ────────────────────────────────────────
|
||||
|
||||
// SummarizeHistory builds the structured rolling-summary prompt and calls the
|
||||
// injected transport. The transport (OllamaChatTransport in the oracle) is
|
||||
// IO; here it is injectable. Empty/blank content raises (caller degrades).
|
||||
func SummarizeHistory(messages []Message, model string, numPredict int, numCtx int, transport func(prompt, model string, options map[string]interface{}) (string, error)) (string, error) {
|
||||
parts := make([]string, 0, len(messages))
|
||||
for _, m := range messages {
|
||||
role := "?"
|
||||
if r, ok := m["role"].(string); ok && r != "" {
|
||||
role = r
|
||||
}
|
||||
parts = append(parts, role+": "+messageContent(m))
|
||||
}
|
||||
transcript := strings.Join(parts, "\n")
|
||||
prompt := "Summarize this conversation segment into a rolling summary.\n" +
|
||||
"Answer with exactly these four labeled sections:\n" +
|
||||
"Session intent: <what this session is trying to accomplish>\n" +
|
||||
"Decisions: <choices made so far>\n" +
|
||||
"Open items: <unresolved threads>\n" +
|
||||
"Next steps: <what to do next>\n\n" +
|
||||
"Conversation:\n" + transcript
|
||||
options := map[string]interface{}{
|
||||
"temperature": 0.2,
|
||||
"num_predict": numPredict,
|
||||
}
|
||||
if numCtx != 0 {
|
||||
options["num_ctx"] = numCtx
|
||||
}
|
||||
content, err := transport(prompt, model, options)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if strings.TrimSpace(content) == "" {
|
||||
return "", fmt.Errorf("summarizer returned empty content")
|
||||
}
|
||||
return strings.TrimSpace(content), nil
|
||||
}
|
||||
|
||||
// ── replay CLI (legacy truth.json flat-case driver, kept green) ─────────────
|
||||
|
||||
func main() {
|
||||
if len(os.Args) < 2 || os.Args[1] != "replay" {
|
||||
fmt.Fprintln(os.Stderr, "usage: context-broker replay <truth.json>")
|
||||
@@ -52,8 +626,18 @@ func main() {
|
||||
}
|
||||
var cases []map[string]any
|
||||
if err := json.Unmarshal(raw, &cases); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
// New-format truth doc ({frozen_ts, generator, rows}) — the legacy
|
||||
// flat ops ride inside row["input"]["ops"] / scalar cases; nothing to
|
||||
// replay here, the battery test covers them.
|
||||
var doc struct {
|
||||
Rows []json.RawMessage `json:"rows"`
|
||||
}
|
||||
if err2 := json.Unmarshal(raw, &doc); err2 != nil || len(doc.Rows) == 0 {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("PARITY OK: %d battery rows (go test battery replays them)\n", len(doc.Rows))
|
||||
return
|
||||
}
|
||||
fails := 0
|
||||
for i, c := range cases {
|
||||
@@ -62,19 +646,19 @@ func main() {
|
||||
switch op {
|
||||
case "estimator":
|
||||
text, _ := c["text"].(string)
|
||||
if got := defaultEstimator(text); got != want {
|
||||
if got := DefaultEstimator(text); got != want {
|
||||
fails++
|
||||
fmt.Printf("case %d estimator %q: got %d want %d\n", i, text, got, want)
|
||||
}
|
||||
case "budget":
|
||||
w := int(c["window"].(float64))
|
||||
if got := defaultBudget(w); got != want {
|
||||
if got := DefaultBudget(w); got != want {
|
||||
fails++
|
||||
fmt.Printf("case %d budget %d: got %d want %d\n", i, w, got, want)
|
||||
}
|
||||
case "mem_budget":
|
||||
w := int(c["window"].(float64))
|
||||
if got := defaultMemoryBudget(w); got != want {
|
||||
if got := DefaultMemoryBudget(w); got != want {
|
||||
fails++
|
||||
fmt.Printf("case %d mem_budget %d: got %d want %d\n", i, w, got, want)
|
||||
}
|
||||
@@ -86,3 +670,114 @@ func main() {
|
||||
}
|
||||
fmt.Printf("PARITY OK: %d cases vs the Python oracle\n", len(cases))
|
||||
}
|
||||
|
||||
// ── eval CLI (live-parity harness: one JSON payload on stdin → JSON out) ────
|
||||
|
||||
func init() {
|
||||
if len(os.Args) >= 2 && os.Args[1] == "eval" {
|
||||
var payload map[string]interface{}
|
||||
if err := json.NewDecoder(os.Stdin).Decode(&payload); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
out := evalOne(payload)
|
||||
blob, _ := json.Marshal(out)
|
||||
fmt.Println(string(blob))
|
||||
os.Exit(0)
|
||||
}
|
||||
}
|
||||
|
||||
func evalOne(p map[string]interface{}) interface{} {
|
||||
op, _ := p["op"].(string)
|
||||
switch op {
|
||||
case "est":
|
||||
if t, ok := p["text"]; ok {
|
||||
s, _ := t.(string)
|
||||
return PyDefaultEstimator(&s)
|
||||
}
|
||||
return PyDefaultEstimator(nil)
|
||||
case "budget":
|
||||
return PyDefaultBudget(int(p["window"].(float64)))
|
||||
case "membud":
|
||||
return PyDefaultMemoryBudget(int(p["window"].(float64)))
|
||||
case "memoff":
|
||||
v, _ := p["value"].(string)
|
||||
return PyMemoryDisabledValue(v)
|
||||
case "through":
|
||||
rawMsgs, _ := p["messages"].([]interface{})
|
||||
msgs := make([]Message, 0, len(rawMsgs))
|
||||
for _, rm := range rawMsgs {
|
||||
m, _ := rm.(map[string]interface{})
|
||||
msgs = append(msgs, m)
|
||||
}
|
||||
return ThroughID(msgs, int(p["through"].(float64)))
|
||||
case "admit":
|
||||
rawEntries, _ := p["entries"].([]interface{})
|
||||
entries := make([]AdmissionEntry, 0, len(rawEntries))
|
||||
for _, re := range rawEntries {
|
||||
e, _ := re.(map[string]interface{})
|
||||
b := &ContributionBlock{}
|
||||
b.Name, _ = e["name"].(string)
|
||||
b.Priority = int(e["priority"].(float64))
|
||||
b.Tokens = int(e["tokens"].(float64))
|
||||
b.Source, _ = e["source"].(string)
|
||||
b.Evictable, _ = e["evictable"].(bool)
|
||||
entries = append(entries, AdmissionEntry{Name: b.Name, Block: b})
|
||||
}
|
||||
budget := int(p["budget"].(float64))
|
||||
memBudget := int(p["memory_budget"].(float64))
|
||||
res, err := Admit(entries, budget, memBudget)
|
||||
if err != nil {
|
||||
return map[string]interface{}{"error": "ContextOverflowError", "detail": err.Error()}
|
||||
}
|
||||
blocksOut := make([]map[string]interface{}, 0, len(res.Blocks))
|
||||
for _, r := range res.Blocks {
|
||||
blocksOut = append(blocksOut, map[string]interface{}{
|
||||
"name": r.Name, "priority": r.Priority, "tokens": r.Tokens,
|
||||
"source": r.Source, "evictable": r.Evictable,
|
||||
"admitted": r.Admitted, "reason": r.Reason,
|
||||
})
|
||||
}
|
||||
dropped := make([]string, 0, len(res.Dropped))
|
||||
for _, d := range res.Dropped {
|
||||
dropped = append(dropped, d.Name)
|
||||
}
|
||||
return map[string]interface{}{
|
||||
"admitted": blockNames(res.Admitted), "blocks": blocksOut,
|
||||
"admitted_total": res.AdmittedTotal, "dropped": dropped,
|
||||
"memory_budget": res.MemoryBudget, "memory_used": res.MemoryUsed,
|
||||
}
|
||||
case "render":
|
||||
rawBlocks, _ := p["blocks"].([]interface{})
|
||||
admitted := make([]*ContributionBlock, 0, len(rawBlocks))
|
||||
for _, rb := range rawBlocks {
|
||||
b, _ := rb.(map[string]interface{})
|
||||
blk := &ContributionBlock{}
|
||||
blk.Name, _ = b["name"].(string)
|
||||
blk.Priority = int(b["priority"].(float64))
|
||||
blk.Tokens = int(b["tokens"].(float64))
|
||||
blk.Source, _ = b["source"].(string)
|
||||
switch c := b["content"].(type) {
|
||||
case string:
|
||||
blk.Content = c
|
||||
case []interface{}:
|
||||
msgList := make([]Message, 0, len(c))
|
||||
for _, rm := range c {
|
||||
m, _ := rm.(map[string]interface{})
|
||||
msgList = append(msgList, m)
|
||||
}
|
||||
blk.Content = msgList
|
||||
}
|
||||
admitted = append(admitted, blk)
|
||||
}
|
||||
capsule := Render(admitted)
|
||||
return map[string]interface{}{
|
||||
"token_count": capsule.TokenCount,
|
||||
"messages": capsule.Messages,
|
||||
"system_preamble": capsule.SystemPreamble,
|
||||
"user_content": capsule.UserContent,
|
||||
"blocks": blockNames(capsule.Blocks),
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
+2118
-47
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,559 @@
|
||||
package main
|
||||
|
||||
// Battery — the HARD parity gate replaying the frozen oracle truth
|
||||
// (truth.json, generated against the LIVE Python oracle
|
||||
// src/rune/context_broker.py with RUNE_COGNITION_CORE=python pinned)
|
||||
// VERBATIM. Inputs are deep-frozen snapshots (the oracle mutates message
|
||||
// lists); every row byte-compares the Go result against the oracle's.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type TruthDoc struct {
|
||||
FrozenTS string `json:"frozen_ts"`
|
||||
Generator string `json:"generator"`
|
||||
Rows []TruthRow `json:"rows"`
|
||||
}
|
||||
|
||||
type TruthRow struct {
|
||||
Name string `json:"name"`
|
||||
Fn string `json:"fn"`
|
||||
Input map[string]interface{} `json:"input"`
|
||||
Want interface{} `json:"want"`
|
||||
}
|
||||
|
||||
func loadTruth(t *testing.T) *TruthDoc {
|
||||
t.Helper()
|
||||
raw, err := os.ReadFile("truth.json")
|
||||
if err != nil {
|
||||
t.Fatalf("truth.json missing (frozen oracle battery): %v", err)
|
||||
}
|
||||
var doc TruthDoc
|
||||
if err := json.Unmarshal(raw, &doc); err != nil {
|
||||
t.Fatalf("parse truth.json: %v", err)
|
||||
}
|
||||
if len(doc.Rows) == 0 {
|
||||
t.Fatal("truth.json contained no rows")
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, r := range doc.Rows {
|
||||
if seen[r.Name] {
|
||||
t.Fatalf("duplicate frozen row %q", r.Name)
|
||||
}
|
||||
seen[r.Name] = true
|
||||
}
|
||||
return &doc
|
||||
}
|
||||
|
||||
func msgs(raw interface{}) []Message {
|
||||
list, _ := raw.([]interface{})
|
||||
out := make([]Message, 0, len(list))
|
||||
for _, item := range list {
|
||||
m, _ := item.(map[string]interface{})
|
||||
out = append(out, m)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func asInt(v interface{}) int {
|
||||
switch n := v.(type) {
|
||||
case float64:
|
||||
return int(n)
|
||||
case bool:
|
||||
if n {
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
case string:
|
||||
var i int
|
||||
fmt.Sscanf(n, "%d", &i)
|
||||
return i
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func compEvent(e *CompactEvent) map[string]interface{} {
|
||||
blob, _ := json.Marshal(e)
|
||||
var out map[string]interface{}
|
||||
_ = json.Unmarshal(blob, &out)
|
||||
return out
|
||||
}
|
||||
|
||||
// runOne — replay one frozen row; returns "" on parity, else the divergence.
|
||||
func runOne(row *TruthRow) string {
|
||||
switch row.Fn {
|
||||
case "est":
|
||||
var textPtr *string
|
||||
if s, ok := row.Input["text"].(string); ok {
|
||||
textPtr = &s
|
||||
}
|
||||
got := PyDefaultEstimator(textPtr)
|
||||
want := asInt(row.Want)
|
||||
if got != want {
|
||||
return fmt.Sprintf("est: got %d want %d", got, want)
|
||||
}
|
||||
case "cc_core":
|
||||
env, _ := row.Input["env"].(map[string]interface{})
|
||||
lookup := func(string) (string, bool) { return "", false }
|
||||
if v, ok := env["RUNE_COGNITION_CORE"]; ok {
|
||||
s, _ := v.(string)
|
||||
lookup = func(string) (string, bool) { return s, true }
|
||||
}
|
||||
loaderAvail, _ := row.Input["loader_available"].(bool)
|
||||
got := CcCore(lookup, loaderAvail)
|
||||
if row.Want == nil && got != nil {
|
||||
return "cc_core: got core want none"
|
||||
}
|
||||
if row.Want != nil && got == nil {
|
||||
return "cc_core: got none want core"
|
||||
}
|
||||
case "budget":
|
||||
got := PyDefaultBudget(asInt(row.Input["window"]))
|
||||
want := asInt(row.Want)
|
||||
if got != want {
|
||||
return fmt.Sprintf("budget: got %d want %d", got, want)
|
||||
}
|
||||
case "membud":
|
||||
got := PyDefaultMemoryBudget(asInt(row.Input["window"]))
|
||||
want := asInt(row.Want)
|
||||
if got != want {
|
||||
return fmt.Sprintf("membud: got %d want %d", got, want)
|
||||
}
|
||||
case "memoff":
|
||||
v, _ := row.Input["value"].(string)
|
||||
got := PyMemoryDisabledValue(v)
|
||||
want, _ := row.Want.(bool)
|
||||
if got != want {
|
||||
return fmt.Sprintf("memoff %q: got %v want %v", v, got, want)
|
||||
}
|
||||
case "windows":
|
||||
ResetDiscoveredWindows()
|
||||
opsList, _ := row.Input["ops"].([]interface{})
|
||||
for _, rawOp := range opsList {
|
||||
op, _ := rawOp.(map[string]interface{})
|
||||
switch op["op"] {
|
||||
case "register_window":
|
||||
RegisterWindow(op["model_id"].(string), asInt(op["window"]))
|
||||
case "reset_discovered_windows":
|
||||
ResetDiscoveredWindows()
|
||||
case "default_window":
|
||||
mid := ""
|
||||
if s, ok := op["model_id"].(string); ok {
|
||||
mid = s
|
||||
}
|
||||
// Live discovery is IO; the Go seam stays offline (nil) so
|
||||
// the miss branch freezes to the fallback — exactly what the
|
||||
// oracle froze in the offline generator env.
|
||||
DiscoverWindow = nil
|
||||
got := DefaultWindow(mid)
|
||||
wantMap, _ := row.Want.(map[string]interface{})
|
||||
want := asInt(wantMap["last"])
|
||||
if got != want {
|
||||
return fmt.Sprintf("default_window %q: got %d want %d", mid, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
case "through":
|
||||
got := ThroughID(msgs(row.Input["messages"]), asInt(row.Input["through"]))
|
||||
want, _ := row.Want.(string)
|
||||
if got != want {
|
||||
return fmt.Sprintf("through: got %q want %q", got, want)
|
||||
}
|
||||
case "rstate":
|
||||
ResetDiscoveredWindows()
|
||||
for k := range rolling {
|
||||
delete(rolling, k)
|
||||
}
|
||||
opsList, _ := row.Input["ops"].([]interface{})
|
||||
var got *RollingState
|
||||
for _, rawOp := range opsList {
|
||||
op, _ := rawOp.(map[string]interface{})
|
||||
source, _ := op["source"].(string)
|
||||
sid, _ := op["sid"].(string)
|
||||
switch op["op"] {
|
||||
case "store":
|
||||
summary, _ := op["summary"].(string)
|
||||
tid, _ := op["tid"].(string)
|
||||
StoreRollingSummary(source, sid, summary, asInt(op["through"]), tid)
|
||||
case "read":
|
||||
got = RollingSummaryState(source, sid)
|
||||
}
|
||||
}
|
||||
if row.Want == nil {
|
||||
if got != nil {
|
||||
return "rstate: got state want nil"
|
||||
}
|
||||
} else {
|
||||
wantMap, _ := row.Want.(map[string]interface{})
|
||||
if got == nil {
|
||||
return "rstate: got nil want state"
|
||||
}
|
||||
if got.Summary != wantMap["summary"] ||
|
||||
got.ThroughCount != asInt(wantMap["through_count"]) ||
|
||||
got.ThroughID != wantMap["through_id"] {
|
||||
return fmt.Sprintf("rstate: got %+v want %v", got, wantMap)
|
||||
}
|
||||
}
|
||||
case "rstore":
|
||||
m := msgs(row.Input["msgs"])
|
||||
StoreRollingSummary("operator", "s2", "S2", 1, ThroughID(m, 1))
|
||||
got := RollingSummaryState("operator", "s2")
|
||||
wantMap, _ := row.Want.(map[string]interface{})
|
||||
if got == nil || got.Summary != wantMap["summary"] ||
|
||||
got.ThroughCount != asInt(wantMap["through_count"]) ||
|
||||
got.ThroughID != wantMap["through_id"] {
|
||||
return fmt.Sprintf("rstore: got %+v want %v", got, wantMap)
|
||||
}
|
||||
case "summarize":
|
||||
// The oracle froze the transport-failure branch (offline env): any
|
||||
// error type is acceptable — the Go seam reproduces the failure
|
||||
// shape (non-empty error, empty-content ValueError contract).
|
||||
_, err := SummarizeHistory(msgs(row.Input["messages"]), "offline", 256, 0,
|
||||
func(string, string, map[string]interface{}) (string, error) {
|
||||
return "", fmt.Errorf("HTTP Error 404: Not Found")
|
||||
})
|
||||
wantMap, _ := row.Want.(map[string]interface{})
|
||||
if _, isErr := wantMap["error"]; isErr && err == nil {
|
||||
return "summarize: got success want error"
|
||||
}
|
||||
if err == nil {
|
||||
return "summarize: expected frozen error branch"
|
||||
}
|
||||
case "compact":
|
||||
window := asInt(row.Input["window"])
|
||||
sid, _ := row.Input["session"].(string)
|
||||
in := msgs(row.Input["messages"])
|
||||
ResetDiscoveredWindows()
|
||||
for k := range rolling {
|
||||
delete(rolling, k)
|
||||
}
|
||||
var archiver Archiver
|
||||
archivedMsgCount := -1
|
||||
if _, wantsArchive := row.Input["archive"]; wantsArchive {
|
||||
archiver = func(record map[string]interface{}) error {
|
||||
archivedMsgCount = len(record["messages"].([]Message))
|
||||
return nil
|
||||
}
|
||||
}
|
||||
archivePath := ""
|
||||
if _, wantsArchive := row.Input["archive"]; wantsArchive {
|
||||
// deterministic relative fixture path frozen in the input
|
||||
archivePath, _ = row.Input["archive_path"].(string)
|
||||
}
|
||||
summarizerText, _ := row.Input["summarizer"].(string)
|
||||
var summarizer Summarizer
|
||||
switch summarizerText {
|
||||
case "boom":
|
||||
summarizer = func([]Message) (string, error) {
|
||||
return "", fmt.Errorf("summarizer down")
|
||||
}
|
||||
default:
|
||||
summarizer = func(ms []Message) (string, error) {
|
||||
parts := make([]string, 0, len(ms))
|
||||
for _, m := range ms {
|
||||
c := messageContent(m)
|
||||
if len(c) > 10 {
|
||||
c = c[:10]
|
||||
}
|
||||
parts = append(parts, c)
|
||||
}
|
||||
return "SUMMARY<" + strings.Join(parts, "|") + ">", nil
|
||||
}
|
||||
}
|
||||
if rawPrior, ok := row.Input["prior"]; ok {
|
||||
priorMap, _ := rawPrior.(map[string]interface{})
|
||||
priorSid := sid
|
||||
StoreRollingSummary("operator", priorSid,
|
||||
priorMap["summary"].(string), asInt(priorMap["through_count"]),
|
||||
priorMap["through_id"].(string))
|
||||
}
|
||||
compacted, ev := CompactHistory(in, window, archivePath, archiver, summarizer, "operator", sid)
|
||||
wantMap, _ := row.Want.(map[string]interface{})
|
||||
wantEvent, _ := wantMap["event"].(map[string]interface{})
|
||||
if msg := diffEvent(wantEvent, compEvent(ev)); msg != "" {
|
||||
return msg
|
||||
}
|
||||
if wantCompacted, ok := wantMap["compacted"]; ok {
|
||||
wantMsgs := msgs(wantCompacted)
|
||||
if !reflect.DeepEqual(normalize(compacted), normalize(wantMsgs)) {
|
||||
gj, _ := json.Marshal(normalize(compacted))
|
||||
wj, _ := json.Marshal(normalize(wantMsgs))
|
||||
return fmt.Sprintf("compacted divergence:\n got %s\n want %s", gj, wj)
|
||||
}
|
||||
}
|
||||
if _, wantsArchive := row.Input["archive"]; wantsArchive {
|
||||
wantCount := asInt(wantMap["archived_msg_count"])
|
||||
if archivedMsgCount != wantCount {
|
||||
return fmt.Sprintf("archived_msg_count: got %d want %d", archivedMsgCount, wantCount)
|
||||
}
|
||||
}
|
||||
default:
|
||||
return "unknown fn key " + row.Fn
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// normalize renders messages compactly for deep comparison (dicts with
|
||||
// arbitrary key order).
|
||||
func normalize(ms []Message) []map[string]interface{} {
|
||||
out := make([]map[string]interface{}, 0, len(ms))
|
||||
for _, m := range ms {
|
||||
cp := map[string]interface{}{}
|
||||
for k, v := range m {
|
||||
cp[k] = v
|
||||
}
|
||||
out = append(out, cp)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func diffEvent(want, got map[string]interface{}) string {
|
||||
keys := make([]string, 0, len(want))
|
||||
for k := range want {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
for _, k := range keys {
|
||||
wv := fmt.Sprintf("%v", want[k])
|
||||
gv := fmt.Sprintf("%v", got[k])
|
||||
// Python bools render True/False; Go %v renders true/false.
|
||||
wv = strings.NewReplacer("False", "false", "True", "true").Replace(wv)
|
||||
if wv != gv {
|
||||
return fmt.Sprintf("event.%s: got %s want %s", k, gv, wv)
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// runBrokerOne — the ContextBroker-kernel rows (to_dict/collect/admit/render).
|
||||
func runBrokerOne(row *TruthRow) string {
|
||||
switch row.Fn {
|
||||
case "to_dict":
|
||||
name, _ := row.Want.(map[string]interface{})["name"].(string)
|
||||
b := &ContributionBlock{}
|
||||
switch name {
|
||||
case "memory":
|
||||
b = &ContributionBlock{Name: "memory", Priority: 5, Content: "hello world", Tokens: 2, Source: "recall", Evictable: true}
|
||||
case "system":
|
||||
b = &ContributionBlock{Name: "system", Priority: 100, Content: "sys", Tokens: 0, Source: "persona", Evictable: false}
|
||||
}
|
||||
want, _ := row.Want.(map[string]interface{})
|
||||
got := b.ToDict()
|
||||
for _, k := range []string{"name", "priority", "content", "tokens", "source", "evictable"} {
|
||||
wv := fmt.Sprintf("%v", want[k])
|
||||
wv = strings.NewReplacer("False", "false", "True", "true").Replace(wv)
|
||||
if gv := fmt.Sprintf("%v", got[k]); gv != wv {
|
||||
return fmt.Sprintf("to_dict.%s: got %v want %v", k, gv, wv)
|
||||
}
|
||||
}
|
||||
case "collect":
|
||||
cb := NewContextBroker()
|
||||
var failures []string
|
||||
cb.Fail = func(name string, err error) { failures = append(failures, name+": "+err.Error()) }
|
||||
cb.RegisterContributor("ok", func(map[string]interface{}, map[string]interface{}) *ContributionBlock {
|
||||
return &ContributionBlock{Name: "tools", Priority: 9, Content: "tooltext", Tokens: 2, Source: "t"}
|
||||
}, 1)
|
||||
cb.RegisterContributor("empty", func(map[string]interface{}, map[string]interface{}) *ContributionBlock {
|
||||
return &ContributionBlock{Name: "x", Priority: 1, Content: "", Source: "s"}
|
||||
}, 1)
|
||||
cb.RegisterContributor("boom", func(map[string]interface{}, map[string]interface{}) *ContributionBlock {
|
||||
panic(fmt.Errorf("cb down"))
|
||||
}, 1)
|
||||
cb.RegisterContributor("malformed", func(map[string]interface{}, map[string]interface{}) *ContributionBlock {
|
||||
return &ContributionBlock{Name: "y", Priority: 1, Content: 12345, Source: "s"}
|
||||
}, 1)
|
||||
cb.RegisterContributor("list-ok", func(map[string]interface{}, map[string]interface{}) *ContributionBlock {
|
||||
return &ContributionBlock{Name: "history", Priority: 3, Content: []Message{{"role": "user", "content": "hi"}}, Tokens: 1, Source: "h"}
|
||||
}, 2)
|
||||
cb.RegisterContributor("none", func(map[string]interface{}, map[string]interface{}) *ContributionBlock {
|
||||
return nil
|
||||
}, 1)
|
||||
collected, degraded := cb.Collect(map[string]interface{}{}, map[string]interface{}{})
|
||||
var names []string
|
||||
for _, e := range collected {
|
||||
names = append(names, e.Name)
|
||||
}
|
||||
want, _ := row.Want.(map[string]interface{})
|
||||
wantNames := toStrSlice(want["names"])
|
||||
wantDegraded := toStrSlice(want["degraded"])
|
||||
if fmt.Sprint(names) != fmt.Sprint(wantNames) {
|
||||
return fmt.Sprintf("collect.names: got %v want %v", names, wantNames)
|
||||
}
|
||||
if fmt.Sprint(degraded) != fmt.Sprint(wantDegraded) {
|
||||
return fmt.Sprintf("collect.degraded: got %v want %v", degraded, wantDegraded)
|
||||
}
|
||||
// both failure kinds recorded (T1): boom + malformed
|
||||
if len(failures) != 2 {
|
||||
return fmt.Sprintf("collect.failures: got %d want 2 (%v)", len(failures), failures)
|
||||
}
|
||||
case "admit":
|
||||
want, _ := row.Want.(map[string]interface{})
|
||||
var entries []AdmissionEntry
|
||||
var budget, memBudget int
|
||||
switch row.Name {
|
||||
case "admit:priority-recency-budget":
|
||||
budget, memBudget = 1000, DefaultMemoryBudget(0)
|
||||
entries = []AdmissionEntry{
|
||||
{"system", &ContributionBlock{Name: "system", Priority: 100, Tokens: 500, Source: "src-system"}},
|
||||
{"memory", &ContributionBlock{Name: "memory", Priority: 5, Tokens: 300, Source: "src-memory", Evictable: true}},
|
||||
{"history", &ContributionBlock{Name: "history", Priority: 7, Tokens: 900, Source: "src-history", Evictable: true}},
|
||||
{"user", &ContributionBlock{Name: "user", Priority: 100, Tokens: 100, Source: "src-user"}},
|
||||
}
|
||||
case "admit:memory-subbudget-drop":
|
||||
budget, memBudget = 4000, 1200
|
||||
entries = []AdmissionEntry{
|
||||
{"m1", &ContributionBlock{Name: "m1", Priority: 9, Tokens: 700, Source: "src-m1", Evictable: true}},
|
||||
{"m2", &ContributionBlock{Name: "m2", Priority: 8, Tokens: 700, Source: "src-m2", Evictable: true}},
|
||||
}
|
||||
case "admit:mandatory-overflow-raises":
|
||||
budget, memBudget = 1000, DefaultMemoryBudget(0)
|
||||
entries = []AdmissionEntry{
|
||||
{"user", &ContributionBlock{Name: "user", Priority: 100, Tokens: 5000, Source: "src-user"}},
|
||||
}
|
||||
_, err := Admit(entries, budget, memBudget)
|
||||
wantMap, _ := row.Want.(map[string]interface{})
|
||||
if wantMap["error"] == "ContextOverflowError" {
|
||||
if err == nil || !errors.Is(err, ErrContextOverflow) {
|
||||
return "admit.overflow: no raise"
|
||||
}
|
||||
if !strings.Contains(err.Error(), wantMap["detail"].(string)) {
|
||||
return fmt.Sprintf("admit.overflow detail: got %q", err.Error())
|
||||
}
|
||||
return ""
|
||||
}
|
||||
return "admit.overflow: frozen row unexpected shape"
|
||||
}
|
||||
res, err := Admit(entries, budget, memBudget)
|
||||
if err != nil {
|
||||
return "admit: unexpected error " + err.Error()
|
||||
}
|
||||
if got := fmt.Sprint(blockNames(res.Admitted)); got != fmt.Sprint(toStrSlice(want["admitted"])) {
|
||||
return fmt.Sprintf("admit.admitted: got %v want %v", blockNames(res.Admitted), toStrSlice(want["admitted"]))
|
||||
}
|
||||
if wb, ok := want["blocks"]; ok {
|
||||
wantBlocks, _ := wb.([]interface{})
|
||||
if len(wantBlocks) != len(res.Blocks) {
|
||||
return fmt.Sprintf("admit.blocks len: got %d want %d", len(res.Blocks), len(wantBlocks))
|
||||
}
|
||||
for i, wbItem := range wantBlocks {
|
||||
wbMap, _ := wbItem.(map[string]interface{})
|
||||
gb := res.Blocks[i]
|
||||
for _, k := range []string{"name", "priority", "tokens", "source", "evictable", "admitted", "reason"} {
|
||||
wv := fmt.Sprintf("%v", wbMap[k])
|
||||
wv = strings.NewReplacer("False", "false", "True", "true").Replace(wv)
|
||||
var gv string
|
||||
switch k {
|
||||
case "name":
|
||||
gv = gb.Name
|
||||
case "priority":
|
||||
gv = fmt.Sprintf("%d", gb.Priority)
|
||||
case "tokens":
|
||||
gv = fmt.Sprintf("%d", gb.Tokens)
|
||||
case "source":
|
||||
gv = gb.Source
|
||||
case "evictable":
|
||||
gv = fmt.Sprintf("%v", gb.Evictable)
|
||||
case "admitted":
|
||||
gv = fmt.Sprintf("%v", gb.Admitted)
|
||||
case "reason":
|
||||
gv = gb.Reason
|
||||
}
|
||||
if gv != wv {
|
||||
return fmt.Sprintf("admit.blocks[%d].%s: got %q want %q", i, k, gv, wv)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if want["admitted_total"] != nil && res.AdmittedTotal != asInt(want["admitted_total"]) {
|
||||
return fmt.Sprintf("admit.admitted_total: got %d want %v", res.AdmittedTotal, want["admitted_total"])
|
||||
}
|
||||
if wd, ok := want["dropped"]; ok {
|
||||
var gotDropped []string
|
||||
for _, d := range res.Dropped {
|
||||
gotDropped = append(gotDropped, d.Name)
|
||||
}
|
||||
if fmt.Sprint(gotDropped) != fmt.Sprint(toStrSlice(wd)) {
|
||||
return fmt.Sprintf("admit.dropped: got %v want %v", gotDropped, toStrSlice(wd))
|
||||
}
|
||||
}
|
||||
if want["memory_used"] != nil && res.MemoryUsed != asInt(want["memory_used"]) {
|
||||
return fmt.Sprintf("admit.memory_used: got %d want %v", res.MemoryUsed, want["memory_used"])
|
||||
}
|
||||
if want["memory_budget"] != nil && res.MemoryBudget != asInt(want["memory_budget"]) {
|
||||
return fmt.Sprintf("admit.memory_budget: got %d want %v", res.MemoryBudget, want["memory_budget"])
|
||||
}
|
||||
case "render":
|
||||
want, _ := row.Want.(map[string]interface{})
|
||||
admitted := []*ContributionBlock{
|
||||
{Name: "persona", Priority: 100, Content: "ccccccc", Tokens: 7, Source: "src-persona"},
|
||||
{Name: "tools", Priority: 90, Content: "ccccc", Tokens: 5, Source: "src-tools"},
|
||||
{Name: "history", Priority: 70, Content: "cccc", Tokens: 4, Source: "src-history"},
|
||||
{Name: "user", Priority: 100, Content: "cc", Tokens: 2, Source: "src-user"},
|
||||
}
|
||||
capsule := Render(admitted)
|
||||
if capsule.TokenCount != asInt(want["token_count"]) {
|
||||
return fmt.Sprintf("render.token_count: got %d want %v", capsule.TokenCount, want["token_count"])
|
||||
}
|
||||
if capsule.SystemPreamble != want["system_preamble"] {
|
||||
return fmt.Sprintf("render.system_preamble: got %q want %q", capsule.SystemPreamble, want["system_preamble"])
|
||||
}
|
||||
if capsule.UserContent != want["user_content"] {
|
||||
return fmt.Sprintf("render.user_content: got %q want %q", capsule.UserContent, want["user_content"])
|
||||
}
|
||||
wantMsgs := msgs(want["messages"])
|
||||
if !reflect.DeepEqual(normalize(capsule.Messages), normalize(wantMsgs)) {
|
||||
gj, _ := json.Marshal(capsule.Messages)
|
||||
wj, _ := json.Marshal(wantMsgs)
|
||||
return fmt.Sprintf("render.messages:\n got %s\n want %s", gj, wj)
|
||||
}
|
||||
if fmt.Sprint(blockNames(capsule.Blocks)) != fmt.Sprint(toStrSlice(want["blocks"])) {
|
||||
return fmt.Sprintf("render.blocks: got %v want %v", blockNames(capsule.Blocks), toStrSlice(want["blocks"]))
|
||||
}
|
||||
default:
|
||||
return "unknown broker fn key " + row.Fn
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func toStrSlice(v interface{}) []string {
|
||||
list, ok := v.([]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(list))
|
||||
for _, item := range list {
|
||||
s, _ := item.(string)
|
||||
out = append(out, s)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestTruthBattery(t *testing.T) {
|
||||
doc := loadTruth(t)
|
||||
fails := 0
|
||||
for i := range doc.Rows {
|
||||
var msg string
|
||||
switch doc.Rows[i].Fn {
|
||||
case "to_dict", "collect", "admit", "render":
|
||||
msg = runBrokerOne(&doc.Rows[i])
|
||||
default:
|
||||
msg = runOne(&doc.Rows[i])
|
||||
}
|
||||
if msg != "" {
|
||||
fails++
|
||||
t.Errorf("DIVERGED %s: %s", doc.Rows[i].Name, msg)
|
||||
}
|
||||
}
|
||||
if fails > 0 {
|
||||
t.Fatalf("battery: %d/%d divergences vs the frozen Python oracle", fails, len(doc.Rows))
|
||||
}
|
||||
t.Logf("PARITY OK: %d frozen rows replayed", len(doc.Rows))
|
||||
}
|
||||
Reference in new issue
Block a user