conductor/rune
1
2 Fork 1
Code Issues 29 Pull Requests 3 Actions Packages Projects Releases 563 Wiki Activity

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

Browse Source 
Full parity port of src/rune/construction_pattern.py (dichotomy wave,
#2479 tranche): ConstructionPattern construct/revise,
record_pattern_proposal, evaluate_applicability,
derive_pattern_obligations, combine_pattern_obligations as
cp_* FFI entry points (doc-envelope law; freed via
capability_scoring's cs_free_string — no re-exported free symbol).

Frozen oracle truth: 56 cases (scripts/gen_construction_pattern_truth.py,
RUNE_CP_CORE=python pinned) exercising every public function and branch —
constructor validation errors, open-world ext, revision history
accumulation + 200-char reason truncation, exclusion override paths,
boundary-aware short-token vs containment matching, ':'-prefix entities,
objective '/'-split (dashes kept), authority relationship scans with the
falsy-from_id quirk, evidence truncation, EXCEPTED defaults,
accepted/rejected/unresolved routing, duplicate collapse + conflicts.

ctypes loader src/rune/cp_loader.py mirrors broker_loader.py; facade
delegation on the four pure functions with oracle fallback and
RUNE_CP_CORE=python pin. Live parity 56/56 vs oracle; battery 9/9;
clippy 0; construction surface tests 88/88 green with and without the
kernel live.
This commit is contained in:
Nick committed 2026-10-11 21:51:05 +01:00
1 parent d7a5a99bab
commit 681312c69f
8 files changed
+7766 -137
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/gen_construction_pattern_truth.py
+362
View File
@@ -0,0 +1,362 @@
#!/usr/bin/env python3
"""Freeze construction_pattern oracle truth for the Rust port battery
(dichotomy wave #2479 tranche, batch: construction_pattern).
Exercises EVERY public function and branch of
src/rune/construction_pattern.py:
ConstructionPattern.__init__ (defaults, sorting, ext, all validation errors)
ConstructionPattern.to_dict / from_dict (roundtrip incl. ext)
ConstructionPattern.revise (changes, history accumulation, 200-char
reason truncation, sorted evidence/changed)
record_pattern_proposal (bounds errors, evidence sorting, 0/1 bounds)
evaluate_applicability (exclusions by objective-substring and by
constraint membership, first-match-wins,
no-signals UNKNOWN, no-groups UNKNOWN,
surface_kinds via surface lists AND via
resolutions, short-token vs containment path
and entity matching, ':'-prefix entities,
objective_any '/' splitting (dashes NOT
split), contracts_match via authorities
(implements/exposes/provides/serves),
APPLIES / PARTIAL / DOES_NOT_APPLY,
evidence [:5] truncation)
derive_pattern_obligations (when None, always True vs truthy-not-True,
path_contains hit/miss + [:3] truncation,
entities_match via paths/ui/regions,
contracts_touched via need constraints and
via authority relationships, unrecognized
shapes, EXCEPTED with defaults,
accepted/rejected/unresolved routing)
combine_pattern_obligations (duplicate collapse + also_from, same-id
different-text conflicts, id sort, empty)
Each case records its INPUT and the oracle's EXPECTED output/error; replayed
VERBATIM by src/rust/construction_pattern/src/branch_battery.rs.
Output: src/rust/construction_pattern/src/frozen_cp_truth.json
"""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
os.environ["RUNE_CP_CORE"] = "python" # POISONED-TRUTH LAW: freeze the ORACLE
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from rune import construction_pattern as cp # noqa: E402
OUT = Path(__file__).resolve().parents[1] / "src" / "rust" / "construction_pattern" / "src" / "frozen_cp_truth.json"
cases = []
def need(objective="repair the deploy promotion ordering",
affected_surface=("scripts/deploy_stage.py", "scripts/deploy_promotion.py"),
constraints=None):
class N:
pass
n = N()
n.objective = objective
n.affected_surface = list(affected_surface)
n.constraints = dict(constraints if constraints is not None else {"contracts": ["contract:monotonic_promotion"]})
return n
def serialize_need(n):
return {"objective": n.objective, "affected_surface": n.affected_surface, "constraints": n.constraints}
SPINE = {
"entities": [
{"id": "module:deploy_stage", "kind": "module"},
{"id": "contract:monotonic_promotion", "kind": "contract"},
{"id": "module:deploy_qualify", "kind": "module"},
{"id": "contract:fail_closed_deployment", "kind": "contract"},
],
"relationships": [
{"type": "implements", "from": "module:deploy_stage", "to": "contract:monotonic_promotion"},
{"type": "exposes", "from": "module:deploy_stage", "to": "contract:exposed_api"},
{"type": "serves", "from": "module:deploy_stage", "to": "contract:served_edge"},
{"type": "implements", "from": "module:deploy_qualify", "to": "contract:fail_closed_deployment"},
{"type": "provides", "from": "module:deploy_qualify", "to": "contract:provided_gate"},
{"type": "serves", "from": "module:deploy_qualify", "to": "contract:second_edge"},
{"type": "depends_on", "from": "module:deploy_stage", "to": "module:deploy_qualify"},
],
}
AUTH = {"rune_spine": SPINE}
SURFACE = {
"affected_paths": [
"scripts/deploy_stage.py", "scripts/deploy_promotion.py",
"src/rune/rune_tui.py", "src/rune/builtin.py",
],
"affected_existing_entities": ["module:deploy_stage", "module:deploy_qualify", "ui:chat"],
"affected_ui_surfaces": ["ui:version_display"],
"affected_regions": ["region:left"],
"workflow_config_surfaces": [".gitea/workflows/deploy.yml"],
"resolutions": [
{"id": "module:deploy_stage", "kind": "source_path", "verdict": "FOUND"},
{"id": "ui:version_display", "kind": "ui_component", "verdict": "FOUND"},
{"id": "region:left", "kind": "layout_region", "verdict": "FOUND"},
{"id": "cfg:deploy", "kind": "workflow_config", "verdict": "FOUND"},
{"id": "module:deploy_stage", "kind": "structural_entity", "verdict": "FOUND"},
],
}
def case(name, op, inp, fn):
import copy as _copy
# Deep-copy the input BEFORE running the oracle: to_dict() shares
# nested references (ext) with the live object, and revise() mutates
# revision_history in place — without the copy the recorded input
# aliases oracle-mutated state.
inp = _copy.deepcopy(inp)
try:
expected = fn()
except ValueError as e:
expected = {"error": str(e)}
cases.append({"name": name, "op": op, "input": inp, "expected": expected})
# ── ConstructionPattern.__init__ / to_dict / from_dict ──────────────────────
FULL_KW = dict(
revision=3, maturity="EXERCISED",
signals={"path_contains": ["deploy"], "objective_any": ["promotion"]},
obligations=[
{"id": "o2", "obligation": "second duty", "when": {"always": True}},
{"id": "o1", "obligation": "first duty", "when": {"path_contains": ["deploy"]}},
],
provenance=["incident:zeta", "incident:alpha"],
exclusions=["queue-only-project", "no-deploy-zone"],
toolchain_notes=["zinc", "amber"],
runtime_effects=["r2", "r1"],
)
def construct_case(name, kwargs):
case(name, "construct", {"kwargs": kwargs},
lambda: cp.ConstructionPattern.from_dict(kwargs).to_dict())
construct_case("construct-full", {"pattern_id": "p.full", "purpose": "full pattern",
"wibble": "future-data", **FULL_KW})
construct_case("construct-minimal", {"pattern_id": "p.min", "purpose": "minimal pattern"})
construct_case("construct-ext-open-world", {"pattern_id": "p.ext", "purpose": "open world",
"wibble": "future-data", "count": 7})
construct_case("construct-empty-id", {"pattern_id": "", "purpose": "purpose"})
construct_case("construct-empty-purpose", {"pattern_id": "p", "purpose": ""})
construct_case("construct-obl-missing-id",
{"pattern_id": "p", "purpose": "purpose",
"obligations": [{"obligation": "x", "when": {}}]})
construct_case("construct-obl-missing-text",
{"pattern_id": "p", "purpose": "purpose", "obligations": [{"id": "x", "when": {}}]})
construct_case("construct-obl-missing-when",
{"pattern_id": "p", "purpose": "purpose", "obligations": [{"id": "x", "obligation": "x"}]})
construct_case("construct-from-dict-roundtrip",
cp.ConstructionPattern("p.rt", "roundtrip", wibble={"deep": [1, 2]}, **FULL_KW).to_dict())
# ── revise ───────────────────────────────────────────────────────────────────
REV_BASE = dict(revision=2, maturity="UNIT-PROVEN",
provenance=["contract:c9"], exclusions=["b-ex", "a-ex"])
def revise_case(name, base_kwargs, reason, evidence, changes, ext=None):
base = cp.ConstructionPattern("p.rev", "revisable", **(ext or {}), **base_kwargs)
case(name, "revise",
{"pattern": base.to_dict(), "reason": reason, "evidence": evidence, "changes": changes},
lambda: base.revise(reason=reason, evidence=evidence, changes=changes).to_dict())
revise_case("revise-with-changes", REV_BASE, "new incident evidence",
["z-evidence", "a-evidence"], {"maturity": "EXERCISED"})
revise_case("revise-history-accumulates", {"revision": 2, "revision_history": [
{"prior_revision": 1, "reason": "initial correction", "evidence": ["old-evidence"], "changed": ["signals"]}]},
"second revision", ["new-evidence"], {})
revise_case("revise-reason-truncated-200", {}, "r" * 250, ["e"], None)
revise_case("revise-open-world-ext-carried", {}, "carries ext", ["e"], {"purpose": "revised purpose"},
ext={"wibble": "kept"})
# ── record_pattern_proposal ─────────────────────────────────────────────────
def propose_case(name, pattern_id="x", provider="p", model="m", evidence=(), confidence=0.5):
case(name, "propose",
{"pattern_id": pattern_id, "provider": provider, "model": model,
"evidence": list(evidence), "confidence": confidence},
lambda: cp.record_pattern_proposal(pattern_id, provider=provider, model=model,
evidence=list(evidence), confidence=confidence))
propose_case("propose-valid", pattern_id="some-pattern", provider="local", model="m1",
evidence=["z-hunch", "a-hunch"], confidence=0.5)
case("propose-bounds-zero-one", "propose",
{"pattern_id": "x", "provider": "p", "model": "m", "evidence": [], "confidence": [0.0, 1.0]},
lambda: [cp.record_pattern_proposal("x", provider="p", model="m", evidence=[], confidence=0.0),
cp.record_pattern_proposal("x", provider="p", model="m", evidence=[], confidence=1.0)])
propose_case("propose-empty-provider", provider="")
propose_case("propose-empty-model", model="")
propose_case("propose-confidence-above-1", confidence=1.5)
propose_case("propose-confidence-below-0", confidence=-0.1)
# ── evaluate_applicability ──────────────────────────────────────────────────
def eval_case(name, pattern_kwargs, n=None, surface=None, authorities=None):
p = cp.ConstructionPattern("p.eval", "evaluated pattern", **pattern_kwargs)
nn = n or need()
s = surface or SURFACE
a = authorities or AUTH
case(name, "evaluate",
{"pattern": p.to_dict(), "need": serialize_need(nn), "surface_result": s, "authorities": a},
lambda: cp.evaluate_applicability(p, nn, s, a))
eval_case("eval-all-groups-applies",
{"signals": {"surface_kinds": ["source_path", "ui_component", "layout_region",
"workflow_config", "structural_entity"],
"path_contains": ["deploy"], "entities_match": ["module:"],
"objective_any": ["promotion"],
"contracts_match": ["contract:monotonic_promotion"]}})
eval_case("eval-partial",
{"signals": {"path_contains": ["deploy"], "entities_match": ["nope-never", "zzz"],
"objective_any": ["zzzneverword"]}})
eval_case("eval-none-declared-does-not-apply",
{"signals": {"path_contains": ["/nonexistent/xyz/"], "objective_any": ["zzznever"]}})
eval_case("eval-no-signals-unknown", {})
eval_case("eval-no-groups-declared-unknown", {"signals": {"surface_kinds": [], "path_contains": []}})
eval_case("eval-exclusion-objective-substring",
{"signals": {"path_contains": ["deploy"]}, "exclusions": ["no deployment system in this project"]})
eval_case("eval-exclusion-constraint-membership",
{"signals": {"path_contains": ["deploy"]}, "exclusions": ["message-queue-only"]},
n=need(constraints={"contracts": [], "pattern_exclusions": ["message-queue-only"]}))
eval_case("eval-exclusion-first-match-wins",
{"signals": {"path_contains": ["deploy"]},
"exclusions": ["no deployment system in this project", "message-queue-only"]},
n=need(constraints={"contracts": [], "pattern_exclusions": ["message-queue-only"]}))
eval_case("eval-exclusion-declared-but-unmatched",
{"signals": {"path_contains": ["deploy"]}, "exclusions": ["totally-unrelated-exclusion"]},
n=need(constraints={"contracts": [], "pattern_exclusions": ["another-unrelated"]}))
eval_case("eval-surface-kinds-via-resolutions",
{"signals": {"surface_kinds": ["structural_entity", "nope_kind"]}})
eval_case("eval-path-short-token",
{"signals": {"path_contains": ["ui", "tui", "cfg", "deploy"]}})
eval_case("eval-entity-prefix-and-tokens",
{"signals": {"entities_match": ["ui:", "module:", "ui", "spine", "zzz"]}})
eval_case("eval-objective-slash-split",
{"signals": {"objective_any": ["deploy", "release", "widget"]}},
n=need(objective="adjust the release/deploy widget", constraints={"contracts": []}))
eval_case("eval-objective-dash-not-split",
{"signals": {"objective_any": ["deploy", "pre-deploy"]}},
n=need(objective="verify the pre-deploy check", constraints={"contracts": []}))
eval_case("eval-contracts-all-rel-kinds",
{"signals": {"contracts_match": ["contract:monotonic_promotion", "contract:exposed_api",
"contract:served_edge", "contract:fail_closed_deployment",
"contract:provided_gate", "contract:second_edge"]}})
eval_case("eval-contracts-miss-non-rel-type",
{"signals": {"contracts_match": ["module:deploy_qualify"]}})
eval_case("eval-contracts-second-authority",
{"signals": {"contracts_match": ["contract:scaffold_only"]}},
authorities={"rune_spine": SPINE,
"project_scaffold": {"relationships": [
{"type": "implements", "from": "module:deploy_stage", "to": "contract:scaffold_only"}]}})
eval_case("eval-empty-entity-id-quirk",
{"signals": {"contracts_match": ["contract:monotonic_promotion"]}},
surface={**SURFACE, "affected_existing_entities": ["", "module:deploy_qualify"]})
# ── derive_pattern_obligations ──────────────────────────────────────────────
O = lambda oid, text, when: {"id": oid, "obligation": text, "when": when}
def derive_case(name, obligations, n=None, exceptions=None, surface=None, authorities=None):
p = cp.ConstructionPattern("p.derive", "deriving pattern",
provenance=["incident:prov-1", "contract:prov-0"],
obligations=obligations)
nn = n or need()
s = surface or SURFACE
a = authorities or AUTH
case(name, "derive",
{"pattern": p.to_dict(), "need": serialize_need(nn), "surface_result": s,
"authorities": a, "project_exceptions": exceptions},
lambda: cp.derive_pattern_obligations(p, nn, s, a, project_exceptions=exceptions))
derive_case("derive-when-none-unknown", [O("mystery", "do the unconditioned thing", None)])
derive_case("derive-always-required", [O("always-ob", "regression proof", {"always": True})])
derive_case("derive-always-truthy-not-true",
[O("truthy-ob", "truthy always is not always", {"always": 1})])
derive_case("derive-path-hit-truncation",
[O("path-ob", "touch the deploy paths", {"path_contains": ["deploy", "rune"]})])
derive_case("derive-path-miss", [O("path-ob", "touch nothing real", {"path_contains": ["/absent/xyz/"]})])
derive_case("derive-entities-via-ui-and-regions",
[O("ui-ob", "respect the ui surface", {"entities_match": ["version_display"]}),
O("region-ob", "respect the region", {"entities_match": ["left"]}),
O("entity-ob", "respect the module", {"entities_match": ["module:"]}),
O("miss-ob", "matches nothing", {"entities_match": ["zzz-never"]})])
derive_case("derive-contracts-from-need-constraints",
[O("ct-ob", "honor the explicit contract",
{"contracts_touched": ["contract:explicit_need", "contract:absent"]})],
n=need(constraints={"contracts": ["contract:explicit_need"]}))
derive_case("derive-contracts-from-authorities",
[O("ct-ob", "honor the authority-implied contract",
{"contracts_touched": ["contract:provided_gate", "contract:served_edge", "contract:nowhere"]})])
derive_case("derive-contracts-miss",
[O("ct-ob", "cannot honor the unknown contract", {"contracts_touched": ["contract:not_anywhere"]})])
derive_case("derive-unrecognized-shape", [O("weird-ob", "do the unknown thing", {"unsupported_shape": True})])
derive_case("derive-mixed-routing",
[O("acc-ob", "always applies", {"always": True}),
O("rej-ob", "path never matches", {"path_contains": ["/absent/xyz/"]}),
O("unres-ob", "unknown shape", {"weird": 1})])
derive_case("derive-exception-with-defaults",
[O("exc-ob", "suppressed obligation", {"always": True}),
O("kept-ob", "surviving obligation", {"always": True})],
exceptions=[{"obligation": "exc-ob"}])
derive_case("derive-exception-full",
[O("exc-ob", "suppressed obligation", {"always": True})],
exceptions=[{"obligation": "exc-ob", "reason": "project policy documented",
"evidence": "policy doc", "authority": "user"}])
# ── combine_pattern_obligations ─────────────────────────────────────────────
def derivation(pattern_id, rows, revision=1):
"""rows: (id, verdict, text) — mirrors derive_pattern_obligations output."""
out = {"pattern": pattern_id, "revision": revision, "accepted": [], "rejected": [], "unresolved": []}
for oid, verdict, text in rows:
entry = {"id": oid, "obligation": text, "pattern": pattern_id, "revision": revision,
"provenance": "PATTERN", "pattern_provenance": [f"prov:{pattern_id}"],
"verdict": verdict, "reason": "reason"}
out[{"REQUIRED": "accepted", "APPLICABLE": "accepted",
"NOT_APPLICABLE": "rejected"}.get(verdict, "unresolved")].append(entry)
return out
def combine_case(name, derivs):
case(name, "combine", {"derivations": derivs}, lambda: cp.combine_pattern_obligations(derivs))
combine_case("combine-empty", [])
combine_case("combine-single", [derivation("a", [("shared", "REQUIRED", "regression proof"),
("z-only", "APPLICABLE", "z duty")])])
combine_case("combine-duplicate-collapse", [
derivation("a", [("shared", "REQUIRED", "regression proof")]),
derivation("b", [("shared", "REQUIRED", "regression proof")]),
derivation("c", [("shared", "REQUIRED", "regression proof")])])
combine_case("combine-conflict-different-text", [
derivation("x", [("shared", "REQUIRED", "version A of the rule")]),
derivation("y", [("shared", "REQUIRED", "version B of the rule")])])
combine_case("combine-sort-by-id", [
derivation("p1", [("zz", "REQUIRED", "last"), ("aa", "REQUIRED", "first")]),
derivation("p2", [("mm", "APPLICABLE", "middle")])])
combine_case("combine-conflict-then-continue", [
derivation("x", [("bad", "REQUIRED", "text one"), ("ok", "REQUIRED", "fine duty")]),
derivation("y", [("bad", "REQUIRED", "text two")])])
# ── write truth ──────────────────────────────────────────────────────────────
OUT.write_text(json.dumps({"generator": "scripts/gen_construction_pattern_truth.py",
"module": "src/rune/construction_pattern.py", "cases": cases}, indent=1) + "\n")
print(f"{len(cases)} cases -> {OUT}")
src/rune/construction_pattern.py
+51
View File
@@ -34,8 +34,32 @@ revision consumed so "which builds used the old rule" stays answerable.
from __future__ import annotations
import os
from rune import structural_store as ss
def _core():
"""The Rust pattern core, or None when pinned to the oracle / not built.
RUNE_CP_CORE=python pins the oracle; the kernel is a drop-in and every
caller falls back to the oracle path on any kernel error."""
if os.environ.get("RUNE_CP_CORE", "auto") == "python":
return None
try:
from rune import cp_loader
return cp_loader if cp_loader.available() else None
except Exception:
return None
def _need_dict(need) -> dict:
return {
"objective": need.objective,
"affected_surface": list(need.affected_surface),
"constraints": dict(need.constraints),
}
__all__ = [
"ConstructionPattern",
"evaluate_applicability",
@@ -168,6 +192,12 @@ def record_pattern_proposal(
Recorded with verification state PENDING until deterministic project
evidence, research evidence, proven construction evidence, or an
explicit user decision accepts it."""
core = _core()
if core is not None:
try:
return core.propose(pattern_id, provider, model, list(evidence), confidence)
except Exception:
pass # oracle fallback below; never crash the caller
if not provider or not model:
raise ValueError("pattern proposal requires provider+model")
if not 0.0 <= confidence <= 1.0:
@@ -256,6 +286,13 @@ def evaluate_applicability(
need explicitly declares the pattern non-applicable, DOES_NOT_APPLY
with the exclusion recorded.
"""
core = _core()
if core is not None:
try:
return core.evaluate(pattern.to_dict(), _need_dict(need),
dict(surface_result), dict(authorities))
except Exception:
pass # oracle fallback below; never crash the caller
# §3/§18: PROJECT evidence overrides the pattern. Exclusions may be
# declared by the project/need (pattern_exclusions constraint) or by the
# pattern itself (non-applicability domains).
@@ -427,6 +464,14 @@ def derive_pattern_obligations(
verdict, reason/evidence, and provenance kind PATTERN (distinct from
Tranche 5's STRUCTURAL derivation — §8 provenance separation).
Project exceptions (§18) suppress with provenance retained."""
core = _core()
if core is not None:
try:
return core.derive(pattern.to_dict(), _need_dict(need), dict(surface_result),
dict(authorities),
project_exceptions=list(project_exceptions or []))
except Exception:
pass # oracle fallback below; never crash the caller
project_exceptions = list(project_exceptions or [])
out = {
"pattern": pattern.pattern_id,
@@ -474,6 +519,12 @@ def combine_pattern_obligations(derivations: list[dict]) -> dict:
same obligation id collapse (provenance lists merge); conflicting
obligations (same id, contradictory verdicts across patterns) surface
as conflicts requiring evidence/user decision — never silently chosen."""
core = _core()
if core is not None:
try:
return core.combine(list(derivations))
except Exception:
pass # oracle fallback below; never crash the caller
merged: dict[str, dict] = {}
conflicts: list[dict] = []
for d in derivations:
src/rune/cp_loader.py
+86 -41
View File
@@ -1,11 +1,22 @@
#!/usr/bin/env python3
"""ctypes loader for the Rust construction_projection core (facade burn-down).
"""ctypes loader for the Rust construction_pattern core (dichotomy wave,
#2479 tranche). Mirrors broker_loader.py's contract:
FFI contract (doc-envelope law):
* cp_project({contract_entity, project_id, ...}) -> projection dict
* cp_free_string(ptr)
FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
* cp_construct({pattern dict}) -> normalized pattern dict
* cp_revise({pattern, reason, evidence, changes?}) -> new pattern dict
* cp_propose({pattern_id, provider, model, evidence, confidence}) -> dict
* cp_evaluate({pattern, need, surface_result, authorities}) -> verdict dict
* cp_derive({pattern, need, surface_result, authorities,
project_exceptions?}) -> derivation dict
* cp_combine({derivations}) -> {"obligations", "conflicts"}
The `need` is a plain dict {objective, affected_surface, constraints} —
the kernel never sees the live Need object.
Free: capability_scoring's cs_free_string (shared process). Override the
.so with CP_RUST_LIB (a FILE path).
"""
from __future__ import annotations
import ctypes
@@ -14,64 +25,98 @@ import os
from pathlib import Path
_REPO = Path(__file__).resolve().parents[2]
_CRATE_DIR = _REPO / "src" / "rust" / "construction_projection"
_CRATE_DIR = _REPO / "src" / "rust" / "construction_pattern"
_LIB_PATH = Path(
os.environ.get(
"CP_RUST_LIB", _CRATE_DIR / "target" / "release" / "libconstruction_projection.so"
)
os.environ.get("CP_RUST_LIB", _CRATE_DIR / "target" / "release" / "libconstruction_pattern.so")
)
_lib = None
class ConstructionPatternCoreUnavailable(FileNotFoundError):
"""The Rust pattern core is not built. Callers fall back to Python."""
def available() -> bool:
return _LIB_PATH.is_file()
_SYMS = ("cp_construct", "cp_revise", "cp_propose", "cp_evaluate", "cp_derive", "cp_combine")
def _load() -> ctypes.CDLL:
global _lib
if _lib is not None:
return _lib
if not _LIB_PATH.exists():
raise FileNotFoundError(f"construction_projection not built: {_LIB_PATH}")
if not _LIB_PATH.is_file():
raise ConstructionPatternCoreUnavailable(str(_LIB_PATH))
lib = ctypes.CDLL(str(_LIB_PATH))
lib.cp_free_string.argtypes = [ctypes.c_void_p]
lib.cp_free_string.restype = None
fn = lib.cp_project
fn.argtypes = [ctypes.c_char_p]
fn.restype = ctypes.c_void_p
for sym in _SYMS:
getattr(lib, sym).argtypes = [ctypes.c_void_p]
getattr(lib, sym).restype = ctypes.c_void_p
# free comes from capability_scoring (same process) if loaded; else our own malloc path
try:
lib.cp_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()
ptr = getattr(lib, sym)(json.dumps(payload).encode("utf-8"))
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")
try:
text = ctypes.cast(ptr, ctypes.c_char_p).value.decode("utf-8")
finally:
lib.cp_free_string(ptr)
env = json.loads(text)
if not env.get("ok"):
raise RuntimeError(f"{sym} core error: {env.get('err')}")
return env["value"]
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 project(contract_entity: dict, contract_id: str = "", project_id: str = "", originating_goal_id: str = "",
operation_id: str = "", waiting_state: str = "", live_correlation: str = "",
stage: str = "", builder_state: str = "", verification_state: str = "",
checkpoint: str = "", failure_class: str = "", completion_state: str = "") -> dict:
return _call("cp_project", {"contract_entity": contract_entity,
"contract_id": contract_id,
"project_id": project_id,
"originating_goal_id": originating_goal_id,
"operation_id": operation_id,
"waiting_state": waiting_state,
"live_correlation": live_correlation,
"stage": stage, "builder_state": builder_state,
"verification_state": verification_state,
"checkpoint": checkpoint,
"failure_class": failure_class,
"completion_state": completion_state})
def construct(pattern: dict) -> dict:
return _call("cp_construct", dict(pattern))
def revise(pattern: dict, reason: str, evidence: list, changes: dict | None = None) -> dict:
payload = {"pattern": pattern, "reason": reason, "evidence": list(evidence)}
if changes is not None:
payload["changes"] = dict(changes)
return _call("cp_revise", payload)
def propose(pattern_id: str, provider: str, model: str, evidence: list, confidence: float) -> dict:
return _call("cp_propose", {
"pattern_id": pattern_id, "provider": provider, "model": model,
"evidence": list(evidence), "confidence": confidence,
})
def evaluate(pattern: dict, need: dict, surface_result: dict, authorities: dict) -> dict:
return _call("cp_evaluate", {
"pattern": pattern, "need": need,
"surface_result": surface_result, "authorities": authorities,
})
def derive(pattern: dict, need: dict, surface_result: dict, authorities: dict,
project_exceptions: list | None = None) -> dict:
payload = {"pattern": pattern, "need": need,
"surface_result": surface_result, "authorities": authorities}
if project_exceptions is not None:
payload["project_exceptions"] = list(project_exceptions)
return _call("cp_derive", payload)
def combine(derivations: list) -> dict:
return _call("cp_combine", {"derivations": list(derivations)})
src/rust/construction_pattern/Cargo.lock
Generated
+34 -6
View File
@@ -6,10 +6,37 @@ version = 4
name = "construction_pattern"
version = "0.1.0"
dependencies = [
"serde",
"serde_json",
]
[[package]]
name = "equivalent"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
[[package]]
name = "foldhash"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
[[package]]
name = "hashbrown"
version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
[[package]]
name = "indexmap"
version = "2.14.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
dependencies = [
"equivalent",
"hashbrown",
]
[[package]]
name = "itoa"
version = "1.0.18"
@@ -47,7 +74,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
dependencies = [
"serde_core",
"serde_derive",
]
[[package]]
@@ -72,10 +98,12 @@ dependencies = [
[[package]]
name = "serde_json"
version = "1.0.151"
version = "1.0.154"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6"
dependencies = [
"foldhash",
"indexmap",
"itoa",
"memchr",
"serde",
@@ -85,9 +113,9 @@ dependencies = [
[[package]]
name = "syn"
version = "3.0.6"
version = "3.0.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee"
checksum = "d62a2e0561533f2ca2561d0cf27fd9fedb640a1bf2616ff5d5c80d99017faadc"
dependencies = [
"proc-macro2",
"quote",
src/rust/construction_pattern/Cargo.toml
+1 -2
View File
@@ -7,5 +7,4 @@ edition = "2021"
crate-type = ["rlib", "cdylib", "staticlib"]
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
serde_json = { version = "1", features = ["preserve_order"] }
src/rust/construction_pattern/src/branch_battery.rs
+166 -37
View File
@@ -1,36 +1,173 @@
//! branch_battery.rs — #2479 RUST batch 34: the pattern-matching kernels.
//! ORACLE-VERIFIED: the token-matching discipline is correct — short
//! signals (<4 chars) must match WHOLE tokens, not substrings (the
//! 'ui'-in-'builtin' false-positive class is blocked by design).
//! branch_battery.rs — frozen-oracle battery for the construction_pattern
//! port (dichotomy wave #2479 tranche).
//!
//! Truth embedded VERBATIM from src/frozen_cp_truth.json (frozen by
//! scripts/gen_construction_pattern_truth.py against the live Python
//! oracle, RUNE_CP_CORE=python pinned): every case replays its recorded
//! INPUT through the kernel and asserts byte-identical EXPECTED output
//! (errors compare on the exact ValueError text).
#[cfg(test)]
mod battery {
// batch 155: the kernels are now SHIPPED pub fns (promoted from here).
use crate::{entity_matches, path_matches};
use crate::{
combine_pattern_obligations, construct_pattern, derive_pattern_obligations,
entity_matches, evaluate_applicability, path_matches, record_pattern_proposal,
revise_pattern,
};
use serde_json::Value;
fn truth() -> Value {
serde_json::from_str(include_str!("frozen_cp_truth.json")).expect("frozen truth parses")
}
fn cases(op: &str) -> Vec<Value> {
truth()["cases"]
.as_array()
.expect("cases array")
.iter()
.filter(|c| c["op"] == op)
.cloned()
.collect()
}
fn obj(v: &Value) -> serde_json::Map<String, Value> {
v.as_object().expect("object").clone()
}
fn run_construct(c: &Value) -> Result<Value, String> {
construct_pattern(&obj(&c["input"]["kwargs"])).map(Value::Object)
}
fn run_revise(c: &Value) -> Result<Value, String> {
let i = &c["input"];
let evidence: Vec<Value> = i["evidence"].as_array().cloned().unwrap_or_default();
let changes = if i["changes"].is_null() { None } else { Some(obj(&i["changes"])) };
revise_pattern(&obj(&i["pattern"]), i["reason"].as_str().unwrap_or(""), &evidence, changes.as_ref())
.map(Value::Object)
}
fn run_propose(c: &Value) -> Result<Value, String> {
let i = &c["input"];
if let Some(bounds) = i["confidence"].as_array() {
let mut out = Vec::new();
for b in bounds {
out.push(record_pattern_proposal(
i["pattern_id"].as_str().unwrap_or(""),
i["provider"].as_str().unwrap_or(""),
i["model"].as_str().unwrap_or(""),
vec![],
b.as_f64().unwrap(),
)?);
}
return Ok(Value::Array(out));
}
let evidence: Vec<String> =
i["evidence"].as_array().map(|a| a.iter().filter_map(|e| e.as_str().map(String::from)).collect()).unwrap_or_default();
record_pattern_proposal(
i["pattern_id"].as_str().unwrap_or(""),
i["provider"].as_str().unwrap_or(""),
i["model"].as_str().unwrap_or(""),
evidence,
i["confidence"].as_f64().unwrap_or(f64::NAN),
)
}
fn run_evaluate(c: &Value) -> Value {
let i = &c["input"];
evaluate_applicability(&obj(&i["pattern"]), &obj(&i["need"]), &obj(&i["surface_result"]), &obj(&i["authorities"]))
}
fn run_derive(c: &Value) -> Value {
let i = &c["input"];
let exc: Vec<Value> =
if i["project_exceptions"].is_null() { vec![] } else { i["project_exceptions"].as_array().cloned().unwrap_or_default() };
derive_pattern_obligations(&obj(&i["pattern"]), &obj(&i["need"]), &obj(&i["surface_result"]), &obj(&i["authorities"]), &exc)
}
fn run_combine(c: &Value) -> Value {
let derivs: Vec<Value> = c["input"]["derivations"].as_array().cloned().unwrap_or_default();
combine_pattern_obligations(&derivs)
}
fn check(name: &str, got: Result<Value, String>, expected: &Value) {
match expected.get("error") {
Some(Value::String(msg)) => match got {
Err(e) => assert_eq!(&e, msg, "{name}: error text"),
Ok(v) => panic!("{name}: kernel succeeded ({v}) where oracle raised"),
},
_ => match got {
Ok(v) => assert_eq!(&v, expected, "{name}: value mismatch"),
Err(e) => panic!("{name}: unexpected kernel error {e}"),
},
}
}
#[test]
fn battery_construct() {
let cs = cases("construct");
assert!(!cs.is_empty());
for c in &cs {
check(c["name"].as_str().unwrap(), run_construct(c), &c["expected"]);
}
}
#[test]
fn battery_revise() {
let cs = cases("revise");
assert!(!cs.is_empty());
for c in &cs {
check(c["name"].as_str().unwrap(), run_revise(c), &c["expected"]);
}
}
#[test]
fn battery_propose() {
let cs = cases("propose");
assert!(!cs.is_empty());
for c in &cs {
check(c["name"].as_str().unwrap(), run_propose(c), &c["expected"]);
}
}
#[test]
fn battery_evaluate() {
let cs = cases("evaluate");
assert!(!cs.is_empty());
for c in &cs {
let got = run_evaluate(c);
assert_eq!(&got, &c["expected"], "{}: value mismatch", c["name"]);
}
}
#[test]
fn battery_derive() {
let cs = cases("derive");
assert!(!cs.is_empty());
for c in &cs {
let got = run_derive(c);
assert_eq!(&got, &c["expected"], "{}: value mismatch", c["name"]);
}
}
#[test]
fn battery_combine() {
let cs = cases("combine");
assert!(!cs.is_empty());
for c in &cs {
let got = run_combine(c);
assert_eq!(&got, &c["expected"], "{}: value mismatch", c["name"]);
}
}
// ── oracle-verified matcher discipline (batch 155, preserved) ──
#[test]
fn path_token_discipline() {
// ORACLE-VERIFIED: short signals match WHOLE tokens only
assert!(
!path_matches("ui", "src/rune/rune_tui.py"),
"'ui' is not a token in rune_tui.py"
);
assert!(
!path_matches("ui", "src/rune/builtin.py"),
"the 'ui'-in-'builtin' FP class"
);
assert!(
path_matches("spine", "src/rune/spine.py"),
"long signal: containment"
);
assert!(
path_matches("tui", "src/rune/rune_tui.py"),
"'tui' IS a token"
);
assert!(
!path_matches("cfg", "config/settings.py"),
"'cfg' not a token of config"
);
assert!(!path_matches("ui", "src/rune/rune_tui.py"), "'ui' is not a token in rune_tui.py");
assert!(!path_matches("ui", "src/rune/builtin.py"), "the 'ui'-in-'builtin' FP class");
assert!(path_matches("spine", "src/rune/spine.py"), "long signal: containment");
assert!(path_matches("tui", "src/rune/rune_tui.py"), "'tui' IS a token");
assert!(!path_matches("cfg", "config/settings.py"), "'cfg' not a token of config");
}
#[test]
@@ -39,21 +176,13 @@ mod battery {
assert!(entity_matches("module:", "module:spine"), "prefix match");
assert!(!entity_matches("ui:", "region:left"), "prefix mismatch");
assert!(entity_matches("spine", "module:spine"), "long containment");
// ORACLE-VERIFIED: 'ui' (no colon, <4 chars) does NOT match 'ui:chat'
// because the entity token-split keeps ':' as part of the token
assert!(
!entity_matches("ui", "ui:chat"),
"oracle truth: token discipline"
);
assert!(!entity_matches("ui", "ui:chat"), "oracle truth: token discipline");
}
#[test]
fn long_signal_containment() {
assert!(path_matches("structural", "src/rune/structural_store.py"));
assert!(entity_matches(
"construction",
"module:construction_pattern"
));
assert!(entity_matches("construction", "module:construction_pattern"));
assert!(!path_matches("structural", "src/rune/other.py"));
}
}
src/rust/construction_pattern/src/frozen_cp_truth.json
+6315
View File
File diff suppressed because it is too large. Load diff
src/rust/construction_pattern/src/lib.rs
+751 -51
View File
@@ -1,74 +1,774 @@
//! construction_pattern — reusable construction knowledge (W6).
use serde::{Deserialize, Serialize};
//! construction_pattern — reusable evidence about HOW a class of thing
//! is normally built (SCAFFOLD-TRANCHE-006).
//!
//! FULL parity port of src/rune/construction_pattern.py (dichotomy wave,
//! #2479 tranche). A pattern is GUIDANCE, not authority (§3): project
//! evidence always wins. Applicability is evidence-backed and deterministic;
//! obligation derivation is filtered by the project; provenance answers
//! "why is this here"; revisions keep old-rule traceability (§17).
//!
//! FFI (doc-envelope law), one entry point per public oracle function:
//! cp_construct({pattern dict}) → normalized pattern dict
//! cp_revise({pattern, reason, evidence, changes?}) → new pattern dict
//! cp_propose({pattern_id, provider, model, evidence, confidence}) → dict
//! cp_evaluate({pattern, need, surface_result, authorities}) → verdict dict
//! cp_derive({pattern, need, surface_result, authorities,
//! project_exceptions?}) → derivation dict
//! cp_combine({derivations}) → {"obligations", "conflicts"}
//! The `need` arrives as a plain dict {objective, affected_surface,
//! constraints} — the kernel never sees the live Need object.
//!
//! The boundary-aware matchers are the oracle-verified kernels promoted by
//! batch 155 (the 'ui'-in-'builtin' false-positive class is blocked by
//! design): short (<4 char, char-counted) signals match whole tokens only.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConstructionPattern {
pub pattern_id: String,
pub description: String,
pub obligations: Vec<serde_json::Value>,
pub revision: String,
}
use serde_json::{json, Map, Value};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PatternApplicability {
pub verdict: String,
pub reason: String,
}
// ── verdict lattices (string constants, oracle-identical) ─────────────────
pub fn check_applicability(
_pattern: &ConstructionPattern,
project_evidence: &serde_json::Value,
) -> PatternApplicability {
// Patterns are GUIDANCE, not authority. Project evidence wins.
if project_evidence
.get("overrides_pattern")
.and_then(|o| o.as_bool())
.unwrap_or(false)
{
return PatternApplicability {
verdict: "DOES_NOT_APPLY".to_string(),
reason: "project evidence overrides pattern".to_string(),
};
pub const APPLIES: &str = "APPLIES";
pub const PARTIAL: &str = "PARTIAL";
pub const DOES_NOT_APPLY: &str = "DOES_NOT_APPLY";
pub const UNKNOWN: &str = "UNKNOWN";
pub const OBL_REQUIRED: &str = "REQUIRED";
pub const OBL_APPLICABLE: &str = "APPLICABLE";
pub const OBL_NOT_APPLICABLE: &str = "NOT_APPLICABLE";
pub const OBL_UNKNOWN: &str = "UNKNOWN";
const KNOWN_KEYS: &[&str] = &[
"pattern_id",
"purpose",
"revision",
"maturity",
"signals",
"obligations",
"provenance",
"exclusions",
"toolchain_notes",
"runtime_effects",
];
const REL_TYPES: &[&str] = &["implements", "exposes", "provides", "serves"];
// ── Python-repr helpers (reason/error strings embed Python list/dict reprs) ──
/// Single-quoted Python string repr (escapes \\, \', \n, \r, \t).
fn py_quote(s: &str) -> String {
let mut out = String::with_capacity(s.len() + 2);
out.push('\'');
for c in s.chars() {
match c {
'\\' => out.push_str("\\\\"),
'\'' => out.push_str("\\'"),
'\n' => out.push_str("\\n"),
'\r' => out.push_str("\\r"),
'\t' => out.push_str("\\t"),
_ => out.push(c),
}
}
PatternApplicability {
verdict: "APPLIES".to_string(),
reason: "no override".to_string(),
out.push('\'');
out
}
/// Python `repr` of a JSON value: True/False/None, single-quoted strings,
/// `[...]` lists, `{k: v}` dicts in insertion order.
fn py_repr(v: &Value) -> String {
match v {
Value::Null => "None".to_string(),
Value::Bool(true) => "True".to_string(),
Value::Bool(false) => "False".to_string(),
Value::Number(n) => n.to_string(),
Value::String(s) => py_quote(s),
Value::Array(items) => {
let inner: Vec<String> = items.iter().map(py_repr).collect();
format!("[{}]", inner.join(", "))
}
Value::Object(map) => {
let inner: Vec<String> = map.iter().map(|(k, v)| format!("{}: {}", py_quote(k), py_repr(v))).collect();
format!("{{{}}}", inner.join(", "))
}
}
}
/// pathMatches — boundary-aware path-signal matching (batch 155: promoted
/// from the test module to the shipped lib — the kernels ARE the product).
/// Short (<4 char) signals match path TOKENS only (the 'ui'-in-'builtin'
/// false-positive class); longer signals may use containment.
fn py_str_list(items: &[String]) -> String {
let inner: Vec<String> = items.iter().map(|s| py_quote(s)).collect();
format!("[{}]", inner.join(", "))
}
// ── shared boundary-aware matchers (Tranche 7 §0) ──────────────────────────
/// Oracle-faithful token split: runs of non-token chars are ONE separator,
/// with leading/trailing empties kept (`re.split` semantics — only the
/// empties are observable, via the empty-signal quirk).
fn split_tokens(lowered: &str, keep_colon: bool) -> Vec<String> {
let is_tok = |c: char| c.is_ascii_lowercase() || c.is_ascii_digit() || (keep_colon && c == ':');
let mut out = Vec::new();
let mut cur = String::new();
let mut pending_empty = true;
for c in lowered.chars() {
if is_tok(c) {
cur.push(c);
pending_empty = false;
} else if !cur.is_empty() || pending_empty {
out.push(std::mem::take(&mut cur));
pending_empty = false;
}
}
out.push(cur);
out
}
/// Boundary-aware path-signal matching: short (<4 char) signals match path
/// TOKENS only (the 'ui'-in-'builtin' false-positive class); longer signals
/// may use containment.
pub fn path_matches(signal: &str, path: &str) -> bool {
let s = signal.to_lowercase();
if s.len() >= 4 {
if s.chars().count() >= 4 {
return path.to_lowercase().contains(&s);
}
let lower = path.to_lowercase();
let tokens: std::collections::HashSet<&str> = lower
.split(|c: char| !c.is_ascii_lowercase() && !c.is_ascii_digit())
.collect();
tokens.contains(s.as_str())
split_tokens(&path.to_lowercase(), false).iter().any(|t| t == &s)
}
/// entityMatches — same discipline for entity ids; ':'-suffixed signals
/// match the namespace prefix (structural kinds carry meaning there).
/// Same discipline for entity ids. Prefix signals ending in ':' (e.g.
/// 'ui:', 'region:') match the id's namespace prefix — structural kinds
/// carry meaning in their prefix, and prefix equality is exact, not a
/// substring accident.
pub fn entity_matches(signal: &str, entity_id: &str) -> bool {
let s = signal.to_lowercase();
if s.ends_with(':') {
return entity_id.to_lowercase().starts_with(&s);
return entity_id.to_lowercase().starts_with(s.as_str());
}
if s.len() >= 4 {
if s.chars().count() >= 4 {
return entity_id.to_lowercase().contains(&s);
}
let lower = entity_id.to_lowercase();
let tokens: std::collections::HashSet<&str> = lower
.split(|c: char| !c.is_ascii_lowercase() && !c.is_ascii_digit() && c != ':')
.collect();
tokens.contains(s.as_str())
split_tokens(&entity_id.to_lowercase(), true).iter().any(|t| t == &s)
}
#[cfg(test)]
// ── small JSON accessors ──────────────────────────────────────────────────
fn str_list(v: &Value) -> Vec<String> {
v.as_array()
.map(|a| a.iter().filter_map(|x| x.as_str().map(String::from)).collect())
.unwrap_or_default()
}
fn sorted_strings(mut v: Vec<String>) -> Vec<String> {
v.sort();
v
}
/// `ss.relationships(doc, from_id=e)` — including the falsy-filter quirk:
/// an empty from_id applies NO filter (Python `if from_id and ...`).
fn relationships<'a>(doc: &'a Value, from_id: &str) -> Vec<&'a Map<String, Value>> {
doc.get("relationships")
.and_then(|r| r.as_array())
.map(|rels| {
rels.iter()
.filter_map(|r| r.as_object())
.filter(|r| from_id.is_empty() || r.get("from").and_then(|f| f.as_str()).unwrap_or("") == from_id)
.collect()
})
.unwrap_or_default()
}
// ── pattern object (open-world) ───────────────────────────────────────────
/// `ConstructionPattern.from_dict` / constructor / `to_dict`: validate the
/// required fields, then normalize (sorted provenance-family lists, ext
/// passthrough). The input and output share one shape — `to_dict` output
/// round-trips through here unchanged.
pub fn construct_pattern(input: &Map<String, Value>) -> Result<Map<String, Value>, String> {
let pattern_id = input.get("pattern_id").and_then(|v| v.as_str()).unwrap_or("");
if pattern_id.is_empty() {
return Err("pattern_id required".to_string());
}
let purpose = input.get("purpose").and_then(|v| v.as_str()).unwrap_or("");
if purpose.is_empty() {
return Err("purpose required".to_string());
}
if let Some(obs) = input.get("obligations").and_then(|v| v.as_array()) {
for o in obs {
for key in ["id", "obligation", "when"] {
if o.get(key).is_none() {
return Err(format!("obligation missing {}: {}", py_quote(key), py_repr(o)));
}
}
}
}
let mut out = Map::new();
out.insert("pattern_id".into(), Value::String(pattern_id.to_string()));
out.insert("purpose".into(), Value::String(purpose.to_string()));
out.insert(
"revision".into(),
input.get("revision").cloned().unwrap_or(json!(1)),
);
out.insert(
"maturity".into(),
input.get("maturity").cloned().unwrap_or(json!("CANDIDATE")),
);
out.insert(
"signals".into(),
input.get("signals").cloned().unwrap_or_else(|| Value::Object(Map::new())),
);
out.insert(
"obligations".into(),
input.get("obligations").cloned().unwrap_or_else(|| Value::Array(vec![])),
);
for key in ["provenance", "exclusions", "toolchain_notes", "runtime_effects"] {
let vals = input.get(key).map(str_list).unwrap_or_default();
out.insert(key.into(), json!(sorted_strings(vals)));
}
for (k, v) in input.iter() {
if !KNOWN_KEYS.contains(&k.as_str()) {
out.insert(k.clone(), v.clone());
}
}
Ok(out)
}
/// `ConstructionPattern.revise` (§17): returns a NEW pattern at revision+1
/// recording the change — prior meaning is never invisibly mutated.
pub fn revise_pattern(
pattern: &Map<String, Value>,
reason: &str,
evidence: &[Value],
changes: Option<&Map<String, Value>>,
) -> Result<Map<String, Value>, String> {
let pattern_id = pattern.get("pattern_id").and_then(|v| v.as_str()).unwrap_or("").to_string();
let revision = pattern.get("revision").and_then(|v| v.as_i64()).unwrap_or(1);
let mut merged = pattern.clone();
merged.remove("pattern_id");
if let Some(ch) = changes {
for (k, v) in ch.iter() {
merged.insert(k.clone(), v.clone());
}
}
merged.insert("revision".into(), json!(revision + 1));
let mut ev: Vec<String> = evidence.iter().filter_map(|e| e.as_str().map(String::from)).collect();
ev.sort();
let mut changed: Vec<String> = changes.map(|ch| ch.keys().cloned().collect()).unwrap_or_default();
changed.sort();
let entry = json!({
"prior_revision": revision,
"reason": reason.chars().take(200).collect::<String>(),
"evidence": ev,
"changed": changed,
});
match merged.get_mut("revision_history") {
Some(Value::Array(hist)) => hist.push(entry),
_ => {
merged.insert("revision_history".into(), Value::Array(vec![entry]));
}
}
merged.insert("pattern_id".into(), Value::String(pattern_id));
construct_pattern(&merged)
}
// ── model proposal seam (§5 — recorded, never authoritative) ────────────────
/// A model may PROPOSE a pattern; it may not certify applicability.
pub fn record_pattern_proposal(
pattern_id: &str,
provider: &str,
model: &str,
evidence: Vec<String>,
confidence: f64,
) -> Result<Value, String> {
if provider.is_empty() || model.is_empty() {
return Err("pattern proposal requires provider+model".to_string());
}
if !(0.0..=1.0).contains(&confidence) {
return Err("confidence must be in [0,1]".to_string());
}
Ok(json!({
"kind": "pattern_proposal",
"pattern_id": pattern_id,
"provider": provider,
"model": model,
"evidence": sorted_strings(evidence),
"confidence": confidence,
"verification_state": "PENDING",
}))
}
// ── applicability (§4 — deterministic signals only) ─────────────────────────
fn surface_kind_present(kind: &str, surface: &Map<String, Value>) -> bool {
let mapped = match kind {
"source_path" => "affected_paths",
"workflow_config" => "workflow_config_surfaces",
"ui_component" => "affected_ui_surfaces",
"layout_region" => "affected_regions",
_ => "",
};
if surface.get(mapped).and_then(|v| v.as_array()).map(|a| !a.is_empty()).unwrap_or(false) {
return true;
}
surface
.get("resolutions")
.and_then(|v| v.as_array())
.map(|rs| rs.iter().any(|r| r.get("kind").and_then(|k| k.as_str()).unwrap_or("") == kind))
.unwrap_or(false)
}
/// Deterministic applicability verdict from the pattern's declared signals.
pub fn evaluate_applicability(
pattern: &Map<String, Value>,
need: &Map<String, Value>,
surface: &Map<String, Value>,
authorities: &Map<String, Value>,
) -> Value {
let pattern_id = pattern.get("pattern_id").and_then(|v| v.as_str()).unwrap_or("").to_string();
let objective = need.get("objective").and_then(|v| v.as_str()).unwrap_or("").to_string();
let exclusions = pattern.get("exclusions").map(str_list).unwrap_or_default();
let need_exclusions = need
.get("constraints")
.and_then(|c| c.get("pattern_exclusions"))
.map(str_list)
.unwrap_or_default();
// §3/§18: PROJECT evidence overrides the pattern.
let mut declared = exclusions;
declared.extend(need_exclusions.clone());
let objective_low = objective.to_lowercase();
for ex in &declared {
if objective_low.contains(&ex.to_lowercase()) || need_exclusions.iter().any(|x| x == ex) {
return json!({
"pattern": pattern_id,
"verdict": DOES_NOT_APPLY,
"reason": format!("declared exclusion matched: {ex}"),
"evidence": {"exclusion": ex, "source": "project/need declaration"},
});
}
}
let signals = pattern.get("signals").and_then(|v| v.as_object()).cloned().unwrap_or_default();
if signals.is_empty() {
return json!({
"pattern": pattern_id,
"verdict": UNKNOWN,
"reason": "pattern declares no applicability signals — nothing deterministic to check",
"evidence": {},
});
}
let mut groups_hit: Vec<String> = Vec::new();
let mut groups_declared: i64 = 0;
let mut evidence = Map::new();
let paths: Vec<String> = surface
.get("affected_paths")
.map(str_list)
.unwrap_or_default()
.into_iter()
.chain(surface.get("workflow_config_surfaces").map(str_list).unwrap_or_default())
.collect();
let entities: Vec<String> =
surface.get("affected_existing_entities").map(str_list).unwrap_or_default();
let mut check = |group: &str, candidates: Vec<String>, present: &dyn Fn(&str) -> bool| {
if candidates.is_empty() {
return;
}
groups_declared += 1;
let hits: Vec<String> = candidates.iter().filter(|c| present(c)).cloned().collect();
if !hits.is_empty() {
groups_hit.push(group.to_string());
evidence.insert(group.to_string(), json!(hits.iter().take(5).collect::<Vec<_>>()));
}
};
check(
"surface_kinds",
signals.get("surface_kinds").map(str_list).unwrap_or_default(),
&|c| surface_kind_present(c, surface),
);
check(
"path_contains",
signals.get("path_contains").map(str_list).unwrap_or_default(),
&|c| paths.iter().any(|p| path_matches(c, p)),
);
check(
"entities_match",
signals.get("entities_match").map(str_list).unwrap_or_default(),
&|c| entities.iter().any(|e| entity_matches(c, e)),
);
let objective_words: Vec<String> =
objective.replace('/', " ").split_whitespace().map(|w| w.to_lowercase()).collect();
check(
"objective_any",
signals
.get("objective_any")
.map(str_list)
.unwrap_or_default()
.into_iter()
.map(|w| w.to_lowercase())
.collect(),
&|c| objective_words.iter().any(|w| w == c),
);
if let Some(contracts) = signals.get("contracts_match").and_then(|v| v.as_array()) {
if !contracts.is_empty() {
groups_declared += 1;
let wanted: Vec<&str> = contracts.iter().filter_map(|c| c.as_str()).collect();
let mut hits: Vec<String> = Vec::new();
for doc in authorities.values() {
for e in &entities {
for r in relationships(doc, e) {
let rtype = r.get("type").and_then(|t| t.as_str()).unwrap_or("");
let to = r.get("to").and_then(|t| t.as_str()).unwrap_or("");
if REL_TYPES.contains(&rtype) && wanted.contains(&to) {
hits.push(format!("{e}->{to}"));
}
}
}
}
if !hits.is_empty() {
groups_hit.push("contracts_match".to_string());
evidence.insert("contracts_match".into(), json!(hits.iter().take(5).collect::<Vec<_>>()));
}
}
}
if groups_declared == 0 {
return json!({
"pattern": pattern_id,
"verdict": UNKNOWN,
"reason": "no applicable signal groups declared",
"evidence": {},
});
}
let verdict = if groups_hit.len() as i64 == groups_declared {
APPLIES
} else if !groups_hit.is_empty() {
PARTIAL
} else {
DOES_NOT_APPLY
};
let mut hit_sorted = groups_hit.clone();
hit_sorted.sort();
let detail = if hit_sorted.is_empty() { "none".to_string() } else { py_str_list(&hit_sorted) };
json!({
"pattern": pattern_id,
"verdict": verdict,
"reason": format!("{}/{} signal groups produced evidence: {}", groups_hit.len(), groups_declared, detail),
"evidence": evidence,
})
}
// ── obligation derivation (§6-7 — filtered by project evidence) ─────────────
/// Evaluate one obligation's `when` condition against project evidence.
pub fn obligation_when_hit(
obligation: &Map<String, Value>,
need: &Map<String, Value>,
surface: &Map<String, Value>,
authorities: &Map<String, Value>,
) -> (String, String) {
let paths: Vec<String> = surface
.get("affected_paths")
.map(str_list)
.unwrap_or_default()
.into_iter()
.chain(surface.get("workflow_config_surfaces").map(str_list).unwrap_or_default())
.collect();
let entities: Vec<String> = surface
.get("affected_existing_entities")
.map(str_list)
.unwrap_or_default()
.into_iter()
.chain(surface.get("affected_ui_surfaces").map(str_list).unwrap_or_default())
.chain(surface.get("affected_regions").map(str_list).unwrap_or_default())
.collect();
let when = match obligation.get("when") {
None | Some(Value::Null) => {
return (OBL_UNKNOWN.to_string(), "obligation declares no applicability condition".to_string());
}
Some(w) => w,
};
if when.get("always").and_then(|v| v.as_bool()).unwrap_or(false) {
return (
OBL_REQUIRED.to_string(),
"pattern-level obligation (always applies when the pattern applies)".to_string(),
);
}
if let Some(subs) = when.get("path_contains") {
let hits: Vec<String> =
str_list(subs).iter().flat_map(|sub| paths.iter().filter(|p| path_matches(sub, p)).cloned()).collect();
if !hits.is_empty() {
return (OBL_APPLICABLE.to_string(), format!("affected path(s): {}", py_str_list(&hits[..3.min(hits.len())])));
}
return (OBL_NOT_APPLICABLE.to_string(), format!("no affected path matches {}", py_repr(subs)));
}
if let Some(subs) = when.get("entities_match") {
let hits: Vec<String> = str_list(subs)
.iter()
.flat_map(|sub| entities.iter().filter(|e| entity_matches(sub, e)).cloned())
.collect();
if !hits.is_empty() {
return (
OBL_APPLICABLE.to_string(),
format!("affected entity/entities: {}", py_str_list(&hits[..3.min(hits.len())])),
);
}
return (OBL_NOT_APPLICABLE.to_string(), format!("no affected entity matches {}", py_repr(subs)));
}
if let Some(wanted) = when.get("contracts_touched") {
let mut touched: Vec<String> = need
.get("constraints")
.and_then(|c| c.get("contracts"))
.map(str_list)
.unwrap_or_default();
for doc in authorities.values() {
for e in &entities {
for r in relationships(doc, e) {
let rtype = r.get("type").and_then(|t| t.as_str()).unwrap_or("");
let to = r.get("to").and_then(|t| t.as_str()).unwrap_or("").to_string();
if REL_TYPES.contains(&rtype) && !touched.contains(&to) {
touched.push(to);
}
}
}
}
let hits: Vec<String> =
str_list(wanted).into_iter().filter(|c| touched.contains(c)).collect();
if !hits.is_empty() {
return (OBL_APPLICABLE.to_string(), format!("contract(s) touched: {}", py_str_list(&hits)));
}
return (OBL_NOT_APPLICABLE.to_string(), format!("touched contracts do not include {}", py_repr(wanted)));
}
let mut keys: Vec<String> = when.as_object().map(|m| m.keys().cloned().collect()).unwrap_or_default();
keys.sort();
(OBL_UNKNOWN.to_string(), format!("unrecognized when-condition shape: {}", py_str_list(&keys)))
}
/// Derive this pattern's obligations FILTERED by project evidence (§7).
pub fn derive_pattern_obligations(
pattern: &Map<String, Value>,
need: &Map<String, Value>,
surface: &Map<String, Value>,
authorities: &Map<String, Value>,
project_exceptions: &[Value],
) -> Value {
let pattern_id = pattern.get("pattern_id").and_then(|v| v.as_str()).unwrap_or("").to_string();
let revision = pattern.get("revision").cloned().unwrap_or(json!(1));
let provenance = pattern.get("provenance").cloned().unwrap_or_else(|| Value::Array(vec![]));
let mut accepted = Vec::new();
let mut rejected = Vec::new();
let mut unresolved = Vec::new();
if let Some(obs) = pattern.get("obligations").and_then(|v| v.as_array()) {
for o in obs {
let o_obj = o.as_object().cloned().unwrap_or_default();
let mut entry = Map::new();
entry.insert("id".into(), o.get("id").cloned().unwrap_or(Value::Null));
entry.insert("obligation".into(), o.get("obligation").cloned().unwrap_or(Value::Null));
entry.insert("pattern".into(), Value::String(pattern_id.clone()));
entry.insert("revision".into(), revision.clone());
entry.insert("provenance".into(), Value::String("PATTERN".to_string()));
entry.insert("pattern_provenance".into(), provenance.clone());
let oid = o.get("id").and_then(|v| v.as_str()).unwrap_or("");
if let Some(exc) = project_exceptions
.iter()
.find(|x| x.get("obligation").and_then(|v| v.as_str()).unwrap_or("") == oid)
{
entry.insert("verdict".into(), Value::String("EXCEPTED".to_string()));
entry.insert(
"reason".into(),
exc.get("reason").cloned().unwrap_or_else(|| Value::String(String::new())),
);
// Oracle field names on the exception are `evidence` /
// `authority`; they land on the entry as exception_*.
entry.insert(
"exception_evidence".into(),
exc.get("evidence").cloned().unwrap_or_else(|| Value::String(String::new())),
);
entry.insert(
"exception_authority".into(),
exc.get("authority").cloned().unwrap_or_else(|| Value::String("user".to_string())),
);
rejected.push(Value::Object(entry));
continue;
}
let (verdict, reason) = obligation_when_hit(&o_obj, need, surface, authorities);
entry.insert("verdict".into(), Value::String(verdict.clone()));
entry.insert("reason".into(), Value::String(reason));
if verdict == OBL_REQUIRED || verdict == OBL_APPLICABLE {
accepted.push(Value::Object(entry));
} else if verdict == OBL_NOT_APPLICABLE {
rejected.push(Value::Object(entry));
} else {
unresolved.push(Value::Object(entry));
}
}
}
json!({
"pattern": pattern_id,
"revision": revision,
"accepted": accepted,
"rejected": rejected,
"unresolved": unresolved,
})
}
/// Combine multiple patterns' derivations (§26-27): duplicates of the same
/// obligation id collapse (provenance lists merge); conflicting obligations
/// surface as conflicts — never silently chosen.
pub fn combine_pattern_obligations(derivations: &[Value]) -> Value {
let mut merged: Map<String, Value> = Map::new();
let mut order: Vec<String> = Vec::new();
let mut conflicts = Vec::new();
for d in derivations {
if let Some(acc) = d.get("accepted").and_then(|v| v.as_array()) {
for entry in acc {
let oid = entry.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string();
if let Some(prev) = merged.get_mut(&oid) {
let prev_text = prev.get("obligation").and_then(|v| v.as_str()).unwrap_or("");
let new_text = entry.get("obligation").and_then(|v| v.as_str()).unwrap_or("");
if prev_text != new_text {
conflicts.push(json!({
"obligation_id": oid,
"conflict": "same id, different text",
"sources": [
prev.get("pattern").cloned().unwrap_or(Value::Null),
entry.get("pattern").cloned().unwrap_or(Value::Null),
],
}));
continue;
}
let prev_obj = prev.as_object_mut().unwrap();
match prev_obj.get_mut("also_from") {
Some(Value::Array(arr)) => {
arr.push(entry.get("pattern").cloned().unwrap_or(Value::Null))
}
_ => {
prev_obj.insert(
"also_from".into(),
json!([entry.get("pattern").cloned().unwrap_or(Value::Null)]),
);
}
}
continue;
}
order.push(oid.clone());
merged.insert(oid, entry.clone());
}
}
}
order.sort();
let obligations: Vec<Value> = order.into_iter().filter_map(|k| merged.remove(&k)).collect();
json!({"obligations": obligations, "conflicts": conflicts})
}
// ── FFI (doc-envelope law) ──────────────────────────────────────────────
use std::ffi::{CStr, CString};
use std::os::raw::c_char;
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 obj_arg(a: &Value) -> Result<Map<String, Value>, String> {
a.as_object().cloned().ok_or_else(|| "argument must be a JSON object".to_string())
}
/// cp_construct({pattern dict}) → normalized pattern dict.
#[no_mangle]
pub extern "C" fn cp_construct(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| obj_arg(&a)).and_then(|a| construct_pattern(&a).map(Value::Object));
envelope(r)
}
/// cp_revise({pattern, reason, evidence, changes?}) → new pattern dict.
#[no_mangle]
pub extern "C" fn cp_revise(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| obj_arg(&a)).and_then(|a| {
let pattern = a.get("pattern").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let reason = a.get("reason").and_then(|v| v.as_str()).unwrap_or("");
let evidence: Vec<Value> =
a.get("evidence").and_then(|v| v.as_array()).cloned().unwrap_or_default();
let changes = a.get("changes").and_then(|v| v.as_object()).cloned();
revise_pattern(&pattern, reason, &evidence, changes.as_ref()).map(Value::Object)
});
envelope(r)
}
/// cp_propose({pattern_id, provider, model, evidence, confidence}) → dict.
#[no_mangle]
pub extern "C" fn cp_propose(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| obj_arg(&a)).and_then(|a| {
let evidence = a.get("evidence").map(str_list).unwrap_or_default();
let confidence = a.get("confidence").and_then(|v| v.as_f64()).unwrap_or(f64::NAN);
record_pattern_proposal(
a.get("pattern_id").and_then(|v| v.as_str()).unwrap_or(""),
a.get("provider").and_then(|v| v.as_str()).unwrap_or(""),
a.get("model").and_then(|v| v.as_str()).unwrap_or(""),
evidence,
confidence,
)
});
envelope(r)
}
/// cp_evaluate({pattern, need, surface_result, authorities}) → verdict dict.
#[no_mangle]
pub extern "C" fn cp_evaluate(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| obj_arg(&a)).map(|a| {
let pattern = a.get("pattern").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let need = a.get("need").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let surface = a.get("surface_result").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let authorities = a.get("authorities").and_then(|v| v.as_object()).cloned().unwrap_or_default();
evaluate_applicability(&pattern, &need, &surface, &authorities)
});
envelope(r)
}
/// cp_derive({pattern, need, surface_result, authorities,
/// project_exceptions?}) → derivation dict.
#[no_mangle]
pub extern "C" fn cp_derive(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| obj_arg(&a)).map(|a| {
let pattern = a.get("pattern").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let need = a.get("need").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let surface = a.get("surface_result").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let authorities = a.get("authorities").and_then(|v| v.as_object()).cloned().unwrap_or_default();
let exceptions: Vec<Value> =
a.get("project_exceptions").and_then(|v| v.as_array()).cloned().unwrap_or_default();
derive_pattern_obligations(&pattern, &need, &surface, &authorities, &exceptions)
});
envelope(r)
}
/// cp_combine({derivations}) → {"obligations", "conflicts"}.
#[no_mangle]
pub extern "C" fn cp_combine(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| obj_arg(&a)).map(|a| {
let derivs: Vec<Value> =
a.get("derivations").and_then(|v| v.as_array()).cloned().unwrap_or_default();
combine_pattern_obligations(&derivs)
});
envelope(r)
}
// NOTE: no cp_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.
#[path = "branch_battery.rs"]
#[cfg(test)]
mod branch_battery;