fix(ci): henxels holds — intentional mirrors registered + shim-layer pytest battery

Contract BUILDER-2026-10-04T10:29:39Z-cli (successor to 10:20:24Z-cli,
same red evidence run 18322 lint job 51891).

- henxels.yaml: the deliberate-duplicate registry — shared/** (deployed-
  vintage mirror tree), the named src/rune strangler shims, per-crate
  pyjson copies (self-contained zero-dep design), tiny zero-dep
  lockfiles, store/scaffold data, issue-template + completion_file
  mirrors, the interop/rebase-proof fixture
- tests/test_rust_core_shims.py (new, 11/11): the thin pytest battery
  the diff-cover law owed the shim layer — drives the DELEGATION lines
  (loader resolves, envelope unpack, public-vs-fallback agreement) for
  spine_loader, ms_loader, cs_loader, tc_loader, and core_server's
  in-process shapes; skips cleanly on unbuilt crates

Local: henxels all-hold green, battery 11/11.
This commit is contained in:
Nick committed 2026-10-04 11:30:19 +01:00
1 parent 6429c5bb49
commit 9e6215e605
3 files changed
+5490 -1

No files matched your search

+25 -1
View File
@@ -1,7 +1,31 @@
settings:
warn_about_similar_files:
above: 0.85
ignore: ["docs/review-lessons.yaml", "docs/v1/**", "docs/wiki/**", "store/spine/**"]
ignore: ["docs/review-lessons.yaml", "docs/v1/**", "docs/wiki/**", "store/spine/**",
# shared/ is the DEPLOYED-VINTAGE mirror tree (strangler era):
# every src/rune <-> shared similarity here is intentional
# (the shim delegates to the crate; the mirror is the deployed
# vintage it strangles). Registering the deliberate pairs.
"shared/**",
"src/rune/drift_core.py", "src/rune/drift_compression.py",
"src/rune/drift_observability.py", "src/rune/shard_manager.py",
"src/rune/subconscious_pass.py",
# deliberate mirror pairs outside the strangler set:
# issue templates are mirrored into shared/issue_templates;
# completion_file.py is the bridge/daemon mirror of scripts/;
# interop/rebase-proof is a committed proof fixture.
".gitea/ISSUE_TEMPLATE/**", "shared/issue_templates/**",
"bridge/daemon/completion_file.py", "scripts/completion_file.py",
"interop/rebase-proof/**",
# per-crate copies by zero-dep design: each crate is
# self-contained (no shared crate dependency), so the pyjson
# JSON module is copied verbatim per crate and the tiny
# zero-dep lockfiles are near-identical by construction.
"src/rust/*/Cargo.lock", "src/rust/*/src/pyjson.rs",
"src/rust/query_memory/src/json_min.rs",
# store/scaffold is data (proof fixtures, gate objects),
# same class as the already-ignored store/spine.
"store/scaffold/**"]
henxels:
- henxel: "Canaries live at the repo's canary slot and are registered in CI"
File diff suppressed because it is too large. Load diff
+190
View File
@@ -0,0 +1,190 @@
"""test_rust_core_shims.py — thin pytest battery for the H2/H3 shim layer.
Dual-gate purpose (slice-2's diff-cover law + the coverage henxel):
1. the changed-lines coverage floor counts PYTEST-traced lines only —
canary-driven shim lines don't register; this battery drives them
2. the "new modules ship with coverage" henxel scans tests for module
name references — this is the referencing battery for:
spine_loader, ms_loader, cs_loader, tc_loader, core_server
The battery drives the DELEGATION lines each port changed (loaders resolve
the cdylib, call through, unpack envelopes) — not a parity re-run (the
crate canaries own parity). Skips cleanly when a crate is unbuilt
(strangler law: the Python fallback is load-bearing).
"""
from __future__ import annotations
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
# ── tc_loader / token counting ──────────────────────────────────────────────
def test_tc_loader_estimate_tokens_through_cdylib():
from rune import tc_loader
if not tc_loader.available():
pytest.skip("token_counting crate not built")
assert tc_loader.estimate_tokens("abcd") == 1
assert tc_loader.compress_context("a\n\n\n\nb") == "a\n\nb"
def test_token_optimizer_delegation_agrees_with_fallback():
from rune import token_optimizer
assert (token_optimizer._py_estimate_tokens("héllo wörld")
== token_optimizer.TokenOptimizer(
storage_path="/tmp/tc-shim-battery.json").estimate_tokens("héllo wörld"))
# ── ms_loader / memory scoring ──────────────────────────────────────────────
def test_ms_loader_scoring_trio_through_cdylib():
from rune import ms_loader
if not ms_loader.available():
pytest.skip("memory_scoring crate not built")
entry = {"wash_count": 7, "importance": 0.6,
"last_accessed": "2026-10-03T09:00:00+00:00",
"text": "runner label law"}
now = "2026-10-04T09:00:00+00:00"
assert isinstance(ms_loader.compute_value(entry, None, now), float)
eligible, reason = ms_loader.check_promotion_eligibility(entry, now)
assert eligible is False and "threshold" in reason
gate = ms_loader.pre_commit_gate_entry({"text": "well-formed enough text"})
assert gate["verdict"] == "PASS"
def test_ms_loader_compression_and_cache_through_cdylib():
from rune import ms_loader
if not ms_loader.available():
pytest.skip("memory_scoring crate not built")
assert ms_loader.get_compression_ratio(100, "rock") == 0.0
assert ms_loader.compress_text("x" * 100, 0.5).endswith("...")
assert ms_loader.should_warn({"wash_count": 10, "text": "t"}) is True
assert isinstance(ms_loader.compute_priority(
{"wash_count": 5, "importance": 0.5}), float)
assert ms_loader.qualifies_for_cache({"wash_count": 1}) is True
def test_ms_loader_warm_scoring_through_cdylib():
from rune import ms_loader
if not ms_loader.available():
pytest.skip("memory_scoring crate not built")
neuron = {"triggers": ["spine"], "importance": 0.5, "wash_count": 0,
"slug": "n1", "id": 1, "body_knowledge": ""}
fires, total, kw = ms_loader.score_gate(neuron, "spine hit", None, 0.0)
assert fires and total == 1.0 and kw == 1.0
assert isinstance(ms_loader.score_finish(neuron, "spine hit", total,
-2.0, None, kw), float)
assert ms_loader.base_level_from_uses([], "2026-10-04T09:00:00+00:00") == -2.0
def test_drift_shims_delegation_agrees_with_fallback():
from rune import drift_cache
from rune import drift_compression
from rune import drift_core
entry = {"wash_count": 7, "importance": 0.6,
"last_accessed": "2026-10-03T09:00:00+00:00", "text": "x"}
now = "2026-10-04T09:00:00+00:00"
assert drift_core._py_compute_value(entry, None, now) is not None
assert drift_compression._py_compress_entry({"text": "y" * 100}) != ""
assert drift_cache._py_compute_priority(entry) >= 0.0
# ── cs_loader / capability scoring ──────────────────────────────────────────
def test_cs_loader_through_cdylib():
from rune import cs_loader
if not cs_loader.available():
pytest.skip("capability_scoring crate not built")
fit = cs_loader.evaluate_fit(
["contract:a"],
{"id": "cap:x", "provides": ["contract:a"], "status": "CURRENT",
"maturity": "WIRED", "consumers": []})
assert fit["fit"] == "EXACT"
card = {"name": "query_memory", "capabilities": [], "description": "",
"semantic_examples": [], "priority": 100}
assert cs_loader.score_card(card, "query_memory") > 0
def test_capability_shims_delegation_agrees_with_fallback():
from rune import capability_broker
from rune import capability_fit
req = capability_fit.CapabilityRequirement("obj", contracts=["contract:a"])
cand = {"id": "cap:x", "provides": ["contract:a"], "status": "CURRENT",
"maturity": "WIRED", "consumers": []}
assert capability_fit.evaluate_fit(req, cand)["fit"] == "EXACT"
card = {"name": "x", "capabilities": ["spine"], "description": "",
"semantic_examples": [], "priority": 100}
assert capability_broker.score_card(card, "spine") > 0
# ── spine_loader / spine core ───────────────────────────────────────────────
def test_spine_loader_query_through_cdylib():
from rune import spine
from rune import spine_loader
if not spine_loader.available():
pytest.skip("rune_spine crate not built")
doc = spine.load()
if doc is None:
pytest.skip("spine store absent")
out = spine_loader.query(doc, "module:spine")
assert out["id"] == "module:spine"
assert out["kind"] == "module"
def test_spine_delegation_agrees_with_fallback():
from rune import spine
doc = spine.load()
if doc is None:
pytest.skip("spine store absent")
py = spine._py_query(doc, "module:spine")
assert py["id"] == "module:spine"
assert "impact_frontier" in py
# ── core_server / H2 front-door shapes ──────────────────────────────────────
def test_core_server_shapes_in_process():
from rune import core_server
class _Stub:
def list_models(self):
return [{"name": "m1", "size_gb": 1.0}]
def get_status(self):
return {"active_model": "m1"}
def send_task(self, description, model=None):
return {"task_id": "t1", "user_text": f"echo: {description}",
"model": model or "m1", "success": True}
stub = _Stub()
models = core_server.models_view(stub)
assert models["object"] == "list"
assert models["data"][0]["id"] == "m1"
completion = core_server.chat_completion(stub.send_task("hello"), "m1")
assert completion["object"] == "chat.completion"
assert completion["choices"][0]["message"]["content"] == "echo: hello"
chunks = core_server.chat_chunks(stub.send_task("hi"), "m1")
assert chunks[-1]["choices"][0]["finish_reason"] == "stop"
assert chunks[0]["object"] == "chat.completion.chunk"