conductor/rune
1
2 Fork 1
Code Issues 33 Pull Requests 6 Actions 8 Packages Projects Releases 564 Wiki Activity

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

Browse Source 
parity: every public function of MessageGraph ported branch-for-branch —
_load (schema-v1 gate, corrupt/wrong-schema silent-empty, duplicate ids
last-wins with BOTH order entries kept, verbatim extra fields, non-dict
doc AttributeError, non-list nodes TypeError), save (arrival-order doc;
mkdir + json.dumps indent=2 FILE IO stays in the facade), add (the falsy
law: an explicit None parent still picks up the head but carries no edge,
a falsy "" parent stores "" and also carries no edge, ghost parents
allowed, kind checked before edge, Python-repr error strings), tool_call
(json.dumps content rendering stays in the facade — serialization
isomorphism), tool_result (existence check only — any node can parent a
result), revise (the root-revision quirk: old parent None substitutes
the current head, which at add-time is the old node itself), branch,
lineage (superseded skip, ghost superseded_by fallthrough, seen-set
cycle guard, missing-parent break), history (supersedes walk back to
root then forward via superseded_by — the FULL trail from any point on
it), migrate_flat (role/content defaults, existing-file seed), cue
(first supersedes/branches_from edge in iteration order wins, falsy
superseded_by is no cue).

Determinism seams: the uuid4 hex id and the _now() timestamp are kernel
ARGUMENTS (id/ts) — the frozen truth replays them verbatim. Graph wire:
{nodes: {id: node} insertion-ordered (serde_json preserve_order),
order: [ids]} — nodes ride as opaque Values so loaded extra fields
survive byte-verbatim. Errors ride typed (ValueError/KeyError/
AttributeError/TypeError) and the loader raises the matching Python
exception. The facade keeps the oracle's guard-clause ORDER (kind/edge/
existence checks BEFORE the uuid draw) so the id sequence is identical
under both cores. mg_loader mirrors broker_loader (doc-envelope,
cs_free_string, MG_RUST_LIB override); RUNE_MG_CORE=python pins the
fallback.

The freeze caught a real divergence first: Python 3.14's string-index
TypeError now reads "string indices must be integers, not 'str'" — the
kernel mirrors the runtime's wording, pinned by truth. Battery: 12
tests replay 68 frozen-oracle cases VERBATIM (include_str!); live
parity 68/68 byte-identical (truth regenerated through the kernel path
and diffed against the freeze); clippy 0; Python suite green under both
pins (9+9).
This commit is contained in:
Nick committed 2026-10-12 02:26:30 +01:00
1 parent 6786b4a922
commit 1f7a266497
8 files changed
+5232 -5
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/gen_message_graph_truth.py
+654
View File
@@ -0,0 +1,654 @@
#!/usr/bin/env python3
"""Freeze message_graph oracle truth for the Rust port battery (dichotomy
wave #2479 tranche).
Exercises EVERY public function of src/rune/message_graph.py across every
branch: load (missing file, corrupt JSON, wrong/missing schema, valid doc,
duplicate ids, verbatim extra fields, non-dict doc -> AttributeError,
non-list nodes -> TypeError), save (doc shape + file bytes incl. mkdir),
add (default-parent chain, explicit parent with/without edge, explicit
None -> head, falsy parent "", edge to a ghost parent, unicode, every
kind, bad kind/edge -> ValueError), tool_call (json.dumps content shape,
nested args, parent / default head), tool_result (link, missing call ->
KeyError, non-tool-call parent allowed), revise (root revision picks up
the default-head parent quirk, child revision, chained revisions, missing
-> KeyError), branch (ok, missing -> KeyError), lineage (empty graph,
default head, explicit id, superseded skip, missing-parent break, ghost
superseded_by target, unknown id, parent cycle), history (single, full
chain, from the middle, missing id), migrate_flat (chain, empty
transcript, missing role/content keys, existing-file seed) and cue (no
cue, replaces, branched from, superseded by, falsy superseded_by, edge
iteration order).
Nondeterminism seams: the facade generates the uuid4 hex id and the
_now() timestamp and passes them to the kernel as plain arguments — here
both are pinned (sequential fake uuid4, frozen clock) so the frozen truth
replays VERBATIM on the Rust side and the kernel-side parity pass is a
byte-diff of this file.
Default run pins RUNE_MG_CORE=python (the ORACLE). The kernel-side parity
pass re-runs with RUNE_MG_CORE=auto and MG_TRUTH_OUT set; identical bytes
== 100% live parity.
Output: /tmp/mg_truth.json — replayed VERBATIM by
src/rust/message_graph/src/branch_battery.rs.
"""
from __future__ import annotations
import copy
import json
import os
import sys
import tempfile
import uuid
from pathlib import Path
os.environ.setdefault("RUNE_MG_CORE", "python") # freeze the ORACLE, never the kernel
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from rune import message_graph as mg # noqa: E402
# ── determinism seams (see module docstring) ─────────────────────────────
class _FakeUUID:
def __init__(self, hex_: str) -> None:
self.hex = hex_
_seq = [0]
def _fake_uuid4():
_seq[0] += 1
return _FakeUUID(f"deadbeef{_seq[0]:04d}")
uuid.uuid4 = _fake_uuid4
CLOCK = "2031-01-01T00:00:00.000000+00:00"
mg._now = lambda: CLOCK # noqa: E731 (the oracle's clock, pinned)
TMP = Path(tempfile.mkdtemp(prefix="mg_truth_"))
GROUPS = ("load", "save", "add", "tool_call", "tool_result", "revise",
"branch", "lineage", "history", "migrate_flat", "cue")
groups: dict[str, list] = {g: [] for g in GROUPS}
def case(group: str, name: str, input_, expected) -> None:
groups[group].append({"name": name, "input": input_, "expected": expected})
def wire(g: "mg.MessageGraph") -> dict:
"""Graph state over the wire: nodes as an insertion-ordered object,
order as the arrival list."""
return {"nodes": copy.deepcopy(g.nodes), "order": list(g.order)}
def exc_of(fn, *a, **kw) -> dict:
try:
fn(*a, **kw)
except Exception as e: # noqa: BLE001 — the oracle's exception IS the truth
return {"exception": {"type": type(e).__name__, "message": str(e)}}
raise AssertionError("expected the oracle to raise")
def fresh(name: str, doc_text: str | None = None) -> "mg.MessageGraph":
p = TMP / f"{name}.json"
if doc_text is not None:
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(doc_text)
return mg.MessageGraph(path=p)
# ══ load ═════════════════════════════════════════════════════════════════
# missing file -> empty graph
g = fresh("load_missing")
case("load", "load-missing-file", {"text": None}, {"graph": wire(g)})
# empty object: no schema -> ignored
g = fresh("load_empty_obj", "{}")
case("load", "load-no-schema", {"text": "{}"}, {"graph": wire(g)})
# wrong schema
g = fresh("load_wrong_schema", json.dumps({"schema": 2, "nodes": []}))
case("load", "load-wrong-schema", {"text": json.dumps({"schema": 2, "nodes": []})},
{"graph": wire(g)})
# corrupt JSON
g = fresh("load_corrupt", "not json {")
case("load", "load-corrupt-json", {"text": "not json {"}, {"graph": wire(g)})
# valid doc: nodes ingested verbatim in file order
doc = {"schema": 1, "nodes": [
{"id": "aaa000000001", "kind": "message", "role": "user", "content": "q",
"ts": CLOCK, "parent": None, "edges": [], "superseded_by": None},
{"id": "bbb000000002", "kind": "reply", "role": "assistant", "content": "a",
"ts": CLOCK, "parent": "aaa000000001", "edges": [], "superseded_by": None},
]}
g = fresh("load_valid", json.dumps(doc))
case("load", "load-valid", {"text": json.dumps(doc)}, {"graph": wire(g)})
# duplicate ids: last wins in nodes, BOTH stay in order
doc = {"schema": 1, "nodes": [
{"id": "ccc000000003", "kind": "message", "role": "user", "content": "v1",
"ts": CLOCK, "parent": None, "edges": [], "superseded_by": None},
{"id": "ccc000000003", "kind": "message", "role": "user", "content": "v2",
"ts": CLOCK, "parent": None, "edges": [], "superseded_by": None},
]}
g = fresh("load_dup", json.dumps(doc))
case("load", "load-duplicate-ids-last-wins-order-keeps-both",
{"text": json.dumps(doc)}, {"graph": wire(g)})
# extra fields ride verbatim
doc = {"schema": 1, "nodes": [
{"id": "ddd000000004", "kind": "message", "role": "user", "content": "x",
"ts": CLOCK, "parent": None, "edges": [], "superseded_by": None,
"extra": {"deep": [1, 2.5, None, True]}, "π": "unicode key"},
]}
g = fresh("load_extra", json.dumps(doc))
case("load", "load-extra-fields-verbatim", {"text": json.dumps(doc)},
{"graph": wire(g)})
# non-dict doc -> the oracle's doc.get raises AttributeError
def _err_file(name: str, text: str) -> Path:
p = TMP / f"{name}.json"
p.write_text(text)
return p
_p = _err_file("load_list", "[1, 2]")
case("load", "load-list-doc-attributeerror", {"text": "[1, 2]"},
exc_of(mg.MessageGraph, path=_p))
_p = _err_file("load_str", '"hello"')
case("load", "load-string-doc-attributeerror", {"text": '"hello"'},
exc_of(mg.MessageGraph, path=_p))
_p = _err_file("load_int", "5")
case("load", "load-int-doc-attributeerror", {"text": "5"},
exc_of(mg.MessageGraph, path=_p))
# nodes present but not a list -> the for-loop raises TypeError
_p = _err_file("load_dictnodes", json.dumps({"schema": 1, "nodes": {"aaa": {}}}))
case("load", "load-dict-nodes-typeerror",
{"text": json.dumps({"schema": 1, "nodes": {"aaa": {}}})},
exc_of(mg.MessageGraph, path=_p))
# ══ add ══════════════════════════════════════════════════════════════════
# default parent on an empty graph: parent None, edges []
g = fresh("add_empty")
node = g.add("user", "hello")
case("add", "add-default-parent-empty-graph",
{"graph": {"nodes": {}, "order": []}, "id": node["id"],
"ts": node["ts"], "role": "user", "content": "hello", "kind": "message",
"parent": None, "edge": "parent"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# default parent chains to the head; edges stay [] (parent arg was None)
g = fresh("add_chain")
g.add("user", "q1")
before = wire(g)
node = g.add("assistant", "a1")
case("add", "add-default-parent-head-no-edge",
{"graph": before, "id": node["id"], "ts": node["ts"], "role": "assistant",
"content": "a1", "kind": "message", "parent": None, "edge": "parent"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# explicit parent + explicit edge
g = fresh("add_explicit")
a = g.add("user", "q")
before = wire(g)
node = g.add("assistant", "a", kind="reply", parent=a["id"], edge="reply_to")
case("add", "add-explicit-parent-edge",
{"graph": before, "id": node["id"], "ts": node["ts"], "role": "assistant",
"content": "a", "kind": "reply", "parent": a["id"], "edge": "reply_to"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# explicit parent, default edge "parent"
g = fresh("add_explicit_default_edge")
a = g.add("user", "q")
before = wire(g)
node = g.add("assistant", "a", parent=a["id"])
case("add", "add-explicit-parent-default-edge",
{"graph": before, "id": node["id"], "ts": node["ts"], "role": "assistant",
"content": "a", "kind": "message", "parent": a["id"], "edge": "parent"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# explicit None parent with a non-empty order: parent becomes the HEAD but
# edges stay [] (the arg was None — falsy)
g = fresh("add_none_parent")
a = g.add("user", "q")
before = wire(g)
node = g.add("assistant", "a", parent=None)
case("add", "add-explicit-none-parent-becomes-head",
{"graph": before, "id": node["id"], "ts": node["ts"], "role": "assistant",
"content": "a", "kind": "message", "parent": None, "edge": "parent"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# falsy parent "": stored as parent, but NO edge (falsy)
g = fresh("add_falsy_parent")
a = g.add("user", "q")
before = wire(g)
node = g.add("assistant", "a", parent="")
case("add", "add-falsy-empty-parent-no-edge",
{"graph": before, "id": node["id"], "ts": node["ts"], "role": "assistant",
"content": "a", "kind": "message", "parent": "", "edge": "parent"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# edge to a ghost parent is allowed (no referential check in add)
g = fresh("add_ghost_parent")
a = g.add("user", "q")
before = wire(g)
node = g.add("assistant", "a", parent="ghost0000001", edge="reply_to")
case("add", "add-ghost-parent-allowed",
{"graph": before, "id": node["id"], "ts": node["ts"], "role": "assistant",
"content": "a", "kind": "message", "parent": "ghost0000001",
"edge": "reply_to"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# unicode content
g = fresh("add_unicode")
node = g.add("user", "こんにちは ☕ héllo")
case("add", "add-unicode-content",
{"graph": {"nodes": {}, "order": []}, "id": node["id"], "ts": node["ts"],
"role": "user", "content": "こんにちは ☕ héllo", "kind": "message",
"parent": None, "edge": "parent"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# every kind
for k in ("message", "reply", "tool_call", "tool_result", "revision", "branch_head"):
g = fresh(f"add_kind_{k}")
node = g.add("user", f"content-{k}", kind=k)
case("add", f"add-kind-{k}",
{"graph": {"nodes": {}, "order": []}, "id": node["id"], "ts": node["ts"],
"role": "user", "content": f"content-{k}", "kind": k,
"parent": None, "edge": "parent"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# bad kind / bad edge (ValueError with the oracle's repr formatting)
g = fresh("add_bad_kind")
case("add", "add-unknown-kind-valueerror",
{"graph": {"nodes": {}, "order": []}, "id": "deadbeef_err", "ts": CLOCK,
"role": "user", "content": "x", "kind": "nonsense", "parent": None,
"edge": "parent"},
exc_of(g.add, "user", "x", kind="nonsense"))
case("add", "add-unknown-edge-valueerror",
{"graph": {"nodes": {}, "order": []}, "id": "deadbeef_err", "ts": CLOCK,
"role": "user", "content": "x", "kind": "message", "parent": None,
"edge": "nope"},
exc_of(g.add, "user", "x", edge="nope"))
# kind checked BEFORE edge
case("add", "add-both-bad-kind-wins",
{"graph": {"nodes": {}, "order": []}, "id": "deadbeef_err", "ts": CLOCK,
"role": "user", "content": "x", "kind": "nonsense", "parent": None,
"edge": "nope"},
exc_of(g.add, "user", "x", kind="nonsense", edge="nope"))
# ══ tool_call ════════════════════════════════════════════════════════════
g = fresh("tc_basic")
node = g.tool_call("read", {"path": "a.py"})
case("tool_call", "tool-call-basic",
{"graph": {"nodes": {}, "order": []}, "id": node["id"], "ts": node["ts"],
"content": json.dumps({"call": "read", "args": {"path": "a.py"}}),
"parent": None},
{"node": copy.deepcopy(node), "graph": wire(g)})
g = fresh("tc_parent")
a = g.add("assistant", "running")
before = wire(g)
node = g.tool_call("write", {"path": "b.py", "bytes": 3}, parent=a["id"])
case("tool_call", "tool-call-explicit-parent",
{"graph": before, "id": node["id"], "ts": node["ts"],
"content": json.dumps({"call": "write", "args": {"path": "b.py", "bytes": 3}}),
"parent": a["id"]},
{"node": copy.deepcopy(node), "graph": wire(g)})
g = fresh("tc_nested")
node = g.tool_call("search", {"q": "π", "opts": {"deep": [1, 2, {"k": None}],
"flag": True}, "n": 0.5})
case("tool_call", "tool-call-nested-args",
{"graph": {"nodes": {}, "order": []}, "id": node["id"], "ts": node["ts"],
"content": json.dumps({"call": "search",
"args": {"q": "π", "opts": {"deep": [1, 2, {"k": None}],
"flag": True}, "n": 0.5}}),
"parent": None},
{"node": copy.deepcopy(node), "graph": wire(g)})
# ══ tool_result ══════════════════════════════════════════════════════════
g = fresh("tr_ok")
c = g.tool_call("read", {"path": "a.py"})
before = wire(g)
node = g.tool_result(c["id"], "file contents")
case("tool_result", "tool-result-links-call",
{"graph": before, "id": node["id"], "ts": node["ts"], "call_id": c["id"],
"result": "file contents"},
{"node": copy.deepcopy(node), "graph": wire(g)})
# no kind check on the parent: any node can be a "call"
g = fresh("tr_any_node")
a = g.add("user", "not a call")
before = wire(g)
node = g.tool_result(a["id"], "result anyway")
case("tool_result", "tool-result-non-call-parent-allowed",
{"graph": before, "id": node["id"], "ts": node["ts"], "call_id": a["id"],
"result": "result anyway"},
{"node": copy.deepcopy(node), "graph": wire(g)})
g = fresh("tr_missing")
case("tool_result", "tool-result-missing-keyerror",
{"graph": {"nodes": {}, "order": []}, "id": "deadbeef_err", "ts": CLOCK,
"call_id": "ghost0000001", "result": "x"},
exc_of(g.tool_result, "ghost0000001", "x"))
# ══ revise ═══════════════════════════════════════════════════════════════
# root revision: old.get("parent") is None -> add substitutes the HEAD,
# which at that moment is the old node itself (quirk, frozen verbatim)
g = fresh("rev_root")
a = g.add("assistant", "v1")
before = wire(g)
rev = g.revise(a["id"], "v2")
case("revise", "revise-root-head-parent-quirk",
{"graph": before, "node_id": a["id"], "new_content": "v2",
"id": rev["id"], "ts": rev["ts"]},
{"node": copy.deepcopy(rev), "graph": wire(g)})
# child revision: keeps the old parent, edges revises + supersedes
g = fresh("rev_child")
a = g.add("user", "q")
b = g.add("assistant", "v1", parent=a["id"])
before = wire(g)
rev = g.revise(b["id"], "v2")
case("revise", "revise-child-keeps-parent",
{"graph": before, "node_id": b["id"], "new_content": "v2",
"id": rev["id"], "ts": rev["ts"]},
{"node": copy.deepcopy(rev), "graph": wire(g)})
# chained revision: revise the revision
g = fresh("rev_chain")
a = g.add("assistant", "v1")
r1 = g.revise(a["id"], "v2")
before = wire(g)
r2 = g.revise(r1["id"], "v3")
case("revise", "revise-of-revision",
{"graph": before, "node_id": r1["id"], "new_content": "v3",
"id": r2["id"], "ts": r2["ts"]},
{"node": copy.deepcopy(r2), "graph": wire(g)})
g = fresh("rev_missing")
case("revise", "revise-missing-keyerror",
{"graph": {"nodes": {}, "order": []}, "node_id": "ghost0000001",
"new_content": "x", "id": "deadbeef_err", "ts": CLOCK},
exc_of(g.revise, "ghost0000001", "x"))
# ══ branch ═══════════════════════════════════════════════════════════════
g = fresh("branch_ok")
a = g.add("user", "root")
g.add("assistant", "answer", parent=a["id"])
before = wire(g)
head = g.branch(a["id"])
case("branch", "branch-from-message",
{"graph": before, "node_id": a["id"], "id": head["id"], "ts": head["ts"]},
{"node": copy.deepcopy(head), "graph": wire(g)})
g = fresh("branch_missing")
case("branch", "branch-missing-keyerror",
{"graph": {"nodes": {}, "order": []}, "node_id": "ghost0000001",
"id": "deadbeef_err", "ts": CLOCK},
exc_of(g.branch, "ghost0000001"))
# ══ lineage ══════════════════════════════════════════════════════════════
g = fresh("lin_empty")
case("lineage", "lineage-empty-graph", {"graph": wire(g), "node_id": None},
{"result": []})
# default head: full chain oldest-first
g = fresh("lin_chain")
a = g.add("user", "q")
b = g.add("assistant", "a1", parent=a["id"])
c = g.add("assistant", "a2", parent=b["id"])
case("lineage", "lineage-default-head-full-chain",
{"graph": wire(g), "node_id": None},
{"result": [copy.deepcopy(n) for n in g.lineage()]})
assert [n["id"] for n in g.lineage()] == [a["id"], b["id"], c["id"]]
# explicit id: subchain
case("lineage", "lineage-explicit-id-subchain",
{"graph": wire(g), "node_id": b["id"]},
{"result": [copy.deepcopy(n) for n in g.lineage(b["id"])]})
assert [n["id"] for n in g.lineage(b["id"])] == [a["id"], b["id"]]
# superseded nodes yield to their revision; unrelated branches excluded
g = fresh("lin_superseded")
a = g.add("user", "q")
b = g.add("assistant", "v1", parent=a["id"])
other = g.add("user", "unrelated")
rev = g.revise(b["id"], "v2")
case("lineage", "lineage-skips-superseded-default-head",
{"graph": wire(g), "node_id": None},
{"result": [copy.deepcopy(n) for n in g.lineage()]})
assert [n["id"] for n in g.lineage()] == [a["id"], rev["id"]]
case("lineage", "lineage-from-root-excludes-other-branch",
{"graph": wire(g), "node_id": a["id"]},
{"result": [copy.deepcopy(n) for n in g.lineage(a["id"])]})
assert [n["id"] for n in g.lineage(a["id"])] == [a["id"]]
# parent chain into a ghost: walk breaks quietly
g = fresh("lin_ghost_parent")
a = g.add("user", "q")
g.nodes[a["id"]]["parent"] = "ghost0000001"
case("lineage", "lineage-ghost-parent-breaks",
{"graph": wire(g), "node_id": a["id"]},
{"result": [copy.deepcopy(n) for n in g.lineage(a["id"])]})
assert [n["id"] for n in g.lineage(a["id"])] == [a["id"]]
# superseded_by pointing OUTSIDE the graph: node included (guard fails)
g = fresh("lin_ghost_superseded")
a = g.add("user", "q")
g.nodes[a["id"]]["superseded_by"] = "ghost0000001"
case("lineage", "lineage-ghost-superseded-by-included",
{"graph": wire(g), "node_id": a["id"]},
{"result": [copy.deepcopy(n) for n in g.lineage(a["id"])]})
assert [n["id"] for n in g.lineage(a["id"])] == [a["id"]]
# unknown id: empty
g = fresh("lin_unknown")
case("lineage", "lineage-unknown-id-empty",
{"graph": wire(g), "node_id": "ghost0000001"}, {"result": []})
# parent cycle: the seen-set guard stops the walk
doc = {"schema": 1, "nodes": [
{"id": "aaa000000001", "kind": "message", "role": "user", "content": "a",
"ts": CLOCK, "parent": "bbb000000002", "edges": [], "superseded_by": None},
{"id": "bbb000000002", "kind": "message", "role": "user", "content": "b",
"ts": CLOCK, "parent": "aaa000000001", "edges": [], "superseded_by": None},
]}
g = fresh("lin_cycle", json.dumps(doc))
case("lineage", "lineage-parent-cycle-guarded",
{"graph": wire(g), "node_id": "aaa000000001"},
{"result": [copy.deepcopy(n) for n in g.lineage("aaa000000001")]})
assert [n["id"] for n in g.lineage("aaa000000001")] == ["bbb000000002",
"aaa000000001"]
# ══ history ══════════════════════════════════════════════════════════════
# single node: no supersedes edge -> just itself
g = fresh("hist_single")
a = g.add("assistant", "v1")
case("history", "history-single-node",
{"graph": wire(g), "node_id": a["id"]},
{"result": [copy.deepcopy(n) for n in g.history(a["id"])]})
assert [n["content"] for n in g.history(a["id"])] == ["v1"]
# full chain oldest-first, asked from the tip AND from the middle
g = fresh("hist_chain")
a = g.add("assistant", "v1")
r1 = g.revise(a["id"], "v2")
r2 = g.revise(r1["id"], "v3")
case("history", "history-chain-from-tip",
{"graph": wire(g), "node_id": r2["id"]},
{"result": [copy.deepcopy(n) for n in g.history(r2["id"])]})
assert [n["content"] for n in g.history(r2["id"])] == ["v1", "v2", "v3"]
case("history", "history-chain-from-middle-full-trail",
{"graph": wire(g), "node_id": r1["id"]},
{"result": [copy.deepcopy(n) for n in g.history(r1["id"])]})
assert [n["content"] for n in g.history(r1["id"])] == ["v1", "v2", "v3"]
# unknown id: empty trail
g = fresh("hist_missing")
case("history", "history-unknown-id-empty",
{"graph": wire(g), "node_id": "ghost0000001"}, {"result": []})
# revision whose superseded_by target is a ghost: trail stops there
g = fresh("hist_ghost_target")
a = g.add("assistant", "v1")
r1 = g.revise(a["id"], "v2")
g.nodes[r1["id"]]["superseded_by"] = "ghost0000001"
case("history", "history-ghost-superseded-by-stops",
{"graph": wire(g), "node_id": a["id"]},
{"result": [copy.deepcopy(n) for n in g.history(a["id"])]})
assert [n["content"] for n in g.history(a["id"])] == ["v1", "v2"]
# ══ migrate_flat ═════════════════════════════════════════════════════════
def mig_ids(g, n):
"""The ids the oracle's migrate consumed (the last n arrivals)."""
return list(g.order[len(g.order) - n:]) if n else []
flat = [{"role": "user", "content": "q1"}, {"role": "assistant", "content": "a1"}]
g = mg.MessageGraph.migrate_flat(copy.deepcopy(flat), path=TMP / "mig_basic.json")
case("migrate_flat", "migrate-basic-chain",
{"text": None, "transcript": copy.deepcopy(flat),
"ids": mig_ids(g, 2), "tss": [CLOCK, CLOCK]},
{"graph": wire(g)})
assert [g.nodes[i]["parent"] for i in g.order] == [None, g.order[0]]
# empty transcript
g = mg.MessageGraph.migrate_flat([], path=TMP / "mig_empty.json")
case("migrate_flat", "migrate-empty-transcript",
{"text": None, "transcript": [], "ids": [], "tss": []}, {"graph": wire(g)})
# missing role/content keys -> system / "" defaults
flat = [{"role": "user"}, {}, {"content": "no role"}]
g = mg.MessageGraph.migrate_flat(copy.deepcopy(flat), path=TMP / "mig_defaults.json")
case("migrate_flat", "migrate-missing-keys-default-system-empty",
{"text": None, "transcript": copy.deepcopy(flat), "ids": mig_ids(g, 3),
"tss": [CLOCK] * 3},
{"graph": wire(g)})
assert [g.nodes[i]["role"] for i in g.order] == ["user", "system", "system"]
assert [g.nodes[i]["content"] for i in g.order] == ["", "", "no role"]
# existing file seeds the graph first, migration appends after
seed_doc = {"schema": 1, "nodes": [
{"id": "eee000000005", "kind": "message", "role": "user", "content": "old",
"ts": CLOCK, "parent": None, "edges": [], "superseded_by": None},
]}
_seed = _err_file("mig_seed", json.dumps(seed_doc))
g = mg.MessageGraph.migrate_flat([{"role": "user", "content": "new"}],
path=_seed)
case("migrate_flat", "migrate-existing-file-seed",
{"text": json.dumps(seed_doc), "transcript": [{"role": "user", "content": "new"}],
"ids": mig_ids(g, 1), "tss": [CLOCK]},
{"graph": wire(g)})
# ══ cue ══════════════════════════════════════════════════════════════════
g = fresh("cue_plain")
a = g.add("assistant", "v1")
plain = copy.deepcopy(g.nodes[a["id"]])
case("cue", "cue-plain-node", {"node": plain}, {"result": ""})
assert mg.MessageGraph.cue(plain) == ""
rev = g.revise(a["id"], "v2")
case("cue", "cue-revision-replaces",
{"node": copy.deepcopy(g.nodes[rev["id"]])},
{"result": f"replaces {a['id'][:8]}"})
assert mg.MessageGraph.cue(g.nodes[rev["id"]]) == f"replaces {a['id'][:8]}"
case("cue", "cue-superseded-by",
{"node": copy.deepcopy(g.nodes[a["id"]])},
{"result": f"superseded by {rev['id'][:8]}"})
assert mg.MessageGraph.cue(g.nodes[a["id"]]) == f"superseded by {rev['id'][:8]}"
g = fresh("cue_branch")
a = g.add("user", "root")
head = g.branch(a["id"])
case("cue", "cue-branch-head",
{"node": copy.deepcopy(g.nodes[head["id"]])},
{"result": f"branched from {a['id'][:8]}"})
assert mg.MessageGraph.cue(g.nodes[head["id"]]) == f"branched from {a['id'][:8]}"
# falsy superseded_by "" is NOT a cue
node = copy.deepcopy(plain)
node["superseded_by"] = ""
case("cue", "cue-falsy-superseded-by-empty", {"node": node}, {"result": ""})
assert mg.MessageGraph.cue(node) == ""
# first supersedes/branches_from edge in iteration order wins
node = copy.deepcopy(plain)
node["edges"] = [{"type": "parent", "to": "aaa000000001"},
{"type": "supersedes", "to": "bbb000000002"}]
case("cue", "cue-first-supersedes-edge-wins", {"node": node},
{"result": "replaces bbb00000"})
assert mg.MessageGraph.cue(node) == "replaces bbb00000"
node2 = copy.deepcopy(plain)
node2["edges"] = [{"type": "branches_from", "to": "ccc000000003"},
{"type": "supersedes", "to": "bbb000000002"}]
case("cue", "cue-branches-from-beats-later-supersedes", {"node": node2},
{"result": "branched from ccc00000"})
assert mg.MessageGraph.cue(node2) == "branched from ccc00000"
# ══ save ═════════════════════════════════════════════════════════════════
# basic save: doc shape + exact file bytes (json.dumps indent=2)
g = fresh("save_basic")
a = g.add("user", "persist me")
g.save()
doc = {"schema": mg.SCHEMA, "nodes": [g.nodes[i] for i in g.order]}
case("save", "save-basic-doc-and-bytes",
{"graph": {"nodes": {a["id"]: copy.deepcopy(g.nodes[a["id"]])},
"order": [a["id"]]}},
{"doc": copy.deepcopy(doc), "file": (TMP / "save_basic.json").read_text()})
assert (TMP / "save_basic.json").read_text() == json.dumps(doc, indent=2)
# save creates missing parent dirs
deep = TMP / "deep" / "nested" / "dir" / "g.json"
g2 = mg.MessageGraph(path=deep)
g2.add("user", "deep")
g2.save()
doc2 = {"schema": mg.SCHEMA, "nodes": [g2.nodes[i] for i in g2.order]}
case("save", "save-creates-parent-dirs",
{"graph": wire(g2)}, {"doc": copy.deepcopy(doc2), "file": deep.read_text()})
assert deep.exists()
# round trip: save -> fresh load reproduces nodes AND order
g3 = mg.MessageGraph(path=TMP / "save_basic.json")
case("load", "load-save-round-trip",
{"text": (TMP / "save_basic.json").read_text()}, {"graph": wire(g3)})
assert g3.order == g.order
assert g3.nodes[a["id"]]["content"] == "persist me"
# ══ oracle health (the repo's own battery must pass on the oracle) ══════
import unittest # noqa: E402
suite = unittest.defaultTestLoader.discover(
str(Path(__file__).resolve().parents[1] / "tests"),
pattern="test_message_graph.py")
runner = unittest.TextTestRunner(verbosity=0)
if not runner.run(suite).wasSuccessful():
raise AssertionError("oracle regression: test_message_graph.py failed")
# ══ write ════════════════════════════════════════════════════════════════
truth = {
"frozen": "message_graph load/save/add/tool_call/tool_result/revise/"
"branch/lineage/history/migrate_flat/cue",
"oracle": "src/rune/message_graph.py",
"note": "Graph wire: {nodes: {id: node} insertion-ordered, order: [ids]}. "
"Nondeterminism seams (uuid4 hex id, _now timestamp) are facade "
"arguments — frozen CLOCK and sequential deadbeefNNNN ids replay "
"verbatim. save() builds the doc in the kernel; the file write "
"(mkdir + json.dumps indent=2) and tool_call's json.dumps content "
"stay in the facade (serialization isomorphism).",
}
for gname in GROUPS:
truth[gname] = groups[gname]
out = Path(os.environ.get("MG_TRUTH_OUT", "/tmp/mg_truth.json"))
out.write_text(json.dumps(truth, indent=1, ensure_ascii=False) + "\n")
counts = ", ".join(f"{len(groups[g])} {g}" for g in GROUPS)
print(f"froze {sum(len(v) for v in groups.values())} cases -> {out}: {counts}")
src/rune/message_graph.py
+113 -5
View File
@@ -32,6 +32,24 @@ KINDS = {"message", "reply", "tool_call", "tool_result", "revision", "branch_hea
EDGE_TYPES = {"parent", "reply_to", "calls", "results_in", "revises", "supersedes", "branches_from"}
def _mg_core():
"""Dichotomy wave #2479: Rust core first (RUNE_MG_CORE=python pins the
Python fallback — parity harnesses). The kernel owns the graph state
machine (load parsing + schema gate, node construction, revise/branch
wiring, lineage/history walks, flat migration, cues); the uuid/clock
seams, the save FILE IO and the json.dumps renderings stay here."""
if os.environ.get("RUNE_MG_CORE", "auto") == "python":
return None
try:
from rune import mg_loader
if mg_loader.available():
return mg_loader
except ImportError:
pass
return None
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
@@ -48,6 +66,18 @@ class MessageGraph:
# -- persistence ----------------------------------------------------
def _load(self) -> None:
core = _mg_core()
if core is not None:
text = None
if self.path.exists():
try:
text = self.path.read_text()
except OSError:
text = None
state = core.load(text)
self.nodes = state["nodes"]
self.order = state["order"]
return
if not self.path.exists():
return
try:
@@ -61,8 +91,13 @@ class MessageGraph:
self.order.append(n["id"])
def save(self) -> None:
core = _mg_core()
if core is not None:
doc = core.save({"nodes": self.nodes, "order": self.order})
else:
doc = {"schema": SCHEMA, "nodes": [self.nodes[i] for i in self.order]}
self.path.parent.mkdir(parents=True, exist_ok=True)
self.path.write_text(json.dumps({"schema": SCHEMA, "nodes": [self.nodes[i] for i in self.order]}, indent=2))
self.path.write_text(json.dumps(doc, indent=2))
# -- append ---------------------------------------------------------
def add(
@@ -73,6 +108,15 @@ class MessageGraph:
raise ValueError(f"unknown kind {kind!r}")
if edge not in EDGE_TYPES:
raise ValueError(f"unknown edge {edge!r}")
core = _mg_core()
if core is not None:
out = core.add(
{"nodes": self.nodes, "order": self.order},
uuid.uuid4().hex[:12], _now(), role, content, kind, parent, edge,
)
self.nodes = out["graph"]["nodes"]
self.order = out["graph"]["order"]
return out["node"]
node = {
"id": uuid.uuid4().hex[:12],
"kind": kind,
@@ -88,21 +132,49 @@ class MessageGraph:
return node
def tool_call(self, name: str, args: dict, parent: str | None = None) -> dict:
core = _mg_core()
if core is not None:
out = core.tool_call(
{"nodes": self.nodes, "order": self.order},
uuid.uuid4().hex[:12], _now(),
json.dumps({"call": name, "args": args}), parent,
)
self.nodes = out["graph"]["nodes"]
self.order = out["graph"]["order"]
return out["node"]
return self.add("tool", json.dumps({"call": name, "args": args}), kind="tool_call", parent=parent, edge="calls")
def tool_result(self, call_id: str, result: str) -> dict:
call = self.nodes.get(call_id)
if call is None:
if call_id not in self.nodes:
raise KeyError(f"no such call {call_id!r}")
core = _mg_core()
if core is not None:
out = core.tool_result(
{"nodes": self.nodes, "order": self.order},
uuid.uuid4().hex[:12], _now(), call_id, result,
)
self.nodes = out["graph"]["nodes"]
self.order = out["graph"]["order"]
return out["node"]
call = self.nodes[call_id]
return self.add("tool", result, kind="tool_result", parent=call_id, edge="results_in")
# -- revision / branch ----------------------------------------------
def revise(self, node_id: str, new_content: str) -> dict:
"""Regenerate an answer without losing provenance: the old node is
marked superseded; the revision carries the supersedes edge."""
old = self.nodes.get(node_id)
if old is None:
if node_id not in self.nodes:
raise KeyError(f"no such message {node_id!r}")
core = _mg_core()
if core is not None:
out = core.revise(
{"nodes": self.nodes, "order": self.order},
node_id, new_content, uuid.uuid4().hex[:12], _now(),
)
self.nodes = out["graph"]["nodes"]
self.order = out["graph"]["order"]
return out["node"]
old = self.nodes[node_id]
rev = self.add(old["role"], new_content, kind="revision", parent=old.get("parent"), edge="revises")
rev["edges"].append({"type": "supersedes", "to": node_id})
old["superseded_by"] = rev["id"]
@@ -112,6 +184,15 @@ class MessageGraph:
"""Start a new conversation branch off any message."""
if node_id not in self.nodes:
raise KeyError(f"no such message {node_id!r}")
core = _mg_core()
if core is not None:
out = core.branch(
{"nodes": self.nodes, "order": self.order},
node_id, uuid.uuid4().hex[:12], _now(),
)
self.nodes = out["graph"]["nodes"]
self.order = out["graph"]["order"]
return out["node"]
return self.add("system", f"branched from {node_id}", kind="branch_head", parent=node_id, edge="branches_from")
# -- lineage context -------------------------------------------------
@@ -123,6 +204,9 @@ class MessageGraph:
node_id = self.order[-1] if self.order else None
if node_id is None:
return []
core = _mg_core()
if core is not None:
return core.lineage({"nodes": self.nodes, "order": self.order}, node_id)
chain: list[dict] = []
seen = set()
cur = node_id
@@ -142,6 +226,9 @@ class MessageGraph:
def history(self, node_id: str) -> list[dict]:
"""Every revision of a message, oldest first (the provenance trail)."""
core = _mg_core()
if core is not None:
return core.history({"nodes": self.nodes, "order": self.order}, node_id)
trail = []
# find the root: walk supersedes edges backwards
cur = node_id
@@ -168,6 +255,24 @@ class MessageGraph:
"""Schema migration: the flat {role, content} transcript becomes a
chained graph (arrival order, parent links). Compatibility preserved —
no conversation is lost in the upgrade."""
core = _mg_core()
if core is not None:
g = cls.__new__(cls)
g.path = Path(path) if path else GRAPH_PATH
text = None
if g.path.exists():
try:
text = g.path.read_text()
except OSError:
text = None
state = core.migrate_flat(
text, list(transcript),
[uuid.uuid4().hex[:12] for _ in transcript],
[_now() for _ in transcript],
)
g.nodes = state["nodes"]
g.order = state["order"]
return g
g = cls(path=path)
for m in transcript:
g.add(m.get("role", "system"), m.get("content", ""), kind="message")
@@ -177,6 +282,9 @@ class MessageGraph:
@staticmethod
def cue(node: dict) -> str:
"""One quiet relationship line per node (no visual clutter)."""
core = _mg_core()
if core is not None:
return core.cue(node)
for e in node.get("edges", []):
if e["type"] == "supersedes":
return f"replaces {e['to'][:8]}"
src/rune/mg_loader.py
+152
View File
@@ -0,0 +1,152 @@
#!/usr/bin/env python3
"""ctypes loader for the Rust message_graph core (dichotomy wave, #2479
tranche). Mirrors broker_loader.py's contract:
FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
* mg_load({"text": str|null}) -> graph wire {"nodes", "order"}
* mg_save({"graph"}) -> save doc {"schema", "nodes"}
* mg_add({graph, id, ts, role, content, kind, parent, edge}) -> {graph, node}
* mg_tool_call({graph, id, ts, content, parent}) -> {graph, node}
* mg_tool_result({graph, id, ts, call_id, result}) -> {graph, node}
* mg_revise({graph, node_id, new_content, id, ts}) -> {graph, node}
* mg_branch({graph, node_id, id, ts}) -> {graph, node}
* mg_lineage({graph, node_id}) -> [nodes]
* mg_history({graph, node_id}) -> [nodes]
* mg_migrate_flat({text, transcript, ids, tss}) -> graph wire
* mg_cue({"node"}) -> cue string
Graph wire: {"nodes": {id: node} (insertion-ordered), "order": [ids]}.
Errors ride the envelope typed — {"ok": false, "err": msg, "type":
"ValueError"|"KeyError"|"AttributeError"|"TypeError"} — and the loader
raises the MATCHING Python exception (the oracle raises those; the parity
harness compares exception types and messages, so the kernel's guard
clauses must round-trip as the same Python error the oracle raises).
Free: capability_scoring's cs_free_string (shared process). Override the
.so with MG_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" / "message_graph"
_LIB_PATH = Path(
os.environ.get("MG_RUST_LIB", _CRATE_DIR / "target" / "release" / "libmessage_graph.so")
)
_ERR_TYPES = {
"ValueError": ValueError,
"KeyError": KeyError,
"AttributeError": AttributeError,
"TypeError": TypeError,
}
_lib = None
class MessageGraphCoreUnavailable(FileNotFoundError):
"""The Rust message-graph 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 MessageGraphCoreUnavailable(str(_LIB_PATH))
lib = ctypes.CDLL(str(_LIB_PATH))
for sym in ("mg_load", "mg_save", "mg_add", "mg_tool_call", "mg_tool_result",
"mg_revise", "mg_branch", "mg_lineage", "mg_history",
"mg_migrate_flat", "mg_cue"):
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.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"):
etype = doc.get("type", "RuntimeError")
raise _ERR_TYPES.get(etype, RuntimeError)(doc.get("err", f"{sym} failed"))
return doc["value"]
def load(text: str | None) -> dict:
return _call("mg_load", {"text": text})
def save(graph: dict) -> dict:
return _call("mg_save", {"graph": graph})
def add(graph: dict, node_id: str, ts: str, role, content, kind: str,
parent: str | None, edge: str) -> dict:
return _call("mg_add", {"graph": graph, "id": node_id, "ts": ts, "role": role,
"content": content, "kind": kind, "parent": parent,
"edge": edge})
def tool_call(graph: dict, node_id: str, ts: str, content: str,
parent: str | None) -> dict:
return _call("mg_tool_call", {"graph": graph, "id": node_id, "ts": ts,
"content": content, "parent": parent})
def tool_result(graph: dict, node_id: str, ts: str, call_id: str, result) -> dict:
return _call("mg_tool_result", {"graph": graph, "id": node_id, "ts": ts,
"call_id": call_id, "result": result})
def revise(graph: dict, node_id: str, new_content, rev_id: str, ts: str) -> dict:
return _call("mg_revise", {"graph": graph, "node_id": node_id,
"new_content": new_content, "id": rev_id, "ts": ts})
def branch(graph: dict, node_id: str, head_id: str, ts: str) -> dict:
return _call("mg_branch", {"graph": graph, "node_id": node_id,
"id": head_id, "ts": ts})
def lineage(graph: dict, node_id: str | None) -> list:
return _call("mg_lineage", {"graph": graph, "node_id": node_id})
def history(graph: dict, node_id: str) -> list:
return _call("mg_history", {"graph": graph, "node_id": node_id})
def migrate_flat(text: str | None, transcript: list, ids: list, tss: list) -> dict:
return _call("mg_migrate_flat", {"text": text, "transcript": transcript,
"ids": ids, "tss": tss})
def cue(node: dict) -> str:
return _call("mg_cue", {"node": node})
src/rust/message_graph/Cargo.lock
Generated
+137
View File
@@ -0,0 +1,137 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 4
[[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"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
[[package]]
name = "memchr"
version = "2.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
[[package]]
name = "message_graph"
version = "0.1.0"
dependencies = [
"serde",
"serde_json",
]
[[package]]
name = "proc-macro2"
version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
dependencies = [
"unicode-ident",
]
[[package]]
name = "quote"
version = "1.0.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
dependencies = [
"proc-macro2",
]
[[package]]
name = "serde"
version = "1.0.229"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
dependencies = [
"serde_core",
"serde_derive",
]
[[package]]
name = "serde_core"
version = "1.0.229"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
version = "1.0.229"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "serde_json"
version = "1.0.154"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6"
dependencies = [
"foldhash",
"indexmap",
"itoa",
"memchr",
"serde",
"serde_core",
"zmij",
]
[[package]]
name = "syn"
version = "3.0.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "01016da373cd8f7ef12624f796309f5c31ba8d646dd08856c02cd741d823c622"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]]
name = "unicode-ident"
version = "1.0.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
[[package]]
name = "zmij"
version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"
src/rust/message_graph/Cargo.toml
+11
View File
@@ -0,0 +1,11 @@
[package]
name = "message_graph"
version = "0.1.0"
edition = "2021"
[lib]
crate-type = ["rlib", "cdylib", "staticlib"]
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = { version = "1", features = ["preserve_order", "float_roundtrip"] }
src/rust/message_graph/src/branch_battery.rs
+275
View File
@@ -0,0 +1,275 @@
//! branch_battery — replay the FROZEN oracle truth VERBATIM
//! (frozen_mg_truth.json, generated by scripts/gen_message_graph_truth.py
//! against the LIVE Python oracle with RUNE_MG_CORE=python pinned).
//!
//! Every case: rebuild the recorded input (graph wire + seams id/ts),
//! call the kernel, compare the result — and the post-op graph — against
//! the oracle's recorded expectation. Exception cases compare the typed
//! envelope (ValueError/KeyError/AttributeError/TypeError + message).
use crate::{graph_from_wire, MessageGraph};
use serde_json::{json, Value};
const TRUTH: &str = include_str!("frozen_mg_truth.json");
fn cases(group: &str) -> Vec<Value> {
let truth: Value = serde_json::from_str(TRUTH).expect("frozen truth parses");
truth[group]
.as_array()
.unwrap_or_else(|| panic!("truth group {group} missing"))
.clone()
}
fn check_err(case_name: &str, want: &Value, got: &crate::MgErr) -> Result<(), String> {
let exc = &want["expected"]["exception"];
let (want_type, want_msg) = (
exc["type"].as_str().unwrap_or(""),
exc["message"].as_str().unwrap_or(""),
);
// str(KeyError) reprs its argument — the frozen message carries the
// repr quotes; strip them back to the raw message
let want_msg = if want_type == "KeyError" {
want_msg
.strip_prefix('"')
.and_then(|m| m.strip_suffix('"'))
.unwrap_or(want_msg)
.replace("\\'", "'")
} else {
want_msg.to_string()
};
if got.kind.py_name() != want_type || got.message != want_msg {
return Err(format!(
"[{case_name}] exception mismatch: got {}({:?}), want {want_type}({want_msg:?})",
got.kind.py_name(),
got.message
));
}
Ok(())
}
fn expect_field(
case_name: &str,
field: &str,
got: &Value,
want: &Value,
) -> Result<(), String> {
if got != want {
return Err(format!(
"[{case_name}] {field} mismatch:\n got: {got}\n want: {want}"
));
}
Ok(())
}
fn replay(group: &str, c: &Value) -> Result<(), String> {
let name = c["name"].as_str().unwrap_or("?");
let case_name = format!("{group}/{name}");
let input = &c["input"];
let expected = &c["expected"];
match group {
"load" => match MessageGraph::load(input.get("text").and_then(Value::as_str)) {
Err(e) => check_err(&case_name, c, &e),
Ok(g) => {
if expected.get("exception").is_some() {
return Err(format!("[{case_name}] expected exception, got ok"));
}
expect_field(&case_name, "graph", &g.to_wire(), &expected["graph"])
}
},
"save" => {
let g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
expect_field(&case_name, "doc", &g.save(), &expected["doc"])
}
"add" => {
let mut g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
match g.add(
input["id"].as_str().unwrap_or_default(),
input["ts"].as_str().unwrap_or_default(),
input["role"].clone(),
input["content"].clone(),
input["kind"].as_str().unwrap_or_default(),
input["parent"].as_str(),
input["edge"].as_str().unwrap_or_default(),
) {
Err(e) => check_err(&case_name, c, &e),
Ok(node) => {
expect_field(&case_name, "node", &node, &expected["node"])?;
expect_field(&case_name, "graph", &g.to_wire(), &expected["graph"])
}
}
}
"tool_call" => {
let mut g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
match g.tool_call(
input["id"].as_str().unwrap_or_default(),
input["ts"].as_str().unwrap_or_default(),
input["content"].as_str().unwrap_or_default(),
input["parent"].as_str(),
) {
Err(e) => check_err(&case_name, c, &e),
Ok(node) => {
expect_field(&case_name, "node", &node, &expected["node"])?;
expect_field(&case_name, "graph", &g.to_wire(), &expected["graph"])
}
}
}
"tool_result" => {
let mut g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
match g.tool_result(
input["id"].as_str().unwrap_or_default(),
input["ts"].as_str().unwrap_or_default(),
input["call_id"].as_str().unwrap_or_default(),
input["result"].clone(),
) {
Err(e) => check_err(&case_name, c, &e),
Ok(node) => {
expect_field(&case_name, "node", &node, &expected["node"])?;
expect_field(&case_name, "graph", &g.to_wire(), &expected["graph"])
}
}
}
"revise" => {
let mut g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
match g.revise(
input["node_id"].as_str().unwrap_or_default(),
input["new_content"].clone(),
input["id"].as_str().unwrap_or_default(),
input["ts"].as_str().unwrap_or_default(),
) {
Err(e) => check_err(&case_name, c, &e),
Ok(node) => {
expect_field(&case_name, "node", &node, &expected["node"])?;
expect_field(&case_name, "graph", &g.to_wire(), &expected["graph"])
}
}
}
"branch" => {
let mut g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
match g.branch(
input["node_id"].as_str().unwrap_or_default(),
input["id"].as_str().unwrap_or_default(),
input["ts"].as_str().unwrap_or_default(),
) {
Err(e) => check_err(&case_name, c, &e),
Ok(node) => {
expect_field(&case_name, "node", &node, &expected["node"])?;
expect_field(&case_name, "graph", &g.to_wire(), &expected["graph"])
}
}
}
"lineage" => {
let g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
let got = Value::Array(g.lineage(input["node_id"].as_str()));
expect_field(&case_name, "result", &got, &expected["result"])
}
"history" => {
let g = graph_from_wire(input).map_err(|e| format!("[{case_name}] {e:?}"))?;
let got = Value::Array(g.history(input["node_id"].as_str().unwrap_or_default()));
expect_field(&case_name, "result", &got, &expected["result"])
}
"migrate_flat" => {
let text = input.get("text").and_then(Value::as_str);
let transcript: Vec<Value> = input["transcript"].as_array().cloned().unwrap_or_default();
let ids: Vec<String> = input["ids"]
.as_array()
.map(|a| a.iter().map(|v| v.as_str().unwrap_or_default().to_string()).collect())
.unwrap_or_default();
let tss: Vec<String> = input["tss"]
.as_array()
.map(|a| a.iter().map(|v| v.as_str().unwrap_or_default().to_string()).collect())
.unwrap_or_default();
match MessageGraph::migrate_flat(text, &transcript, &ids, &tss) {
Err(e) => check_err(&case_name, c, &e),
Ok(g) => expect_field(&case_name, "graph", &g.to_wire(), &expected["graph"]),
}
}
"cue" => {
let got = json!(MessageGraph::cue(&input["node"]));
expect_field(&case_name, "result", &got, &expected["result"])
}
other => Err(format!("[{other}] unknown truth group")),
}
}
fn run_group(group: &str) -> usize {
let cases = cases(group);
let fails: Vec<String> = cases
.iter()
.filter_map(|c| replay(group, c).err())
.collect();
assert!(
fails.is_empty(),
"{} {group} failures (of {}):\n{}",
fails.len(),
cases.len(),
fails.join("\n---\n")
);
cases.len()
}
#[test]
fn battery_load() {
assert_eq!(run_group("load"), 12);
}
#[test]
fn battery_save() {
assert_eq!(run_group("save"), 2);
}
#[test]
fn battery_add() {
assert_eq!(run_group("add"), 17);
}
#[test]
fn battery_tool_call() {
assert_eq!(run_group("tool_call"), 3);
}
#[test]
fn battery_tool_result() {
assert_eq!(run_group("tool_result"), 3);
}
#[test]
fn battery_revise() {
assert_eq!(run_group("revise"), 4);
}
#[test]
fn battery_branch() {
assert_eq!(run_group("branch"), 2);
}
#[test]
fn battery_lineage() {
assert_eq!(run_group("lineage"), 9);
}
#[test]
fn battery_history() {
assert_eq!(run_group("history"), 5);
}
#[test]
fn battery_migrate_flat() {
assert_eq!(run_group("migrate_flat"), 4);
}
#[test]
fn battery_cue() {
assert_eq!(run_group("cue"), 7);
}
#[test]
fn battery_total_is_68() {
let total: usize = [
"load", "save", "add", "tool_call", "tool_result", "revise", "branch",
"lineage", "history", "migrate_flat", "cue",
]
.iter()
.map(|g| run_group(g))
.sum();
assert_eq!(total, 68);
}
src/rust/message_graph/src/frozen_mg_truth.json
+3174
View File
File diff suppressed because it is too large. Load diff
src/rust/message_graph/src/lib.rs
+716
View File
@@ -0,0 +1,716 @@
//! message_graph — #13: the TUI transcript as a connected graph.
//!
//! FULL parity port of src/rune/message_graph.py (dichotomy wave, #2479
//! tranche). Every message gets a stable ID and a typed kind; relationships
//! are explicit edges. Node: {id, kind, role, content, ts, parent,
//! edges:[{type, to}], superseded_by}. Kinds: message | reply | tool_call |
//! tool_result | revision | branch_head. Edge types: parent | reply_to |
//! calls | results_in | revises | supersedes | branches_from.
//!
//! Nodes ride as opaque `serde_json::Value` objects: loaded documents may
//! carry arbitrary extra fields and they must survive verbatim (the oracle
//! ingests file nodes dict-for-dict). `preserve_order` keeps Python dict
//! insertion order across the wire — the save doc, node key order and the
//! nodes-map order are all observable through `json.dumps(indent=2)` on the
//! facade side.
//!
//! FFI (doc-envelope law): every call takes one JSON doc and returns
//! {"ok": true, "value": ...} | {"ok": false, "err": ..., "type": ...}.
//! * mg_load({"text": str|null}) -> graph wire
//! * mg_save({"graph"}) -> save doc
//! * mg_add({graph, id, ts, role, content, kind, parent, edge}) -> {graph, node}
//! * mg_tool_call({graph, id, ts, content, parent}) -> {graph, node}
//! * mg_tool_result({graph, id, ts, call_id, result}) -> {graph, node}
//! * mg_revise({graph, node_id, new_content, id, ts}) -> {graph, node}
//! * mg_branch({graph, node_id, id, ts}) -> {graph, node}
//! * mg_lineage({graph, node_id}) -> [nodes]
//! * mg_history({graph, node_id}) -> [nodes]
//! * mg_migrate_flat({text, transcript, ids, tss}) -> graph wire
//! * mg_cue({"node"}) -> cue string
//!
//! Graph wire: {"nodes": {id: node} (insertion-ordered), "order": [ids]}.
//! Errors ride typed: "ValueError" | "KeyError" | "AttributeError" — the
//! loader raises the matching Python exception (the oracle raises those).
//!
//! Determinism seams: the uuid4 hex id and the `_now()` timestamp are
//! facade ARGUMENTS (id/ts), never generated here — the frozen-truth
//! battery replays them verbatim. The save FILE IO (mkdir + json.dumps
//! indent=2) and tool_call's json.dumps({"call", "args"}) content
//! rendering stay in the facade: serialization isomorphism is Python's
//! job, structure is the kernel's.
//!
//! NOTE: no mg_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::collections::HashSet;
use std::ffi::{CStr, CString};
use std::os::raw::c_char;
pub const SCHEMA: i64 = 1;
pub const KINDS: [&str; 6] = [
"message", "reply", "tool_call", "tool_result", "revision", "branch_head",
];
pub const EDGE_TYPES: [&str; 7] = [
"parent", "reply_to", "calls", "results_in", "revises", "supersedes",
"branches_from",
];
/// The oracle's exception taxonomy, typed across the envelope.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrKind {
ValueError,
KeyError,
AttributeError,
TypeError,
}
impl ErrKind {
pub fn py_name(self) -> &'static str {
match self {
ErrKind::ValueError => "ValueError",
ErrKind::KeyError => "KeyError",
ErrKind::AttributeError => "AttributeError",
ErrKind::TypeError => "TypeError",
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct MgErr {
pub kind: ErrKind,
pub message: String,
}
impl MgErr {
fn value(msg: impl Into<String>) -> Self {
MgErr { kind: ErrKind::ValueError, message: msg.into() }
}
fn key(msg: impl Into<String>) -> Self {
MgErr { kind: ErrKind::KeyError, message: msg.into() }
}
}
/// The connected transcript. Append-only by design; revisions ADD nodes
/// (the old node is marked superseded, never deleted — provenance
/// survives). `nodes` is insertion-ordered (preserve_order IndexMap) and
/// `order` is the arrival list, mirroring the oracle's dict + list pair
/// (the two CAN diverge: duplicate ids in a loaded doc keep both order
/// entries while the map keeps the last node).
#[derive(Debug, Clone, Default)]
pub struct MessageGraph {
pub nodes: Map<String, Value>,
pub order: Vec<String>,
}
impl MessageGraph {
// -- persistence ----------------------------------------------------
/// `_load`: parse a saved file's text (schema v1 gate; corrupt or
/// foreign input is silently ignored — except a non-object top level,
/// which hits the oracle's `doc.get` and raises AttributeError).
pub fn load(text: Option<&str>) -> Result<MessageGraph, MgErr> {
let mut g = MessageGraph::default();
let Some(text) = text else {
return Ok(g);
};
let doc: Value = match serde_json::from_str(text) {
Ok(v) => v,
Err(_) => return Ok(g), // OSError/JSONDecodeError -> empty graph
};
let Some(obj) = doc.as_object() else {
return Err(MgErr {
kind: ErrKind::AttributeError,
message: format!(
"'{}' object has no attribute 'get'",
py_type(&doc)
),
});
};
if !schema_is_one(obj.get("schema").unwrap_or(&Value::Null)) {
return Ok(g);
}
match obj.get("nodes") {
None | Some(Value::Null) => Ok(g), // .get default []
Some(Value::Array(ns)) => {
for n in ns {
match n.get("id") {
None => {
return Err(MgErr::key("'id'"));
}
Some(Value::String(id)) => {
g.nodes.insert(id.clone(), n.clone());
g.order.push(id.clone());
}
Some(other) => {
// non-string id: out of the id contract; the map
// key takes the JSON form (the oracle's int-key
// dict does not survive the JSON wire anyway)
let key = other.to_string();
g.nodes.insert(key.clone(), n.clone());
g.order.push(key);
}
}
}
Ok(g)
}
// a non-list "nodes" is iterated by the oracle -> its elements
// are the dict's string keys and n["id"] raises TypeError
Some(_) => Err(MgErr {
kind: ErrKind::TypeError,
// Python 3.14 spells the string-index TypeError with the
// offending type appended
message: "string indices must be integers, not 'str'".into(),
}),
}
}
/// `save`: the doc the facade serializes with json.dumps(indent=2) —
/// nodes in arrival order (KeyError-free: order ids always exist here,
/// exactly like the oracle's list comprehension).
pub fn save(&self) -> Value {
let nodes: Vec<Value> = self
.order
.iter()
.map(|id| self.nodes.get(id).cloned().unwrap_or(Value::Null))
.collect();
json!({"schema": SCHEMA, "nodes": nodes})
}
// -- append ---------------------------------------------------------
/// Append a node. parent defaults to the current head (arrival chain).
/// NOTE the oracle's falsy law: the edge entry is written only when the
/// PARENT ARGUMENT is truthy — an explicit None parent still picks up
/// the head, but carries no edge; a falsy parent "" stores "" and also
/// carries no edge.
#[allow(clippy::too_many_arguments)] // faithful to the oracle's signature + id/ts seams
pub fn add(
&mut self,
node_id: &str,
ts: &str,
role: Value,
content: Value,
kind: &str,
parent: Option<&str>,
edge: &str,
) -> Result<Value, MgErr> {
if !KINDS.contains(&kind) {
return Err(MgErr::value(format!("unknown kind '{kind}'")));
}
if !EDGE_TYPES.contains(&edge) {
return Err(MgErr::value(format!("unknown edge '{edge}'")));
}
let parent_value = match parent {
Some(p) => json!(p),
None => match self.order.last() {
Some(head) => json!(head),
None => Value::Null,
},
};
let edges = match parent {
Some(p) if !p.is_empty() => json!([{"type": edge, "to": p}]),
_ => Value::Array(vec![]),
};
let mut node = Map::new();
node.insert("id".into(), json!(node_id));
node.insert("kind".into(), json!(kind));
node.insert("role".into(), role);
node.insert("content".into(), content);
node.insert("ts".into(), json!(ts));
node.insert("parent".into(), parent_value);
node.insert("edges".into(), edges);
node.insert("superseded_by".into(), Value::Null);
let node = Value::Object(node);
self.nodes.insert(node_id.to_string(), node.clone());
self.order.push(node_id.to_string());
Ok(node)
}
pub fn tool_call(
&mut self,
node_id: &str,
ts: &str,
content: &str,
parent: Option<&str>,
) -> Result<Value, MgErr> {
self.add(
node_id,
ts,
json!("tool"),
json!(content),
"tool_call",
parent,
"calls",
)
}
pub fn tool_result(
&mut self,
node_id: &str,
ts: &str,
call_id: &str,
result: Value,
) -> Result<Value, MgErr> {
if !self.nodes.contains_key(call_id) {
return Err(MgErr::key(format!("no such call '{call_id}'")));
}
self.add(
node_id,
ts,
json!("tool"),
result,
"tool_result",
Some(call_id),
"results_in",
)
}
// -- revision / branch ----------------------------------------------
/// Regenerate an answer without losing provenance: the old node is
/// marked superseded; the revision carries the supersedes edge. The
/// oracle quirk stands: a ROOT revision (old parent None) substitutes
/// the current head — which at add-time is the old node itself.
pub fn revise(
&mut self,
node_id: &str,
new_content: Value,
rev_id: &str,
ts: &str,
) -> Result<Value, MgErr> {
let (role, parent_owned) = match self.nodes.get(node_id) {
None => return Err(MgErr::key(format!("no such message '{node_id}'"))),
Some(old) => match old.get("role") {
Some(r) => (
r.clone(),
old.get("parent").and_then(Value::as_str).map(str::to_string),
),
None => return Err(MgErr::key("'role'")),
},
};
let mut rev = self.add(
rev_id,
ts,
role,
new_content,
"revision",
parent_owned.as_deref(),
"revises",
)?;
if let Some(edges) = rev.get_mut("edges").and_then(Value::as_array_mut) {
edges.push(json!({"type": "supersedes", "to": node_id}));
}
self.nodes.insert(rev_id.to_string(), rev.clone());
if let Some(old) = self.nodes.get_mut(node_id) {
old["superseded_by"] = json!(rev_id);
}
Ok(rev)
}
/// Start a new conversation branch off any message.
pub fn branch(
&mut self,
node_id: &str,
head_id: &str,
ts: &str,
) -> Result<Value, MgErr> {
if !self.nodes.contains_key(node_id) {
return Err(MgErr::key(format!("no such message '{node_id}'")));
}
self.add(
head_id,
ts,
json!("system"),
json!(format!("branched from {node_id}")),
"branch_head",
Some(node_id),
"branches_from",
)
}
// -- lineage context -------------------------------------------------
/// The context for a selected message: its ancestors (following parent
/// links) + itself, oldest first, skipping superseded nodes in favor of
/// their revisions. Defaults to the head. The seen-set guards cycles.
pub fn lineage(&self, node_id: Option<&str>) -> Vec<Value> {
let mut cur = match node_id {
Some(id) => Some(id.to_string()),
None => self.order.last().cloned(),
};
let Some(mut cur) = cur.take() else {
return vec![];
};
let mut chain: Vec<Value> = Vec::new();
let mut seen: HashSet<String> = HashSet::new();
while !cur.is_empty() && !seen.contains(&cur) {
seen.insert(cur.clone());
let Some(node) = self.nodes.get(&cur) else {
break;
};
// superseded nodes yield to their revision
if let Some(sup) = node.get("superseded_by").and_then(Value::as_str) {
if !sup.is_empty() && self.nodes.contains_key(sup) {
cur = sup.to_string();
continue;
}
}
chain.push(node.clone());
match node.get("parent").and_then(Value::as_str) {
Some(p) if !p.is_empty() => cur = p.to_string(),
_ => break,
}
}
chain.reverse();
chain
}
/// Every revision of a message, oldest first (the provenance trail).
/// Walks supersedes edges back to the root, then forward through
/// superseded_by — so the FULL trail comes back from any point on it.
pub fn history(&self, node_id: &str) -> Vec<Value> {
let mut cur = node_id.to_string();
while let Some(node) = self.nodes.get(&cur) {
let prev: Vec<&str> = node
.get("edges")
.and_then(Value::as_array)
.map(|edges| {
edges
.iter()
.filter(|e| {
e.get("type").and_then(Value::as_str) == Some("supersedes")
})
.filter_map(|e| e.get("to").and_then(Value::as_str))
.collect()
})
.unwrap_or_default();
match prev.first() {
Some(&p) => cur = p.to_string(),
None => break,
}
}
let mut trail = Vec::new();
let mut cur = if cur.is_empty() { None } else { Some(cur) };
while let Some(c) = cur {
let Some(node) = self.nodes.get(&c) else {
break;
};
trail.push(node.clone());
match node.get("superseded_by").and_then(Value::as_str) {
Some(s) if !s.is_empty() => cur = Some(s.to_string()),
_ => break,
}
}
trail
}
// -- migration -------------------------------------------------------
/// Schema migration: the flat {role, content} transcript becomes a
/// chained graph (arrival order, parent links). Compatibility
/// preserved — no conversation is lost in the upgrade.
pub fn migrate_flat(
text: Option<&str>,
transcript: &[Value],
ids: &[String],
tss: &[String],
) -> Result<MessageGraph, MgErr> {
let mut g = Self::load(text)?;
for (i, m) in transcript.iter().enumerate() {
let role = match m.get("role") {
Some(r) => r.clone(),
None => json!("system"),
};
let content = match m.get("content") {
Some(c) => c.clone(),
None => json!(""),
};
let id = ids
.get(i)
.ok_or_else(|| MgErr::value("ids/transcript length mismatch"))?;
let ts = tss
.get(i)
.ok_or_else(|| MgErr::value("tss/transcript length mismatch"))?;
g.add(id, ts, role, content, "message", None, "parent")?;
}
Ok(g)
}
// -- render cues ------------------------------------------------------
/// One quiet relationship line per node (no visual clutter).
pub fn cue(node: &Value) -> String {
if let Some(edges) = node.get("edges").and_then(Value::as_array) {
for e in edges {
match e.get("type").and_then(Value::as_str) {
Some("supersedes") => {
return format!(
"replaces {}",
take8(e.get("to").and_then(Value::as_str).unwrap_or(""))
);
}
Some("branches_from") => {
return format!(
"branched from {}",
take8(e.get("to").and_then(Value::as_str).unwrap_or(""))
);
}
_ => {}
}
}
}
if let Some(s) = node.get("superseded_by").and_then(Value::as_str) {
if !s.is_empty() {
return format!("superseded by {}", take8(s));
}
}
String::new()
}
// -- wire -------------------------------------------------------------
pub fn to_wire(&self) -> Value {
json!({"nodes": self.nodes, "order": self.order})
}
}
fn take8(s: &str) -> String {
s.chars().take(8).collect()
}
/// Python json.loads type names (for the AttributeError messages).
fn py_type(v: &Value) -> &'static str {
match v {
Value::Null => "NoneType",
Value::Bool(_) => "bool",
Value::Number(n) => {
if n.is_f64() {
"float"
} else {
"int"
}
}
Value::String(_) => "str",
Value::Array(_) => "list",
Value::Object(_) => "dict",
}
}
/// The oracle's `doc.get("schema") != SCHEMA` gate: Python equality —
/// 1.0 == 1 and True == 1 both pass.
fn schema_is_one(v: &Value) -> bool {
v.as_i64() == Some(SCHEMA)
|| v.as_f64() == Some(SCHEMA as f64)
|| v.as_bool() == Some(true)
}
/// Graph state from the wire: {"nodes": {id: node}, "order": [ids]}.
fn graph_from_wire(arg: &Value) -> Result<MessageGraph, MgErr> {
let g = arg.get("graph").ok_or_else(|| arg_err("arg.graph required"))?;
let nodes = g
.get("nodes")
.and_then(Value::as_object)
.ok_or_else(|| arg_err("graph.nodes must be an object"))?
.clone();
let order = g
.get("order")
.and_then(Value::as_array)
.ok_or_else(|| arg_err("graph.order must be an array"))?
.iter()
.map(|v| {
v.as_str()
.map(str::to_string)
.ok_or_else(|| arg_err("graph.order entries must be strings"))
})
.collect::<Result<Vec<String>, MgErr>>()?;
Ok(MessageGraph { nodes, order })
}
fn opt_str<'a>(arg: &'a Value, key: &str) -> Option<&'a str> {
arg.get(key).and_then(Value::as_str)
}
fn require_str<'a>(arg: &'a Value, key: &str) -> Result<&'a str, MgErr> {
opt_str(arg, key).ok_or_else(|| arg_err(format!("arg.{key} required")))
}
// ── FFI (doc-envelope law) ──────────────────────────────────────────────
fn read_arg(ptr: *const c_char) -> Result<Value, MgErr> {
if ptr.is_null() {
return Err(arg_err("null arg"));
}
let s = unsafe { CStr::from_ptr(ptr) }
.to_str()
.map_err(|e| arg_err(e.to_string()))?;
serde_json::from_str(s).map_err(|e| arg_err(e.to_string()))
}
fn envelope(r: Result<Value, MgErr>) -> *mut c_char {
let v = match r {
Ok(value) => json!({"ok": true, "value": value}),
Err(e) => json!({"ok": false, "err": e.message, "type": e.kind.py_name()}),
};
match CString::new(serde_json::to_string(&v).unwrap_or_default()) {
Ok(s) => s.into_raw(),
Err(_) => CString::new("{\"ok\":false,\"err\":\"envelope nul\"}").unwrap().into_raw(),
}
}
fn arg_err(e: impl Into<String>) -> MgErr {
MgErr { kind: ErrKind::ValueError, message: e.into() }
}
#[no_mangle]
pub extern "C" fn mg_load(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
MessageGraph::load(opt_str(&a, "text")).map(|g| g.to_wire())
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_save(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg)
.and_then(|a| graph_from_wire(&a))
.map(|g| g.save());
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_add(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let mut g = graph_from_wire(&a)?;
let node = g.add(
require_str(&a, "id")?,
require_str(&a, "ts")?,
a.get("role").cloned().unwrap_or(Value::Null),
a.get("content").cloned().unwrap_or(Value::Null),
require_str(&a, "kind")?,
opt_str(&a, "parent"),
require_str(&a, "edge")?,
)?;
Ok(json!({"graph": g.to_wire(), "node": node}))
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_tool_call(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let mut g = graph_from_wire(&a)?;
let node = g.tool_call(
require_str(&a, "id")?,
require_str(&a, "ts")?,
require_str(&a, "content")?,
opt_str(&a, "parent"),
)?;
Ok(json!({"graph": g.to_wire(), "node": node}))
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_tool_result(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let mut g = graph_from_wire(&a)?;
let node = g.tool_result(
require_str(&a, "id")?,
require_str(&a, "ts")?,
require_str(&a, "call_id")?,
a.get("result").cloned().unwrap_or(Value::Null),
)?;
Ok(json!({"graph": g.to_wire(), "node": node}))
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_revise(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let mut g = graph_from_wire(&a)?;
let node = g.revise(
require_str(&a, "node_id")?,
a.get("new_content").cloned().unwrap_or(Value::Null),
require_str(&a, "id")?,
require_str(&a, "ts")?,
)?;
Ok(json!({"graph": g.to_wire(), "node": node}))
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_branch(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let mut g = graph_from_wire(&a)?;
let node = g.branch(
require_str(&a, "node_id")?,
require_str(&a, "id")?,
require_str(&a, "ts")?,
)?;
Ok(json!({"graph": g.to_wire(), "node": node}))
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_lineage(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let g = graph_from_wire(&a)?;
Ok(Value::Array(g.lineage(opt_str(&a, "node_id"))))
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_history(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let g = graph_from_wire(&a)?;
Ok(Value::Array(g.history(require_str(&a, "node_id")?)))
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_migrate_flat(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).and_then(|a| {
let transcript = a
.get("transcript")
.and_then(Value::as_array)
.ok_or_else(|| arg_err("arg.transcript must be an array"))?
.clone();
let ids: Vec<String> = a
.get("ids")
.and_then(Value::as_array)
.ok_or_else(|| arg_err("arg.ids must be an array"))?
.iter()
.map(|v| v.as_str().unwrap_or_default().to_string())
.collect();
let tss: Vec<String> = a
.get("tss")
.and_then(Value::as_array)
.ok_or_else(|| arg_err("arg.tss must be an array"))?
.iter()
.map(|v| v.as_str().unwrap_or_default().to_string())
.collect();
MessageGraph::migrate_flat(opt_str(&a, "text"), &transcript, &ids, &tss)
.map(|g| g.to_wire())
});
envelope(r)
}
#[no_mangle]
pub extern "C" fn mg_cue(arg: *const c_char) -> *mut c_char {
let r = read_arg(arg).map(|a| {
Value::String(MessageGraph::cue(
a.get("node").unwrap_or(&Value::Null),
))
});
envelope(r)
}
#[cfg(test)]
mod branch_battery;