feat(toolchain): port to FULL Rust — dichotomy wave

Contract BUILDER-2026-10-11T22:03:40Z-7aeb7a-factory-wave (#2479 tranche).
Faithful port of src/rune/toolchain.py: ToolchainConstraints known/unknown
classification, DecisionRecord normalization, _major_change,
resolve_toolchain (INSUFFICIENT_EVIDENCE gate, KEEP_EXISTING,
CHANGE_MAJOR, greenfield recommendation), CONSTRAINT_KEYS vocabulary.
Doc-envelope FFI (tl_* symbols, free via cs_free_string), ctypes loader
(tl_loader.py, TC_RUST_LIB override), facade delegation with
RUNE_TC_CORE=python pin. Frozen oracle truth: 43 cases across every
public function and branch (frozen_tc_truth.json), replayed VERBATIM by
branch_battery.rs (6 tests). Live parity 613/613 (truth + 570 fuzz,
seed 2479).
This commit is contained in:
Nick committed 2026-10-12 00:37:26 +01:00
1 parent 8289d896e1
commit f83396eecd
7 files changed
+2111 -44

No files matched your search

+232
View File
@@ -0,0 +1,232 @@
#!/usr/bin/env python3
"""Live parity: Rust toolchain kernel vs the Python oracle (dichotomy
wave #2479). Replays the frozen truth corpus PLUS seeded fuzz over every
public function and branch. The module is deterministic — no masking.
Both sides must agree on values AND on raising.
The facade reads RUNE_TC_CORE per call, so one process can flip between
the pinned oracle (RUNE_TC_CORE=python) and the kernel (unset).
"""
from __future__ import annotations
import json
import os
import random
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
LANGS = ["python", "rust", "typescript", "go", "zig", "", "C++",
"café-lang ☕", "r"]
KEYS = sorted([
"existing_language", "existing_runtime", "existing_framework",
"target_platform", "deployment", "performance_sensitive",
"latency_sensitive", "memory_constrained", "concurrency_model",
"interoperability", "gui_native", "expected_lifespan",
"user_constraints", "ecosystem_requirement"])
UNKNOWN_KEYS = ["wibble", "flurb", "custom_axis", "future_thing"]
VALS = ["linux", "server", "tokio", "svelte", True, False, None, 0, 1,
"", ["a", "b"], {"deep": True}, 2.5, "x" * 60]
KEYWORD_POOL = ["existing_language", "target_platform", "deployment",
"performance_sensitive", "latency_sensitive",
"memory_constrained", "gui_native", "wibble"]
def call_core(core, fn, *args, **kwargs):
from rune import toolchain as tc
old = os.environ.get("RUNE_TC_CORE")
os.environ["RUNE_TC_CORE"] = core
try:
return getattr(tc, fn)(*args, **kwargs)
finally:
if old is None:
os.environ.pop("RUNE_TC_CORE", None)
else:
os.environ["RUNE_TC_CORE"] = old
def toj(x):
return x.to_dict() if hasattr(x, "to_dict") else x
def constraints_of(core, kw):
return call_core(core, "ToolchainConstraints", **kw)
def run_pair(invoke):
"""invoke(core) runs the full call chain under one core pin. Returns
(ok, detail): ok True when values match, None when BOTH sides raised
(agreement on refusing), False on any divergence."""
outs = []
for core in ("python", "rust"):
try:
outs.append((toj(invoke(core)), None))
except Exception as e:
outs.append((None, repr(e)))
(py, py_err), (rs, rs_err) = outs
if (py_err is None) != (rs_err is None):
return False, f"py_err={py_err} rs_err={rs_err}"
if py_err is not None:
return None, f"both raised: {py_err}"
return py == rs, f"py={py!r} rs={rs!r}"
def fuzz_cases(rng):
out = []
# constraints classification
for _ in range(120):
kw = {}
for k in KEYWORD_POOL + rng.sample(UNKNOWN_KEYS, rng.randrange(0, 3)):
if rng.random() < 0.5:
kw[k] = rng.choice(VALS)
out.append(("constraints", kw))
# constraints_get
for _ in range(60):
kw = {rng.choice(KEYS + UNKNOWN_KEYS): rng.choice(VALS)}
key = rng.choice(list(kw) + ["missing_key"])
out.append(("constraints_get", kw, key,
rng.choice([None, "dflt", 0, False, ["x"]])))
# decision_record normalization
for _ in range(80):
fields = dict(
selected_language=rng.choice(LANGS),
runtime=rng.choice(["", "tokio", None]),
framework=rng.choice(["", "svelte", None]),
libraries=rng.choice([None, [], ["b", "a"], ["z", "a", "m"]]),
build_system=rng.choice(["", "cargo"]),
alternatives=rng.choice([None, [], [{"language": "go"}]]),
decisive_constraints=rng.choice([None, [], ["z", "a"]]),
rationale=rng.choice(["", "measured 10x"]),
evidence_refs=rng.choice([None, [], ["e2", "e1"]]),
confidence=rng.choice([0.0, 0.4, 0.5, 0.7, 0.8, 0.85]),
user_approval_required=rng.choice([True, False]),
resolution=rng.choice(["", "KEEP_EXISTING", "CHANGE_MAJOR",
"INSUFFICIENT_EVIDENCE"]),
)
fields = {k: v for k, v in fields.items() if rng.random() < 0.85}
out.append(("decision_record", fields))
# major_change
for _ in range(60):
out.append(("major_change", {"existing_language": rng.choice(LANGS)},
rng.choice(LANGS)))
# resolve_toolchain — every branch reachable
for _ in range(250):
kw = {}
if rng.random() < 0.8:
kw["existing_language"] = rng.choice(LANGS)
for k in ("existing_runtime", "existing_framework"):
if rng.random() < 0.3:
kw[k] = rng.choice(["tokio", "node", ""])
for k in ("target_platform", "deployment", "performance_sensitive",
"latency_sensitive", "memory_constrained", "gui_native"):
if rng.random() < 0.3:
kw[k] = rng.choice(VALS)
rkw = {}
if rng.random() < 0.6:
rkw["requested_language"] = rng.choice(
[l for l in LANGS if l] + [None])
if rng.random() < 0.4:
rkw["evidence_refs"] = rng.choice([[], ["b", "a"], ["only"]])
if rng.random() < 0.4:
rkw["performance_evidence"] = rng.choice([
{"significant_benefit": True, "basis": "measured"},
{"significant_benefit": False},
{"significant_benefit": None},
{}, {"other": 1},
])
out.append(("resolve_toolchain", kw, rkw))
return out
def main() -> int:
from rune import toolchain as tc
from rune import tl_loader
if not tl_loader.available():
print("Rust core not built — run cargo build --release in src/rust/toolchain")
return 2
total = passed = agreed_errors = 0
failures: list[str] = []
def record(label, ok, detail=""):
nonlocal total, passed, agreed_errors
total += 1
if ok is None: # both sides raised — agreement on refusal
agreed_errors += 1
passed += 1
elif ok:
passed += 1
else:
failures.append(f"{label}: {detail}")
def resolve_invoke(rkw):
return lambda c: call_core(
c, "resolve_toolchain",
constraints_of(c, rkw[0]), **rkw[1]).to_dict()
# 1. frozen truth corpus, replayed live on BOTH cores
truth = json.loads((Path(__file__).resolve().parents[1] / "src" / "rust" /
"toolchain" / "src" / "frozen_tc_truth.json").read_text())
for fn, cases in truth["cases"].items():
for case in cases:
name = f"truth:{fn}:{case['name']}"
inp = case["input"]
if fn == "constraint_keys":
py = sorted(tc.CONSTRAINT_KEYS)
rs = tl_loader.constraint_keys()
record(name, py == rs, f"py={py} rs={rs}")
elif fn == "constraints":
record(name, *run_pair(
lambda c, i=inp: constraints_of(c, i["kwargs"]).to_dict()))
elif fn == "constraints_get":
record(name, *run_pair(
lambda c, i=inp: constraints_of(c, i["known"])
.get(i["key"], i["default"])))
elif fn == "decision_record":
record(name, *run_pair(
lambda c, i=inp: call_core(c, "DecisionRecord", **i).to_dict()))
elif fn == "major_change":
record(name, *run_pair(
lambda c, i=inp: call_core(
c, "_major_change", constraints_of(c, i["known"]),
i["language"])))
elif fn == "resolve_toolchain":
rkw = {k: v for k, v in inp.items()
if k in ("requested_language", "evidence_refs",
"performance_evidence")}
record(name, *run_pair(resolve_invoke((inp["known"], rkw))))
# 2. seeded fuzz (kernel core vs freshly-pinned oracle per case)
rng = random.Random(2479)
for case in fuzz_cases(rng):
fn = case[0]
name = f"fuzz:{fn}"
if fn == "constraints":
ok, d = run_pair(lambda c, i=case[1]: constraints_of(c, i).to_dict())
elif fn == "constraints_get":
_, kw, key, default = case
ok, d = run_pair(lambda c, k=kw, key=key, df=default:
constraints_of(c, k).get(key, df))
elif fn == "decision_record":
ok, d = run_pair(lambda c, i=case[1]:
call_core(c, "DecisionRecord", **i).to_dict())
elif fn == "major_change":
_, kw, language = case
ok, d = run_pair(lambda c, k=kw, l=language: call_core(
c, "_major_change", constraints_of(c, k), l))
else: # resolve_toolchain
_, kw, rkw = case
ok, d = run_pair(resolve_invoke((kw, rkw)))
record(name, ok, f"case={case} {d}")
print(f"parity {passed}/{total} ({agreed_errors} agreed-error cases)")
if failures:
for f in failures[:12]:
print("FAIL", f[:400])
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
+186
View File
@@ -0,0 +1,186 @@
#!/usr/bin/env python3
"""Freeze toolchain oracle truth for the Rust port battery (dichotomy
wave #2479 tranche, batch: toolchain — RUST side verdict).
Exercises EVERY public function across EVERY branch:
ToolchainConstraints (known/unknown key classification, get with and
without default, to_dict), DecisionRecord (list normalization: sorting
of libraries/decisive_constraints/evidence_refs, None-list defaults,
alternatives copy, confidence), CONSTRAINT_KEYS (frozen vocabulary),
_major_change (existing-language equality branches), resolve_toolchain
(INSUFFICIENT_EVIDENCE short path, KEEP_EXISTING with/without hard
factors and performance evidence, same-language request, CHANGE_MAJOR
with and without significant evidence, greenfield recommendation,
greenfield INSUFFICIENT_EVIDENCE, falsy existing_language edges).
Output: src/rust/toolchain/src/frozen_tc_truth.json — replayed VERBATIM
by src/rust/toolchain/src/branch_battery.rs.
The module is fully deterministic (no wall-clock fields) — every case
compares exactly, no masking.
"""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
os.environ["RUNE_TC_CORE"] = "python" # POISONED-TRUTH LAW: freeze the ORACLE, never the kernel
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from rune import toolchain as tc # noqa: E402
cases: dict[str, list] = {}
def run(fn: str, name: str, *args, **kwargs):
out = getattr(tc, fn)(**kwargs) if args == () else getattr(tc, fn)(*args, **kwargs)
if hasattr(out, "to_dict"):
out = out.to_dict()
cases.setdefault(fn, []).append(
{"name": name, "input": kwargs if args == () else {"args": args, "kwargs": kwargs},
"expected": out})
# ── CONSTRAINT_KEYS: the frozen vocabulary ────────────────────────────
cases["constraint_keys"] = [{"name": "vocabulary",
"input": {},
"expected": sorted(tc.CONSTRAINT_KEYS)}]
# ── ToolchainConstraints: classification, get, to_dict ────────────────
for nm, kw in (
("known-only", dict(existing_language="python", target_platform="linux")),
("unknown-only", dict(wibble="future-thing", flurb=42)),
("mixed", dict(existing_language="rust", existing_runtime="tokio",
not_a_key=[1, 2], performance_sensitive=True, none_val=None)),
("empty", {}),
("every-known-key", {k: f"v-{k}" for k in tc.CONSTRAINT_KEYS}),
("falsy-values", dict(existing_language="", performance_sensitive=False,
memory_constrained=0)),
):
c = tc.ToolchainConstraints(**kw)
cases.setdefault("constraints", []).append(
{"name": nm, "input": {"kwargs": kw}, "expected": c.to_dict()})
for nm, kw, key, default, has_default in (
("get-known", dict(existing_language="python"), "existing_language", None, False),
("get-missing-no-default", dict(existing_language="python"), "existing_runtime", None, False),
("get-missing-with-default", dict(existing_language="python"), "existing_runtime", "", True),
("get-unknown-key-stored", dict(wibble="x"), "wibble", None, False),
("get-known-falsy-value", dict(existing_language=""), "existing_language", "dflt", True),
("get-default-none-explicit", dict(a="1"), "b", None, True),
):
c = tc.ToolchainConstraints(**kw)
val = c.get(key, default) if has_default else c.get(key)
cases.setdefault("constraints_get", []).append(
{"name": nm, "input": {"known": c.known, "key": key,
"default": default if has_default else None},
"expected": val})
# ── DecisionRecord: normalization branches ────────────────────────────
for nm, kw in (
("full-fields", dict(
selected_language="rust", runtime="tokio", framework="axum",
libraries=["zlib", "openssl", "a-lib"], build_system="cargo",
alternatives=[{"language": "go", "rejected_because": "no evidence"}],
decisive_constraints=["requested_language", "existing_language"],
rationale="measured benefit", evidence_refs=["bench-2", "bench-1"],
confidence=0.8, user_approval_required=True, resolution="CHANGE_MAJOR")),
("defaults", dict(selected_language="python")),
("none-lists-become-sorted-empty", dict(
selected_language="python", libraries=None, decisive_constraints=None,
evidence_refs=None, alternatives=None)),
("unsorted-normalized", dict(
selected_language="python", libraries=["b", "a", "c"],
decisive_constraints=["z", "a"], evidence_refs=["e2", "e1", "e0"],
confidence=0.85)),
("empty-lists-kept", dict(
selected_language="python", libraries=[], alternatives=[],
decisive_constraints=[], evidence_refs=[], resolution="KEEP_EXISTING")),
):
d = tc.DecisionRecord(**kw)
cases.setdefault("decision_record", []).append(
{"name": nm, "input": kw, "expected": d.to_dict()})
# ── _major_change ─────────────────────────────────────────────────────
for nm, kw, language in (
("different-language", dict(existing_language="python"), "rust"),
("same-language", dict(existing_language="python"), "python"),
("no-existing", {}, "rust"),
("falsy-existing", dict(existing_language=""), "rust"),
("non-str-existing", dict(existing_language=None), "rust"),
):
c = tc.ToolchainConstraints(**kw)
cases.setdefault("major_change", []).append(
{"name": nm, "input": {"known": c.known, "language": language},
"expected": tc._major_change(c, language)})
# ── resolve_toolchain: every branch ───────────────────────────────────
def res(nm, ckw, **kw):
c = tc.ToolchainConstraints(**ckw)
d = tc.resolve_toolchain(c, **kw)
cases.setdefault("resolve_toolchain", []).append(
{"name": nm,
"input": {"known": c.known, "unknown": c.unknown, **kw},
"expected": d.to_dict()})
# INSUFFICIENT short path: no existing, no request, <2 hard factors
res("insufficient-no-factors", dict(existing_runtime="tokio"))
res("insufficient-one-factor", dict(target_platform="linux"))
# falls through to greenfield INSUFFICIENT: >=2 hard factors, approval required
res("greenfield-insufficient-two-factors", dict(target_platform="linux", deployment="server"))
res("greenfield-insufficient-bools-count", dict(performance_sensitive=True, memory_constrained=True))
# KEEP_EXISTING branches
res("keep-existing-no-hard-factors", dict(existing_language="python"))
res("keep-existing-with-hard-factors", dict(existing_language="rust",
performance_sensitive=True))
res("keep-existing-with-runtime-framework", dict(
existing_language="typescript", existing_runtime="node",
existing_framework="svelte", target_platform="web"))
res("keep-existing-performance-evidence-rejected", dict(
existing_language="python", performance_sensitive=True),
performance_evidence={"significant_benefit": True, "basis": "measured 10x"})
res("keep-existing-empty-perf-evidence", dict(existing_language="python"),
performance_evidence={})
res("keep-existing-same-language-request", dict(existing_language="python"),
requested_language="python")
res("keep-existing-with-evidence-refs", dict(existing_language="go"),
evidence_refs=["ref-b", "ref-a", "ref-c"])
# CHANGE_MAJOR branches
res("change-major-no-evidence", dict(existing_language="python"),
requested_language="rust")
res("change-major-significant", dict(existing_language="python",
performance_sensitive=True),
requested_language="rust",
performance_evidence={"significant_benefit": True, "basis": "measured"})
res("change-major-falsy-significant", dict(existing_language="python"),
requested_language="rust",
performance_evidence={"significant_benefit": False})
res("change-major-null-significant", dict(existing_language="python"),
requested_language="zig", performance_evidence={"significant_benefit": None})
# greenfield recommendation
res("greenfield-requested", dict(target_platform="linux", deployment="server"),
requested_language="zig")
res("greenfield-requested-no-factors", dict(), requested_language="go")
# falsy existing_language edges: "" behaves as absent
res("falsy-existing-no-request", dict(existing_language=""))
res("falsy-existing-with-request", dict(existing_language=""),
requested_language="rust")
# unknown keys ride along but do not affect the decision
res("unknown-keys-ignored", dict(existing_language="python", wibble="x"),
requested_language="python")
truth = {
"frozen": "toolchain full oracle truth (dichotomy wave #2479)",
"oracle": "src/rune/toolchain.py",
"note": ("The module is fully deterministic — no wall-clock fields, "
"every case compares exactly. error = the exact Python "
"exception message where applicable."),
"cases": cases,
}
out = Path(__file__).resolve().parents[1] / "src" / "rust" / "toolchain" / "src" / "frozen_tc_truth.json"
out.write_text(json.dumps(truth, indent=1, sort_keys=True) + "\n")
total = sum(len(v) for v in cases.values())
print(f"frozen {total} cases across {len(cases)} functions -> {out}")
+118
View File
@@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""ctypes loader for the Rust toolchain core (dichotomy wave #2479
tranche). Mirrors broker_loader.py's contract:
FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
* tl_constraint_keys({}) -> sorted constraint-key vocabulary
* tl_constraints({kwargs}) -> {"known", "unknown"}
* tl_constraints_get({known, key, default?}) -> value
* tl_decision_record({...fields}) -> normalized record dict
* tl_resolve_toolchain({known, requested_language?, evidence_refs?,
performance_evidence?}) -> DecisionRecord dict
* tl_major_change({known, language}) -> bool
Free: capability_scoring's cs_free_string (shared process). Override the
.so with TC_RUST_LIB (a FILE path).
"""
from __future__ import annotations
import ctypes
import json
import os
from pathlib import Path
_REPO = Path(__file__).resolve().parents[2]
_CRATE_DIR = _REPO / "src" / "rust" / "toolchain"
_LIB_PATH = Path(
os.environ.get("TC_RUST_LIB", _CRATE_DIR / "target" / "release" / "libtoolchain.so")
)
_SYMBOLS = (
"tl_constraint_keys", "tl_constraints", "tl_constraints_get",
"tl_decision_record", "tl_resolve_toolchain", "tl_major_change",
)
_lib = None
class ToolchainKernelUnavailable(FileNotFoundError):
"""The Rust toolchain core is not built. Callers fall back to Python."""
def available() -> bool:
return _LIB_PATH.is_file()
def _load() -> ctypes.CDLL:
global _lib
if _lib is not None:
return _lib
if not _LIB_PATH.is_file():
raise ToolchainKernelUnavailable(str(_LIB_PATH))
lib = ctypes.CDLL(str(_LIB_PATH))
for sym in _SYMBOLS:
fn = getattr(lib, sym)
fn.argtypes = [ctypes.c_void_p]
fn.restype = ctypes.c_void_p
# free comes from capability_scoring (same process) if loaded; else our own malloc path
try:
lib.cs_free_string
except AttributeError:
scoring = ctypes.CDLL(str(
_REPO / "src" / "rust" / "capability_scoring" / "target" / "release" / "libcapability_scoring.so"
))
scoring.cs_free_string.argtypes = [ctypes.c_void_p]
lib._free = scoring.cs_free_string
_lib = lib
return lib
def _call(sym: str, payload: dict):
lib = _load()
data = json.dumps(payload).encode()
buf = ctypes.create_string_buffer(data)
ptr = getattr(lib, sym)(ctypes.cast(buf, ctypes.c_void_p))
if not ptr:
raise RuntimeError(f"{sym} returned null")
out = ctypes.string_at(ptr).decode()
free = getattr(lib, "_free", None)
if free is not None:
free(ptr)
doc = json.loads(out)
if not doc.get("ok"):
raise RuntimeError(doc.get("err", f"{sym} failed"))
return doc["value"]
def constraint_keys() -> list:
"""Sorted constraint-key vocabulary (the oracle's set, ordered)."""
return _call("tl_constraint_keys", {})
def constraints(kwargs: dict) -> dict:
"""Classify kwargs into {"known": ..., "unknown": ...}."""
return _call("tl_constraints", {"kwargs": kwargs})
def constraints_get(known: dict, key: str, default=None):
"""known.get(key, default) semantics (None default when absent)."""
return _call("tl_constraints_get", {"known": known, "key": key,
"default": default})
def decision_record(**fields) -> dict:
"""Normalized DecisionRecord dict (lists sorted, None → [])."""
return _call("tl_decision_record", fields)
def resolve_toolchain(known: dict, requested_language: str | None = None,
evidence_refs: list | None = None,
performance_evidence: dict | None = None) -> dict:
return _call("tl_resolve_toolchain", {
"known": known, "requested_language": requested_language,
"evidence_refs": list(evidence_refs or []),
"performance_evidence": performance_evidence})
def major_change(known: dict, language: str) -> bool:
return _call("tl_major_change", {"known": known, "language": language})
+59
View File
@@ -17,8 +17,32 @@ never invented certainty.
from __future__ import annotations
import os
__all__ = ["ToolchainConstraints", "resolve_toolchain", "DecisionRecord", "CONSTRAINT_KEYS"]
def _rust(fn: str, *args):
"""Delegate to the Rust kernel (dichotomy wave #2479); returns
_MISSING when the core is unavailable so the Python oracle body runs.
RUNE_TC_CORE=python pins the oracle unconditionally."""
if os.environ.get("RUNE_TC_CORE", "auto") == "python":
return _MISSING
try:
from rune import tl_loader
if not tl_loader.available():
return _MISSING
return getattr(tl_loader, fn)(*args)
except Exception:
return _MISSING
class _Missing:
pass
_MISSING = _Missing()
# Small fixed set of constraint KEYS the resolver can weigh. Values are
# open-world. Nothing here names a language.
CONSTRAINT_KEYS = {
@@ -46,6 +70,11 @@ class ToolchainConstraints:
carried verbatim (future-proof without schema churn)."""
def __init__(self, **kw):
out = _rust("constraints", kw)
if out is not _MISSING:
self.known = out["known"]
self.unknown = out["unknown"]
return
self.known = {}
self.unknown = {}
for k, v in kw.items():
@@ -55,6 +84,9 @@ class ToolchainConstraints:
self.unknown[k] = v
def get(self, key, default=None):
out = _rust("constraints_get", self.known, key, default)
if out is not _MISSING:
return out
return self.known.get(key, default)
def to_dict(self) -> dict:
@@ -80,6 +112,26 @@ class DecisionRecord:
user_approval_required: bool = False,
resolution: str = "",
):
out = _rust(
"decision_record",
selected_language, runtime, framework, libraries, build_system,
alternatives, decisive_constraints, rationale, evidence_refs,
confidence, user_approval_required, resolution,
)
if out is not _MISSING:
self.selected_language = out["selected_language"]
self.runtime = out["runtime"]
self.framework = out["framework"]
self.libraries = out["libraries"]
self.build_system = out["build_system"]
self.alternatives = out["alternatives"]
self.decisive_constraints = out["decisive_constraints"]
self.rationale = out["rationale"]
self.evidence_refs = out["evidence_refs"]
self.confidence = out["confidence"]
self.user_approval_required = out["user_approval_required"]
self.resolution = out["resolution"]
return
self.selected_language = selected_language
self.runtime = runtime
self.framework = framework
@@ -113,6 +165,9 @@ class DecisionRecord:
def _major_change(constraints: ToolchainConstraints, language: str) -> bool:
"""A major toolchain change = introducing/changing a language, runtime or
framework the project does not already use (tranche brief §10)."""
out = _rust("major_change", constraints.known, language)
if out is not _MISSING:
return out
existing = constraints.get("existing_language")
return bool(existing) and language != existing
@@ -137,6 +192,10 @@ def resolve_toolchain(
already uses it or the user asked.
"""
refs = sorted(evidence_refs or [])
out = _rust("resolve_toolchain", constraints.known, requested_language,
evidence_refs, performance_evidence)
if out is not _MISSING:
return DecisionRecord(**out)
existing = constraints.get("existing_language")
# INSUFFICIENT EVIDENCE path: no existing toolchain recorded AND no
+102
View File
@@ -0,0 +1,102 @@
//! toolchain branch battery — parity vs the frozen Python-oracle truth
//! (frozen_tc_truth.json, 43 cases across every public function and
//! branch, frozen 2026-10-11). Embedded VERBATIM via include_str!.
//!
//! The module is fully deterministic (no wall-clock fields) — every
//! case compares exactly, no masking.
use serde_json::Value;
use super::*;
const TRUTH: &str = include_str!("frozen_tc_truth.json");
fn cases(fn_name: &str) -> Vec<Value> {
let truth: Value = serde_json::from_str(TRUTH).unwrap();
truth["cases"][fn_name]
.as_array()
.unwrap()
.clone()
}
fn check(name: &str, actual: Value, case: &Value) {
assert!(
&actual == case.get("expected").expect("case must carry expected"),
"case {name} diverged:\n actual: {actual}\n expected: {}",
case["expected"]
);
}
#[test]
fn constraint_keys_parity_battery() {
for case in cases("constraint_keys") {
let mut keys: Vec<&str> = CONSTRAINT_KEYS.to_vec();
keys.sort_unstable();
check(case["name"].as_str().unwrap(), json!(keys), &case);
}
}
#[test]
fn constraints_parity_battery() {
for case in cases("constraints") {
let a = &case["input"];
let kwargs = a["kwargs"].as_object().unwrap();
check(case["name"].as_str().unwrap(), constraints_new(kwargs), &case);
}
}
#[test]
fn constraints_get_parity_battery() {
for case in cases("constraints_get") {
let a = &case["input"];
let known = a["known"].as_object().unwrap();
let out = constraints_get(known, a["key"].as_str().unwrap(), a.get("default"));
check(case["name"].as_str().unwrap(), out, &case);
}
}
#[test]
fn decision_record_parity_battery() {
for case in cases("decision_record") {
let a = &case["input"];
let out = decision_record(
a["selected_language"].clone(),
a.get("runtime"),
a.get("framework"),
a.get("libraries"),
a.get("build_system").and_then(Value::as_str).unwrap_or(""),
a.get("alternatives"),
a.get("decisive_constraints"),
a.get("rationale").and_then(Value::as_str).unwrap_or(""),
a.get("evidence_refs"),
a.get("confidence").and_then(Value::as_f64).unwrap_or(0.0),
a.get("user_approval_required").and_then(Value::as_bool).unwrap_or(false),
a.get("resolution").and_then(Value::as_str).unwrap_or(""),
);
check(case["name"].as_str().unwrap(), out, &case);
}
}
#[test]
fn major_change_parity_battery() {
for case in cases("major_change") {
let a = &case["input"];
let out = major_change(a["known"].as_object().unwrap(), a["language"].as_str().unwrap());
check(case["name"].as_str().unwrap(), Value::Bool(out), &case);
}
}
#[test]
fn resolve_toolchain_parity_battery() {
for case in cases("resolve_toolchain") {
let a = &case["input"];
let known = a["known"].as_object().unwrap();
let out = resolve_toolchain(
known,
a.get("requested_language").and_then(Value::as_str),
a.get("evidence_refs"),
a.get("performance_evidence").filter(|p| !p.is_null()),
);
check(case["name"].as_str().unwrap(), out, &case);
}
}
File diff suppressed because it is too large. Load diff
+397 -44
View File
@@ -1,51 +1,404 @@
//! toolchain — toolchain detection and validation (W6).
pub fn detect_toolchain(files: &[String]) -> Vec<String> {
let mut toolchains = Vec::new();
for f in files {
match f {
f if f.ends_with(".rs") => {
if !toolchains.contains(&"rust".to_string()) {
toolchains.push("rust".to_string());
}
}
f if f.ends_with(".go") => {
if !toolchains.contains(&"go".to_string()) {
toolchains.push("go".to_string());
}
}
f if f.ends_with(".py") => {
if !toolchains.contains(&"python".to_string()) {
toolchains.push("python".to_string());
}
}
f if f.ends_with("Cargo.toml") => {
if !toolchains.contains(&"cargo".to_string()) {
toolchains.push("cargo".to_string());
}
}
f if f.ends_with("go.mod") => {
if !toolchains.contains(&"go-mod".to_string()) {
toolchains.push("go-mod".to_string());
}
}
f if f.ends_with("package.json") && !toolchains.contains(&"node".to_string()) => {
toolchains.push("node".to_string());
}
_ => {}
//! toolchain — evidence-backed Toolchain Resolution + decision records
//! (SCAFFOLD-TRANCHE-003).
//!
//! FULL parity port of src/rune/toolchain.py (dichotomy wave, #2479
//! tranche, RUST side verdict). First real resolver: infer constraints,
//! inspect the existing project toolchain, produce an inspectable
//! decision record.
//!
//! Strong rule (tranche brief §8, §10): EXISTING PROJECT TOOLCHAIN WINS
//! unless recorded evidence shows a significant benefit to changing it.
//! A major language/runtime/framework change ALWAYS requires user
//! approval. Constraint KEYS are a small fixed vocabulary; constraint
//! VALUES are open-world data — no language ontology lives here, and
//! nothing is Python-default.
//!
//! Honesty rule: constraints that are unknown produce
//! INSUFFICIENT_EVIDENCE, never invented certainty.
//!
//! FFI (doc-envelope law): every `tl_*` symbol takes one JSON arg and
//! returns {"ok": true, "value": …} or {"ok": false, "err": msg}.
//! NOTE: no tl_free_string export — sibling crates sharing the Python
//! process already export identical free symbols; a duplicate breaks the
//! cdylib link. Callers free through capability_scoring's cs_free_string.
use serde_json::{json, Map, Value};
use std::ffi::{CStr, CString};
use std::os::raw::c_char;
/// Small fixed set of constraint KEYS the resolver can weigh. Values are
/// open-world. Nothing here names a language.
pub const CONSTRAINT_KEYS: &[&str] = &[
"existing_language", // str: language the project already uses
"existing_runtime", // str
"existing_framework", // str
"target_platform", // str: linux/macos/windows/web/embedded/...
"deployment", // str: server/desktop/cli/container/embedded
"performance_sensitive", // bool
"latency_sensitive", // bool
"memory_constrained", // bool
"concurrency_model", // str: async/threads/processes/none
"interoperability", // str: must embed/call into what
"gui_native", // bool
"expected_lifespan", // str: short/long
"user_constraints", // list[str]: explicit user demands
"ecosystem_requirement", // str: needed library/domain support
];
/// Keys that drive the hard-factor count in the INSUFFICIENT_EVIDENCE
/// gate (oracle's fixed tuple order; the record sorts them anyway).
const HARD_FACTOR_KEYS: &[&str] = &[
"performance_sensitive",
"latency_sensitive",
"memory_constrained",
"gui_native",
"target_platform",
"deployment",
];
/// Python truthiness for JSON values (0, "", [], {}, null, False are falsy).
fn is_truthy(v: Option<&Value>) -> bool {
match v {
None | Some(Value::Null) => false,
Some(Value::Bool(b)) => *b,
Some(Value::Number(n)) => n.as_f64().map(|f| f != 0.0).unwrap_or(true),
Some(Value::String(s)) => !s.is_empty(),
Some(Value::Array(a)) => !a.is_empty(),
Some(Value::Object(o)) => !o.is_empty(),
}
}
/// The oracle's ToolchainConstraints constructor: split kwargs into the
/// known typed vocabulary and the open-world unknown carry-through.
pub fn constraints_new(kwargs: &Map<String, Value>) -> Value {
let mut known = Map::new();
let mut unknown = Map::new();
for (k, v) in kwargs {
if CONSTRAINT_KEYS.contains(&k.as_str()) {
known.insert(k.clone(), v.clone());
} else {
unknown.insert(k.clone(), v.clone());
}
}
toolchains
json!({"known": known, "unknown": unknown})
}
#[cfg(test)]
mod tests {
use super::*;
/// The oracle's ToolchainConstraints.get: known keys only, default None.
pub fn constraints_get(known: &Map<String, Value>, key: &str, default: Option<&Value>) -> Value {
known.get(key).cloned().unwrap_or_else(|| default.cloned().unwrap_or(Value::Null))
}
#[test]
fn test_detect() {
let tc = detect_toolchain(&["main.rs".into(), "main.go".into(), "Cargo.toml".into()]);
assert!(tc.contains(&"rust".to_string()));
assert!(tc.contains(&"go".to_string()));
assert!(tc.contains(&"cargo".to_string()));
/// Python `sorted(list_of_str)` on an optional list (None → empty).
fn sorted_strs(v: Option<&Value>) -> Vec<String> {
let mut out: Vec<String> = v
.and_then(Value::as_array)
.map(|a| a.iter().filter_map(Value::as_str).map(str::to_string).collect())
.unwrap_or_default();
out.sort();
out
}
/// The oracle's DecisionRecord constructor: durable, inspectable
/// toolchain decision (tranche brief §9). List inputs are normalized
/// (sorted, None → empty); alternatives are carried verbatim.
#[allow(clippy::too_many_arguments)]
pub fn decision_record(
selected_language: Value,
runtime: Option<&Value>,
framework: Option<&Value>,
libraries: Option<&Value>,
build_system: &str,
alternatives: Option<&Value>,
decisive_constraints: Option<&Value>,
rationale: &str,
evidence_refs: Option<&Value>,
confidence: f64,
user_approval_required: bool,
resolution: &str,
) -> Value {
json!({
"selected_language": selected_language,
"runtime": runtime.cloned().unwrap_or(json!("")),
"framework": framework.cloned().unwrap_or(json!("")),
"libraries": sorted_strs(libraries),
"build_system": build_system,
"alternatives": alternatives
.filter(|v| !v.is_null()) // Python: list(alternatives or [])
.cloned()
.unwrap_or(json!([])),
"decisive_constraints": sorted_strs(decisive_constraints),
"rationale": rationale,
"evidence_refs": sorted_strs(evidence_refs),
"confidence": confidence,
"user_approval_required": user_approval_required,
"resolution": resolution,
})
}
/// A major toolchain change = introducing/changing a language, runtime or
/// framework the project does not already use (tranche brief §10).
pub fn major_change(known: &Map<String, Value>, language: &str) -> bool {
let existing = known.get("existing_language");
is_truthy(existing) && existing != Some(&json!(language))
}
/// Produce an evidence-backed toolchain decision.
///
/// Rules:
/// - existing project language WINS unless significant recorded evidence
/// (performance_evidence with measured basis) shows a substantial benefit
/// from a different language AND the user approves.
/// - requested_language differing from the project's language is a MAJOR
/// change → user_approval_required=True (never silent).
/// - unknown critical constraints → INSUFFICIENT_EVIDENCE, not guesses.
/// - Python is NOT a default: it is selected only because the project
/// already uses it or the user asked.
pub fn resolve_toolchain(
known: &Map<String, Value>,
requested_language: Option<&str>,
evidence_refs: Option<&Value>,
performance_evidence: Option<&Value>,
) -> Value {
let refs = sorted_strs(evidence_refs);
let existing = known.get("existing_language");
let requested = requested_language.map(|r| Value::String(r.to_string()));
// INSUFFICIENT EVIDENCE path: no existing toolchain recorded AND no
// sufficient constraints to justify any selection.
let hard_factors: Vec<&str> = HARD_FACTOR_KEYS
.iter()
.copied()
.filter(|k| known.contains_key(*k))
.collect();
let requested_truthy = requested.as_ref().map(|r| is_truthy(Some(r))).unwrap_or(false);
if !is_truthy(existing) && !requested_truthy && hard_factors.len() < 2 {
return decision_record(
json!(""),
None,
None,
None,
"",
None,
Some(&json!(hard_factors)),
"no existing project toolchain recorded and too few \
constraints known to select honestly; refusing to \
invent certainty",
Some(&json!(refs)),
0.0,
false,
"INSUFFICIENT_EVIDENCE",
);
}
// DEFAULT: existing toolchain wins
let same_request = requested
.as_ref()
.map(|r| Some(r) == existing)
.unwrap_or(false);
if is_truthy(existing) && (requested.is_none() || same_request) {
let mut decisive = vec![json!("existing_language")];
if is_truthy(performance_evidence) {
decisive.push(json!(
"performance_evidence(rejected: does not override existing toolchain without user approval)"
));
}
return decision_record(
existing.cloned().unwrap_or(json!("")),
known.get("existing_runtime"),
known.get("existing_framework"),
None,
known
.get("existing_build_system")
.and_then(Value::as_str)
.unwrap_or(""),
None,
Some(&json!(decisive)),
"existing project toolchain wins (§8); no recorded evidence shows significant benefit from change",
Some(&json!(refs)),
if !hard_factors.is_empty() { 0.85 } else { 0.7 },
false,
"KEEP_EXISTING",
);
}
// REQUESTED change away from existing: recommendable, never silent
if is_truthy(existing) && requested_truthy && !same_request {
let significant = performance_evidence
.and_then(|p| p.get("significant_benefit"))
.map(|s| is_truthy(Some(s)))
.unwrap_or(false);
let mut decisive = vec![json!("existing_language"), json!("requested_language")];
if significant {
decisive.push(json!("performance_evidence"));
}
return decision_record(
requested.clone().unwrap_or(json!("")),
known.get("existing_runtime"),
None,
None,
"",
Some(&json!([{"language": existing, "rejected_because": "requested change"}])),
Some(&json!(decisive)),
"requested language differs from the project's \
established toolchain; recommendation only — \
user approval required before material change",
Some(&json!(refs)),
if significant { 0.8 } else { 0.5 },
true,
"CHANGE_MAJOR",
);
}
// Greenfield: we have NO language-selection intelligence (tranche brief
// §28: do not claim general language-selection from fixtures). Without a
// requested language, the honest verdict is INSUFFICIENT_EVIDENCE —
// never a guess, never silently Python. With a request, record it as a
// recommendation requiring approval.
if requested_truthy {
return decision_record(
requested.clone().unwrap_or(json!("")),
None,
None,
None,
"",
None,
Some(&json!(hard_factors)),
"greenfield request recorded as recommendation; user approval required",
Some(&json!(refs)),
0.4,
true,
"CHANGE_MAJOR",
);
}
decision_record(
json!(""),
None,
None,
None,
"",
None,
Some(&json!(hard_factors)),
"greenfield: no requested language and no language-selection evidence; refusing to invent a choice",
Some(&json!(refs)),
0.0,
true,
"INSUFFICIENT_EVIDENCE",
)
}
// ── FFI (doc-envelope law) ──────────────────────────────────────────────
fn read_arg(ptr: *const c_char) -> Result<Value, String> {
if ptr.is_null() {
return Err("null argument pointer".into());
}
let x = unsafe { CStr::from_ptr(ptr) }
.to_str()
.map_err(|e| e.to_string())?;
serde_json::from_str(x).map_err(|e| e.to_string())
}
fn envelope(r: Result<Value, String>) -> *mut c_char {
let v = match r {
Ok(x) => json!({"ok": true, "value": x}),
Err(e) => json!({"ok": false, "err": e}),
};
CString::new(serde_json::to_string(&v).unwrap_or_default())
.map(|c| c.into_raw())
.unwrap_or(std::ptr::null_mut())
}
fn arg_obj<'a>(a: &'a Value, k: &str) -> Result<&'a Map<String, Value>, String> {
a.get(k)
.and_then(Value::as_object)
.ok_or_else(|| format!("missing {k}"))
}
/// tl_constraint_keys({}) → the frozen constraint-key vocabulary (sorted;
/// the oracle's set is unordered, comparisons use sorted order).
#[no_mangle]
pub extern "C" fn tl_constraint_keys(_arg: *const c_char) -> *mut c_char {
let mut keys: Vec<&str> = CONSTRAINT_KEYS.to_vec();
keys.sort_unstable();
envelope(Ok(json!(keys)))
}
/// tl_constraints({"kwargs": {...}}) → {"known": {...}, "unknown": {...}}.
#[no_mangle]
pub extern "C" fn tl_constraints(arg: *const c_char) -> *mut c_char {
envelope(read_arg(arg).and_then(|a| {
let kwargs = arg_obj(&a, "kwargs")?;
Ok(constraints_new(kwargs))
}))
}
/// tl_constraints_get({"known": {...}, "key": k, "default": v?}) → value.
#[no_mangle]
pub extern "C" fn tl_constraints_get(arg: *const c_char) -> *mut c_char {
envelope(read_arg(arg).and_then(|a| {
let known = arg_obj(&a, "known")?;
let key = a.get("key").and_then(Value::as_str).unwrap_or("");
Ok(constraints_get(known, key, a.get("default")))
}))
}
/// tl_decision_record({selected_language, runtime?, framework?,
/// libraries?, build_system?, alternatives?, decisive_constraints?,
/// rationale?, evidence_refs?, confidence?, user_approval_required?,
/// resolution?}) → normalized record dict.
#[no_mangle]
pub extern "C" fn tl_decision_record(arg: *const c_char) -> *mut c_char {
envelope(read_arg(arg).and_then(|a| {
let selected = a
.get("selected_language")
.cloned()
.ok_or_else(|| "missing selected_language".to_string())?;
Ok(decision_record(
selected,
a.get("runtime"),
a.get("framework"),
a.get("libraries"),
a.get("build_system").and_then(Value::as_str).unwrap_or(""),
a.get("alternatives"),
a.get("decisive_constraints"),
a.get("rationale").and_then(Value::as_str).unwrap_or(""),
a.get("evidence_refs"),
a.get("confidence").and_then(Value::as_f64).unwrap_or(0.0),
a.get("user_approval_required").and_then(Value::as_bool).unwrap_or(false),
a.get("resolution").and_then(Value::as_str).unwrap_or(""),
))
}))
}
/// tl_resolve_toolchain({"known": {...}, "requested_language"?: str|null,
/// "evidence_refs"?: [str], "performance_evidence"?: obj|null}) →
/// DecisionRecord dict.
#[no_mangle]
pub extern "C" fn tl_resolve_toolchain(arg: *const c_char) -> *mut c_char {
envelope(read_arg(arg).and_then(|a| {
let known = arg_obj(&a, "known")?;
Ok(resolve_toolchain(
known,
a.get("requested_language").and_then(Value::as_str),
a.get("evidence_refs"),
a.get("performance_evidence").filter(|p| !p.is_null()),
))
}))
}
/// tl_major_change({"known": {...}, "language": str}) → bool.
#[no_mangle]
pub extern "C" fn tl_major_change(arg: *const c_char) -> *mut c_char {
envelope(read_arg(arg).and_then(|a| {
let known = arg_obj(&a, "known")?;
let language = a
.get("language")
.and_then(Value::as_str)
.ok_or_else(|| "missing language".to_string())?;
Ok(Value::Bool(major_change(known, language)))
}))
}
#[path = "branch_battery.rs"]
#[cfg(test)]
mod branch_battery;