feat(rebase): wave 0 — whole-project inventory, per-function issue factory, Rust interop proof unit

Wave 0 of the approved language-rebasing tranche (R1):

- scripts/rebase_inventory.py: AST-based whole-project inventory →
  docs/rebase-inventory.json (1,852 true units = 1,030 top-level fns
  + 822 methods; module-level walk skips defs; class-method leak fixed)
- docs/rebase-wave-coverage.json: waves 2–7 coverage {239,251,129,544,415,274}
- scripts/rebase_issue_factory.py: one-issue-per-unit Gitea factory,
  dry-run default, deterministic titles + dedup, --limit defaults to
  unlimited for create; 1,852 issues created across waves 2–7
- interop/rebase-proof + shared/native_units.py + tests: std-only Rust
  proof unit for the §5 native-interop contract (argv→JSON, --version,
  exit 2 on failure); thin typed Python adapter; 8/8 tests pass with
  corrected ok-semantics (ok = exit 0 AND not timed out)
- config/umbrellas.json: native_units registered in capability umbrella
  (manifest law — the map travels with the change)
- docs/rebase-wave0-report.md: Wave 0 close-out report

.gitignore

@@ -11,3 +11,6 @@ build/
 .env
 .venv/
 state/
+
+# Rust build output
+interop/*/target/

config/umbrellas.json

@@ -118,7 +118,8 @@
         "local_tool_runtime",
         "forge_client",
         "message_queue",
-        "connection_pool"
+        "connection_pool",
+        "native_units"
       ],
       "canaries": [
         "canary_p1_01",

docs/rebase-wave-coverage.json

@@ -0,0 +1,12 @@
+{
+ "units_per_wave": {
+  "2": 239,
+  "3": 251,
+  "4": 129,
+  "5": 544,
+  "6": 415,
+  "7": 274
+ },
+ "unassigned_modules": {},
+ "unassigned_module_count": 0
+}

docs/rebase-wave0-report.md

@@ -0,0 +1,50 @@
+# Wave 0 close-out report — language-rebasing tranche R1
+
+Branch: `tranche-r1-wave-0` · Date: 2026-09-29
+
+## 1. Inventory (docs/rebase-inventory.json)
+
+- 177 Python modules, ~52,750 lines, 227 classes
+- TRUE unit count: **1,852 = 1,030 top-level functions + 822 class methods**
+- Zero duplicate call edges, zero method leaks (generator double-walk defect
+  root-caused and fixed in scripts/rebase_inventory.py)
+- Cross-verified against an independent fresh AST walk: zero missing units;
+  six inventory-"extras" are real nested/if-try-block classes the naive walk
+  misses (inventory is MORE complete, not less)
+
+## 2. Issue factory (scripts/rebase_issue_factory.py)
+
+- One issue per unit following R1 §7; wave-batched (2–7); dedup-safe via
+  deterministic title `[rebase w{wave}] {relpath}::{unit}` + existing-titles
+  scan per run (retries never duplicate — preserve-first)
+- `plan` → {2: 239, 3: 251, 4: 129, 5: 544, 6: 415, 7: 274} = 1,852,
+  zero unassigned
+- `sample --wave 5` → full §7 template holds; `_compute_hash` exactly once as
+  labelled callee with line number
+- Contract fix during Wave 0: `--limit` default changed 5 → 0 (unlimited) for
+  `create`; `sample` keeps its 5. A full-wave create with the old default
+  would have silently truncated each wave to five issues.
+
+## 3. Interop proof unit (R1 §5)
+
+- Rust unit at interop/rebase-proof/ (std-only, zero crates): argv in → JSON
+  on stdout, non-zero exit on failure, mandatory `--version`
+- Shell-proven: version flag, JSON out, exit 2 on failure, stderr diagnostic
+- Python thin adapter shared/native_units.py: typed NativeResult, timeout +
+  stderr capture, unit failures as data (never exceptions)
+- Adapter contract fix during Wave 0: `ok` no longer requires a JSON
+  `ok:true` field — plain-text output (`--version`) succeeds on exit code
+- tests/test_native_interop_proof.py: **8 passed** — version contract, sum
+  happy path, hash determinism, failure paths, timeout, stderr capture
+
+## 4. Per-wave issue creation
+
+<!-- FILLED BY THE COMPLETION LOOP — do not hand-edit -->
+
+- Wave 4 (129 units): created=127 skipped(dedup)=2 (includes smoke #417–418)
+- Wave 2 (239 units): created=239 skipped(dedup)=0
+- Wave 3 (251 units): created=251 skipped(dedup)=0
+- Wave 5 (544 units): created=544 skipped(dedup)=0
+- Wave 6 (415 units): created=415 skipped(dedup)=0
+- Wave 7 (274 units): created=274 skipped(dedup)=0
+- TOTAL: 1,850 created + 2 dedup = 1,852 / 1,852 ✅ full coverage

interop/rebase-proof/Cargo.lock

@@ -0,0 +1,7 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 4
+
+[[package]]
+name = "rebase-proof"
+version = "0.1.0"

interop/rebase-proof/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "rebase-proof"
+version = "0.1.0"
+edition = "2021"
+
+# Wave 0 interop proof unit (R1 §5). Std-only by design: no crates, no network,
+# deterministic build. The pattern under proof is the CLI boundary itself, not the
+# computation inside it.
+
+[[bin]]
+name = "rebase-proof"
+path = "src/main.rs"
+
+[profile.release]
+strip = true

interop/rebase-proof/src/main.rs

@@ -0,0 +1,117 @@
+//! Wave 0 interop proof unit (R1 §5 of the language-rebasing tranche).
+//!
+//! Contract under proof:
+//! - standalone binary; argv in, JSON on stdout
+//! - non-zero exit on failure (JSON error object still printed to stdout)
+//! - `--version` flag mandatory
+//! - Python wraps it in a thin adapter (`shared/native_units.py`); tests call
+//!   it the same way.
+//!
+//! Ops: sum / echo / hash (fnv1a64) / sleep / fail. All std-only.
+
+use std::env;
+use std::process::exit;
+use std::thread;
+use std::time::Duration;
+
+const VERSION: &str = env!("CARGO_PKG_VERSION");
+
+/// FNV-1a 64-bit — dependency-free deterministic hash for the proof unit.
+fn fnv1a64(data: &[u8]) -> u64 {
+    let mut hash: u64 = 0xcbf29ce484222325;
+    for byte in data {
+        hash ^= u64::from(*byte);
+        hash = hash.wrapping_mul(0x100000001b3);
+    }
+    hash
+}
+
+fn print_json(line: &str) {
+    println!("{line}");
+}
+
+fn emit_ok(body: &str) {
+    print_json(&format!("{{\"ok\":true,{body}}}"));
+    exit(0);
+}
+
+fn emit_err(message: &str) -> ! {
+    // Escape the minimal set so the JSON stays parseable for arbitrary op names.
+    let escaped: String = message
+        .chars()
+        .map(|c| match c {
+            '"' => "\\\"".to_string(),
+            '\\' => "\\\\".to_string(),
+            '\n' => "\\n".to_string(),
+            '\r' => "\\r".to_string(),
+            '\t' => "\\t".to_string(),
+            other => other.to_string(),
+        })
+        .collect();
+    print_json(&format!("{{\"ok\":false,\"error\":\"{escaped}\"}}"));
+    exit(2);
+}
+
+fn main() {
+    let args: Vec<String> = env::args().skip(1).collect();
+
+    if args.as_slice() == ["--version"] || args.as_slice() == ["-V"] {
+        println!("rebase-proof {VERSION}");
+        exit(0);
+    }
+    if args.is_empty() {
+        eprintln!("usage: rebase-proof <op> [args...] | --version");
+        emit_err("no op given");
+    }
+
+    let op = args[0].as_str();
+    let rest = &args[1..];
+
+    match op {
+        "sum" => {
+            let mut total: i128 = 0;
+            for a in rest {
+                match a.parse::<i128>() {
+                    Ok(v) => total += v,
+                    Err(_) => emit_err(&format!("sum: not an integer: {a}")),
+                }
+            }
+            emit_ok(&format!("\"op\":\"sum\",\"result\":{total}"));
+        }
+        "echo" => {
+            let joined = rest.join(" ");
+            let escaped: String = joined
+                .chars()
+                .map(|c| match c {
+                    '"' => "\\\"".to_string(),
+                    '\\' => "\\\\".to_string(),
+                    '\n' => "\\n".to_string(),
+                    '\r' => "\\r".to_string(),
+                    '\t' => "\\t".to_string(),
+                    other => other.to_string(),
+                })
+                .collect();
+            emit_ok(&format!("\"op\":\"echo\",\"text\":\"{escaped}\""));
+        }
+        "hash" => {
+            let joined = rest.join(" ");
+            let h = fnv1a64(joined.as_bytes());
+            emit_ok(&format!("\"op\":\"hash\",\"fnv1a64\":\"{h:016x}\""));
+        }
+        "sleep" => {
+            let ms: u64 = match rest.first().map(|s| s.parse::<u64>()) {
+                Some(Ok(v)) => v,
+                _ => emit_err("sleep: expected milliseconds"),
+            };
+            thread::sleep(Duration::from_millis(ms));
+            emit_ok(&format!("\"op\":\"sleep\",\"slept_ms\":{ms}"));
+        }
+        "fail" => {
+            // Deliberate failure: stderr diagnostic + JSON error + exit 2.
+            // Proves stderr capture and non-zero exit in the adapter.
+            eprintln!("rebase-proof: deliberate failure for adapter proof");
+            emit_err("deliberate failure");
+        }
+        other => emit_err(&format!("unknown op: {other}")),
+    }
+}

scripts/rebase_inventory.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+"""Tranche R1 Wave 0 — whole-project inventory generator.
+
+Parses every .py file under src/rune/ with the stdlib AST and emits
+docs/rebase-inventory.json: files, classes, functions (with signatures and
+line ranges), imports, and best-effort caller/callee edges (intra-module call
+graph + cross-module edges resolved through import tables).
+
+Deterministic and idempotent: same tree in, same JSON out (sorted keys,
+sorted files). The issue-factory reads this file; nothing else may edit it.
+"""
+from __future__ import annotations
+
+import ast
+import json
+import sys
+from collections import defaultdict
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+SRC = REPO / "src" / "rune"
+OUT = REPO / "docs" / "rebase-inventory.json"
+
+
+def _arg_str(a: ast.arg) -> str:
+    s = a.arg
+    if a.annotation is not None:
+        try:
+            s += f": {ast.unparse(a.annotation)}"
+        except Exception:
+            pass
+    return s
+
+
+def _signature(node: ast.FunctionDef | ast.AsyncFunctionDef) -> str:
+    parts = [_arg_str(a) for a in node.args.args]
+    if node.args.vararg:
+        parts.append(f"*{node.args.vararg.arg}")
+    elif node.args.kwonlyargs:
+        parts.append("*")
+    parts += [_arg_str(a) for a in node.args.kwonlyargs]
+    if node.args.kwarg:
+        parts.append(f"**{node.args.kwarg.arg}")
+    ret = ""
+    if node.returns is not None:
+        try:
+            ret = f" -> {ast.unparse(node.returns)}"
+        except Exception:
+            pass
+    return f"({', '.join(parts)}){ret}"
+
+
+class FileVisitor(ast.NodeVisitor):
+    def __init__(self) -> None:
+        self.functions: list[dict] = []
+        self.classes: list[dict] = []
+        self.imports: list[dict] = []
+        self.calls: list[tuple[int, str]] = []  # (lineno, dotted call name)
+
+    def _fn(self, node, parents: list[str]) -> dict:
+        return {
+            "name": ".".join([*parents, node.name]),
+            "signature": _signature(node),
+            "line_start": node.lineno,
+            "line_end": node.end_lineno or node.lineno,
+            "decorators": [ast.unparse(d) for d in node.decorator_list
+                           if isinstance(d, (ast.Name, ast.Attribute, ast.Call))],
+            "async": isinstance(node, ast.AsyncFunctionDef),
+        }
+
+    def visit_ClassDef(self, node: ast.ClassDef) -> None:
+        methods = []
+        for child in node.body:
+            if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                methods.append(self._fn(child, [node.name]))
+                self.calls.extend(
+                    (n.lineno, ast.unparse(n.func))
+                    for n in ast.walk(child)
+                    if isinstance(n, ast.Call) and not isinstance(n.func, ast.Name if False else type(None))
+                    and isinstance(n.func, (ast.Name, ast.Attribute))
+                )
+        self.classes.append({
+            "name": node.name,
+            "bases": [ast.unparse(b) for b in node.bases],
+            "line_start": node.lineno,
+            "line_end": node.end_lineno or node.lineno,
+            "methods": methods,
+        })
+        # visit only non-function children: methods are already captured above,
+        # and a generic_visit would re-fire visit_FunctionDef on them (duplicate
+        # units + duplicate call edges). Nested classes are still visited.
+        for child in node.body:
+            if not isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                self.visit(child)
+
+    def visit_FunctionDef(self, node) -> None:
+        # top-level only reached here (methods captured in ClassDef)
+        self.functions.append(self._fn(node, []))
+        self.calls.extend(
+            (n.lineno, ast.unparse(n.func))
+            for n in ast.walk(node)
+            if isinstance(n, ast.Call) and isinstance(n.func, (ast.Name, ast.Attribute))
+        )
+        # do not descend into nested defs separately; they are folded into parents
+        # (their calls are already captured by the ast.walk above). No generic_visit:
+        # it would re-collect nested defs as top-level functions and duplicate edges.
+
+    visit_AsyncFunctionDef = visit_FunctionDef
+
+    def visit_Import(self, node: ast.Import) -> None:
+        for a in node.names:
+            self.imports.append({"module": a.name, "as": a.asname, "line": node.lineno})
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        mod = node.module or ""
+        for a in node.names:
+            self.imports.append({"module": f"{mod}.{a.name}" if a.name != "*" else mod,
+                                 "as": a.asname or a.name, "line": node.lineno,
+                                 "from": True})
+
+
+def collect_file(path: Path) -> dict:
+    src_text = path.read_text(encoding="utf-8", errors="replace")
+    tree = ast.parse(src_text)
+    v = FileVisitor()
+    v.visit(tree)
+
+    # module-level calls (outside defs/classes) — walk tree body directly.
+    # SKIP def/class statements: their bodies are walked by the visit_ methods;
+    # walking them here duplicated every call edge inside every top-level function.
+    for stmt in tree.body:
+        if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+            continue
+        for n in ast.walk(stmt):
+            if isinstance(n, ast.Call) and isinstance(n.func, (ast.Name, ast.Attribute)):
+                v.calls.append((n.lineno, ast.unparse(n.func)))
+
+    defined = {f["name"].split(".")[-1] for f in v.functions}
+    defined |= {m["name"].split(".")[-1] for c in v.classes for m in c["methods"]}
+    defined |= {c["name"] for c in v.classes}
+
+    callees = []
+    for lineno, call in v.calls:
+        root = call.split(".")[0].split("(")[0].strip()
+        callees.append({"line": lineno, "call": call,
+                        "internal": root in defined or root == "self"})
+
+    return {
+        "path": str(path.relative_to(REPO)),
+        "lines": len(src_text.splitlines()),
+        "imports": sorted({i["module"] for i in v.imports}),
+        "import_lines": v.imports,
+        "functions": sorted(v.functions, key=lambda f: (f["line_start"], f["name"])),
+        "classes": sorted(v.classes, key=lambda c: c["line_start"]),
+        "calls": callees,
+    }
+
+
+def cross_module_edges(files: dict[str, dict]) -> dict[str, set[str]]:
+    """Resolve import tables: module-stem -> set of imported module stems
+    (rune-internal only), so the factory can attribute cross-module edges."""
+    graph = defaultdict(set)
+    for rel, info in files.items():
+        stem = Path(rel).stem
+        for imp in info["imports"]:
+            top = imp.split(".")[0]
+            second = ".".join(imp.split(".")[:2])
+            if top == "rune":
+                graph[stem].add(imp.replace("rune.", "", 1).split(".")[0])
+            elif not imp.startswith("_") and top not in sys.stdlib_module_names:
+                graph[stem].add(f"ext:{top}")
+    return graph
+
+
+def main() -> None:
+    py_files = sorted(SRC.rglob("*.py"))
+    files: dict[str, dict] = {}
+    for p in py_files:
+        try:
+            info = collect_file(p)
+        except SyntaxError as e:  # a file that cannot parse is a finding, not a skip
+            info = {"path": str(p.relative_to(REPO)), "lines": 0, "imports": [],
+                    "import_lines": [], "functions": [], "classes": [], "calls": [],
+                    "parse_error": f"{type(e).__name__}: {e}"}
+        files[info["path"]] = info
+
+    graph = cross_module_edges(files)
+    total_fns = sum(len(f["functions"]) + sum(len(c["methods"]) for c in f["classes"])
+                    for f in files.values())
+    doc = {
+        "_meta": {
+            "generator": "scripts/rebase_inventory.py",
+            "tranche": "R1 language rebase, Wave 0",
+            "note": "Machine-generated. Regenerate with the script; never hand-edit.",
+            "files": len(files),
+            "functions_total": total_fns,
+            "classes_total": sum(len(f["classes"]) for f in files.values()),
+            "lines_total": sum(f["lines"] for f in files.values()),
+        },
+        "import_graph": {k: sorted(v) for k, v in sorted(graph.items())},
+        "files": dict(sorted(files.items())),
+    }
+    OUT.write_text(json.dumps(doc, indent=1, sort_keys=False) + "\n", encoding="utf-8")
+    m = doc["_meta"]
+    print(f"inventory written: {OUT}")
+    print(f"files={m['files']} functions={m['functions_total']} "
+          f"classes={m['classes_total']} lines={m['lines_total']}")
+
+
+if __name__ == "__main__":
+    main()

scripts/rebase_issue_factory.py

@@ -0,0 +1,356 @@
+#!/usr/bin/env python3
+"""Tranche R1 Wave 0 — per-function issue factory.
+
+Reads docs/rebase-inventory.json and emits one issue per unit (top-level
+function or class method) following tranche R1 §7. Wave-batched, dedup-safe
+(deterministic title + existing-issue scan before create), rate-limited.
+
+Modes:
+  plan      — coverage report: units per wave, unassigned modules (no silent drops)
+  sample    — emit N rendered issues to a file for review (no Gitea call)
+  create    — create issues in Gitea for one wave
+
+Dedup contract: issue titles are deterministic —
+  [rebase w{wave}] {relpath}::{unit}
+The factory fetches existing issue titles once per run and skips any that
+already exist, so retries never duplicate (preserve-first Gitea addendum).
+
+The factory NEVER writes behaviour prose. Current-behaviour paragraphs and
+the three evidence answers (IMPROVE / FIX / MISSING CONNECTORS) are filled at
+disposition time by the working agent — the issue carries the mandatory
+fields and line ranges, not fabricated content.
+"""
+
+import argparse
+import json
+import os
+import sys
+import time
+import urllib.request
+import urllib.error
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+INVENTORY = REPO_ROOT / "docs" / "rebase-inventory.json"
+COVERAGE_OUT = REPO_ROOT / "docs" / "rebase-wave-coverage.json"
+
+GITEA_URL = os.environ.get("GITEA_URL", "https://gitea.thingscouldgetdicey.online")
+REPO = "conductor/rune"
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "shared"))
+from gitea_auth import load_gitea_token  # noqa: E402
+
+TRANCHE_REF = "tranche R1 language rebase (2026-09-29), ~/.config/opencode/tranches/2026-09-29-language-rebase-tranche-R1.md"
+
+# Wave membership from R1 §4 (initial, measured). Directory entries match by
+# path prefix; bare stems match the file stem. Modules matching no entry are
+# reported UNASSIGNED by `plan` — never silently dropped (R1 §9 risk 7).
+WAVE_PLAN = {
+    2: [  # leaf utilities & canaries
+        "pins", "time_awareness", "message_queue", "event_envelope",
+        "attention_capsule", "complexity_scorer", "token_optimizer", "cache",
+        "prompt_sync", "chat_bubble", "canary_*",
+    ],
+    3: [  # memory & persistence
+        "memory_store", "curator", "warm_router", "memory_consolidation",
+        "graph_memory", "sleep_time", "night_mode", "identity_anchors",
+        "query_memory", "memory_env", "warm_neuron_store", "subconscious_pass",
+        "task_persist",
+    ],
+    4: [  # provider/model layer
+        "provider_registry", "model_connect", "model_routing", "model_profiles",
+        "provider_budget", "standardized_providers", "model_registry",
+        "connection_pool", "forge_client",
+    ],
+    5: [  # orchestration, authority, plugins, skills, governance
+        "orchestration/", "microkernel", "contracts", "build_mode", "plugins/",
+        "skills/", "construction_authority", "construction_authority_cli",
+        "construction_decision", "construction_mode", "verification_construction",
+        "authority_recovery", "spine", "drift_core", "drift_ledger",
+        "drift_observability", "persona_gate", "safety_override", "witness",
+        "behavior", "growth_delta", "claim_audit", "verification_layer",
+        "drift_compression", "drift_cache", "self_improve", "persona_definition",
+        "outcome_ledger", "audit_ledger", "subsystem_health",
+    ],
+    6: [  # runtime core & tooling
+        "tool_executor", "tool_protocol", "local_tool_runtime",
+        "capability_broker", "context_broker", "work_manager", "cron_engine",
+        "auto_tasks", "goal", "failure_auto_filer", "hook_engine",
+        "elastic_cognition", "event_bus", "system_registry", "abstraction",
+        "tiered_executor", "progressive_escalation", "tier_manager",
+        "parallel_processor", "shard_manager", "response_adapter",
+        "context_pressure", "auto_context", "cognitive_budgeter",
+        "recall_router", "inference_trace", "task_gate", "salience_gate",
+        "dependency_tracker", "rune_profiler",
+    ],
+    7: [  # operator + TUI, last by design
+        "rune_operator", "rune_tui", "tui_validator", "voice_hybrid",
+        "voice_sync", "voicebox", "voice_injector", "message_widgets",
+        "rune_graphic", "rune_spinner", "__main__",
+    ],
+}
+
+WAVE_NAMES = {
+    2: "leaf utilities & canaries",
+    3: "memory & persistence",
+    4: "provider/model layer",
+    5: "orchestration, authority, plugins, skills, governance",
+    6: "runtime core & tooling",
+    7: "operator + TUI",
+}
+
+
+def match_module(relpath: str):
+    """Return the wave number for a module path, or None if unassigned."""
+    stem = Path(relpath).stem
+    for wave, patterns in WAVE_PLAN.items():
+        for pat in patterns:
+            if pat.endswith("/"):
+                if pat in relpath:
+                    return wave
+            elif pat.endswith("*"):
+                if stem.startswith(pat[:-1]):
+                    return wave
+            elif stem == pat:
+                return wave
+    return None
+
+
+def iter_units(inv):
+    """Yield (relpath, unit_name, entry) for every function and method."""
+    for relpath, mod in sorted(inv["files"].items()):
+        for fn in mod["functions"]:
+            yield relpath, fn["name"], fn
+        for cls in mod["classes"]:
+            for m in cls.get("methods", []):
+                yield relpath, f"{cls['name']}.{m['name']}", m
+
+
+def internal_edges(relpath: str, entry: dict):
+    """Collect internal (in-repo) call names and line numbers for a unit."""
+    lines = (entry["line_start"], entry["line_end"])
+    mod = inv_cache["files"][relpath]
+    calls = [c for c in mod.get("calls", [])
+             if lines[0] <= c["line"] <= lines[1] and c.get("internal")]
+    return calls
+
+
+inv_cache = {}
+
+
+def issue_title(wave: int, relpath: str, unit: str) -> str:
+    return f"[rebase w{wave}] {relpath}::{unit}"
+
+
+def render_issue(wave: int, relpath: str, unit: str, entry: dict) -> dict:
+    calls = internal_edges(relpath, entry)
+    call_lines = "\n".join(
+        f"  - `{c['call']}` (line {c['line']})" for c in calls[:20]
+    ) or "  - (no internal call edges in range)"
+    sig = entry.get("signature", "()")
+    body = f"""\
+
+## Identity (from the Wave 0 inventory — machine-generated)
+
+- **File**: `{relpath}`
+- **Unit**: `{unit}`
+- **Signature**: `def {unit}{sig}` (async: {entry.get('async', False)})
+- **Line range**: {entry['line_start']}–{entry['line_end']}
+- **Decorators**: {entry.get('decorators', []) or 'none'}
+- **Internal call edges** (callers/callees inside the repo, from inventory):
+{call_lines}
+
+## Current behaviour
+
+One honest paragraph from READING THE CODE at the line range above — what it
+actually does, not what the docstring promises. Write this at disposition
+time; the factory does not fabricate it.
+
+## The three questions — each answered WITH EVIDENCE
+
+1. **IMPROVE?** Can the unit be improved within its scope? Evidence or "no — because …".
+   (Bounded to unit scope; larger findings spawn NEW issues, never expand this one.)
+2. **FIX?** Is it broken, buggy, or failing its expected outcome? Evidence: a failing
+   test, a reproduced fault, or a code-level defect citation. A "no" needs evidence too.
+3. **MISSING CONNECTORS?** Are there expected outcomes this unit should connect to but
+   doesn't (unwired events, dead parameters, promised-but-absent calls)? Evidence required.
+
+A silent skip on any question is a factory bug — answer all three.
+
+## Language verdict (per tranche R1 §6 rubric — evidence beats instinct)
+
+Choose exactly one, with the evidence line:
+- **KEEP-PYTHON** — one sentence of justification.
+- **RECODE-<lang>** — rubric scores (all 7 dimensions) + benchmark-gate plan: the compiled
+  replacement must beat the Python original on the MEASURED hot path, or the recode is
+  rejected and recorded. No win, no recode.
+- **MERGE-into:<module>** — name the target and why.
+- **DEPRECATE** — name the caller that no longer needs it.
+
+Go is NOT pre-decided. Python is not pre-decided. The rubric decides.
+
+## Constraint context
+
+- henxels structural contract: this unit's module ≤ 1,200 lines (or on the debt table's
+  exception list — check before adding lines).
+- Spine: any module-surface change requires `store/spine/current.json` regen with the code
+  change (`test_manifest_maps_every_module` is the arbiter for new top-level modules).
+- Debt table: docs/structural-contract.md.
+
+## Pipeline plan
+
+1. Branch `issue-<N>` from current master (existing issue-N convention).
+2. Scaffolding activation for any construction (construction_mode/decision per unit change).
+3. Research rounds are MANDATORY for any RECODE verdict (research-stage-headless; ≥2
+   live-verified external sources per claim; per-source lessons; kill test per candidate).
+4. Implementation behind the interop contract (R1 §5): compiled units are standalone
+   binaries — argv in, JSON on stdout, non-zero exit on failure, `--version` mandatory;
+   Python wraps in a thin adapter; store stays the shared truth.
+5. Local gates: pytest + ruff + spine-store regen (if surface changed) + henxels.
+6. PR → pipeline-gate + ci + required checks → merge-robot → auto-version.
+
+## Verification
+
+- The specific test proving this unit's change: (write at disposition).
+- The keep-green set it must not break: full suite must stay green; named
+  keep-green tests for this unit's callers.
+
+## Provenance
+
+- Tranche: {TRANCHE_REF}
+- Wave: {wave} ({WAVE_NAMES.get(wave, 'unassigned')})
+- Inventory: docs/rebase-inventory.json (regenerate with scripts/rebase_inventory.py — never hand-edit)
+- Factory: scripts/rebase_issue_factory.py
+"""
+    return {
+        "title": issue_title(wave, relpath, unit),
+        "body": body,
+        "labels": ["rebase", f"wave-{wave}"],
+        "meta": {
+            "wave": wave,
+            "file": relpath,
+            "unit": unit,
+            "line_start": entry["line_start"],
+            "line_end": entry["line_end"],
+            "async": entry.get("async", False),
+        },
+    }
+
+
+def fetch_existing_titles() -> set:
+    """Fetch all issue titles in the repo (paginated) for dedup."""
+    token = load_gitea_token()
+    titles = set()
+    page = 1
+    while True:
+        url = (f"{GITEA_URL}/api/v1/repos/{REPO}/issues"
+               f"?state=all&type=issues&limit=50&page={page}")
+        req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            batch = json.loads(resp.read())
+        if not batch:
+            break
+        titles.update(i["title"] for i in batch)
+        if len(batch) < 50:
+            break
+        page += 1
+    return titles
+
+
+def create_issue(entry: dict) -> dict:
+    token = load_gitea_token()
+    url = f"{GITEA_URL}/api/v1/repos/{REPO}/issues"
+    payload = json.dumps({
+        "title": entry["title"],
+        "body": entry["body"],
+        "labels": [],  # label NAMES are set via the labels endpoint by auto-label; keep create minimal
+    }).encode()
+    req = urllib.request.Request(url, data=payload, method="POST",
+                                 headers={"Authorization": f"token {token}",
+                                          "Content-Type": "application/json"})
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read())
+
+
+def cmd_plan(inv):
+    counts = {}
+    unassigned_modules = {}
+    for relpath, unit, entry in iter_units(inv):
+        wave = match_module(relpath)
+        counts[wave] = counts.get(wave, 0) + 1
+        if wave is None:
+            unassigned_modules.setdefault(relpath, inv["files"][relpath]["lines"])
+    report = {
+        "units_per_wave": {str(k if k is not None else "UNASSIGNED"): v
+                           for k, v in sorted(counts.items(), key=lambda kv: str(kv[0]))},
+        "unassigned_modules": dict(sorted(unassigned_modules.items(), key=lambda kv: -kv[1])),
+        "unassigned_module_count": len(unassigned_modules),
+    }
+    print(json.dumps(report, indent=1))
+    COVERAGE_OUT.write_text(json.dumps(report, indent=1) + "\n")
+    print(f"\ncoverage written to {COVERAGE_OUT.relative_to(REPO_ROOT)}", file=sys.stderr)
+    return 0 if not unassigned_modules else 1
+
+
+def cmd_sample(inv, wave, limit, out):
+    emitted = 0
+    records = []
+    for relpath, unit, entry in iter_units(inv):
+        if match_module(relpath) != wave:
+            continue
+        records.append(render_issue(wave, relpath, unit, entry))
+        emitted += 1
+        if emitted >= limit:
+            break
+    Path(out).write_text("\n".join(json.dumps(r) for r in records) + "\n")
+    print(f"sampled {emitted} issues for wave {wave} -> {out}", file=sys.stderr)
+    return 0
+
+
+def cmd_create(inv, wave, limit, dry_gap):
+    existing = fetch_existing_titles()
+    created, skipped = 0, 0
+    for relpath, unit, entry in iter_units(inv):
+        if match_module(relpath) != wave:
+            continue
+        title = issue_title(wave, relpath, unit)
+        if title in existing:
+            skipped += 1
+            continue
+        issue = create_issue(render_issue(wave, relpath, unit, entry))
+        created += 1
+        print(f"created #{issue.get('number')}: {title}", file=sys.stderr)
+        if limit and created >= limit:
+            break
+        time.sleep(dry_gap)
+    print(f"\nwave {wave}: created={created} skipped(dedup)={skipped}", file=sys.stderr)
+    return 0
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("command", choices=["plan", "sample", "create"])
+    ap.add_argument("--wave", type=int, help="wave number (2-7)")
+    ap.add_argument("--limit", type=int, default=0, help="max issues for sample/create (0 = unlimited; sample defaults to 5)")
+    ap.add_argument("--out", default=None, help="sample output file")
+    ap.add_argument("--gap", type=float, default=0.5, help="seconds between creates")
+    args = ap.parse_args()
+
+    inv_cache.update(json.loads(INVENTORY.read_text()))
+    inv = inv_cache
+
+    if args.command == "plan":
+        return cmd_plan(inv)
+    if args.command == "sample":
+        if not args.wave:
+            ap.error("--wave required")
+        out = args.out or f"docs/rebase-factory-sample-w{args.wave}.json"
+        return cmd_sample(inv, args.wave, args.limit or 5, out)
+    if args.command == "create":
+        if not args.wave:
+            ap.error("--wave required")
+        return cmd_create(inv, args.wave, args.limit, args.gap)
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

shared/native_units.py

@@ -0,0 +1,103 @@
+"""Thin adapter for native (compiled) units — R1 §5 interop contract, Wave 0.
+
+The contract under proof: compiled units are standalone binaries exposing a
+CLI boundary — argv in, JSON on stdout, non-zero exit on failure, a mandatory
+``--version`` flag. This module is the Python side: run a unit, enforce a
+timeout, capture stderr, and return a TYPED result. It never raises for unit
+failures (those are data in the result); it raises only for adapter misuse
+(missing binary, bad timeout).
+
+No IPC daemon, no PyO3 — the file-based store stays the shared truth. If
+research later surfaces a materially better pattern it competes through the
+research pipeline; the CLI boundary is the default, not a cage.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+PROOF_UNIT = REPO_ROOT / "interop" / "rebase-proof" / "target" / "release" / "rebase-proof"
+
+
+@dataclass
+class NativeResult:
+    """Typed result of one native-unit invocation.
+
+    ok           — the invocation succeeded (exit 0 AND not timed out); when
+                    the unit emitted JSON with an ``ok`` field, that field must
+                    not be False (plain-text output like ``--version`` has none)
+    exit_code    — raw process exit code
+    data         — parsed JSON object from stdout, or None if unparseable
+    stderr       — captured stderr text (may be non-empty even on success)
+    duration_ms  — wall-clock cost of the invocation
+    timed_out    — True when the unit exceeded the timeout and was killed
+    """
+
+    ok: bool
+    exit_code: int
+    data: dict | None
+    stderr: str
+    duration_ms: int
+    timed_out: bool = False
+    raw_stdout: str = field(default="", repr=False)
+
+
+def run_native(binary: Path, args: list[str], timeout_s: float = 10.0) -> NativeResult:
+    """Invoke a native unit and return a typed result.
+
+    Raises FileNotFoundError only when *binary* itself is absent (adapter
+    misuse) — every unit-level failure is reported via the result, never an
+    exception, so callers can branch on ``result.ok`` without try/except.
+    """
+    if not binary.is_file():
+        raise FileNotFoundError(f"native unit not built: {binary}")
+
+    started = time.monotonic()
+    timed_out = False
+    try:
+        proc = subprocess.run(
+            [str(binary), *args],
+            capture_output=True,
+            text=True,
+            timeout=timeout_s,
+            check=False,  # exit codes are data, never exceptions
+        )
+        exit_code, stdout, stderr = proc.returncode, proc.stdout, proc.stderr
+    except subprocess.TimeoutExpired as exc:
+        timed_out = True
+        exit_code = -1
+        stdout = exc.stdout or "" if isinstance(exc.stdout, str) else (exc.stdout or b"").decode(errors="replace")
+        stderr = exc.stderr or "" if isinstance(exc.stderr, str) else (exc.stderr or b"").decode(errors="replace")
+
+    duration_ms = int((time.monotonic() - started) * 1000)
+
+    data: dict | None = None
+    stripped = stdout.strip()
+    if stripped:
+        try:
+            parsed = json.loads(stripped.splitlines()[-1])
+            if isinstance(parsed, dict):
+                data = parsed
+        except json.JSONDecodeError:
+            data = None
+
+    ok = (not timed_out) and exit_code == 0 and (data is None or data.get("ok") is not False)
+    return NativeResult(
+        ok=ok,
+        exit_code=exit_code,
+        data=data,
+        stderr=stderr,
+        duration_ms=duration_ms,
+        timed_out=timed_out,
+        raw_stdout=stdout,
+    )
+
+
+def run_proof(args: list[str], timeout_s: float = 10.0) -> NativeResult:
+    """Run the Wave 0 proof unit with the standard contract defaults."""
+    return run_native(PROOF_UNIT, args, timeout_s=timeout_s)

tests/test_native_interop_proof.py

@@ -0,0 +1,78 @@
+"""Wave 0 interop proof tests (R1 §5): Rust unit callable from tests via the
+thin Python adapter. Skips when the binary is not built so CI without Rust
+stays green; build locally with:
+    cargo build --release --quiet  (from interop/rebase-proof/)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "src"))
+sys.path.insert(0, str(ROOT))
+
+import pytest  # noqa: E402
+
+from shared.native_units import PROOF_UNIT, NativeResult, run_native, run_proof  # noqa: E402
+
+pytestmark = pytest.mark.skipif(
+    not PROOF_UNIT.is_file(), reason="rebase-proof binary not built (cargo build --release in interop/rebase-proof)"
+)
+
+
+def test_version_flag_is_mandatory_contract():
+    result = run_proof(["--version"])
+    assert result.ok
+    assert result.exit_code == 0
+    assert result.data is None or result.data == {}  # version prints plain text, not JSON
+    assert result.raw_stdout.strip() == "rebase-proof 0.1.0"
+
+
+def test_sum_happy_path_json_shape():
+    result = run_proof(["sum", "1", "2", "3"])
+    assert result.ok
+    assert result.data == {"ok": True, "op": "sum", "result": 6}
+
+
+def test_hash_is_deterministic():
+    first = run_proof(["hash", "hello"])
+    second = run_proof(["hash", "hello"])
+    assert first.ok and second.ok
+    assert first.data == second.data
+    assert first.data["fnv1a64"] == "a430d84680aabd0b"
+
+
+def test_error_op_exits_nonzero_with_json_error():
+    result = run_proof(["fail"])
+    assert not result.ok
+    assert result.exit_code == 2
+    assert result.data == {"ok": False, "error": "deliberate failure"}
+    assert "deliberate failure for adapter proof" in result.stderr
+
+
+def test_unknown_op_is_data_not_exception():
+    result = run_proof(["no-such-op"])
+    assert not result.ok
+    assert result.exit_code == 2
+    assert "unknown op" in result.data["error"]
+
+
+def test_timeout_enforcement_kills_unit():
+    result = run_proof(["sleep", "30000"], timeout_s=0.5)
+    assert result.timed_out
+    assert not result.ok
+    assert result.exit_code == -1
+
+
+def test_missing_binary_raises_adapter_misuse():
+    with pytest.raises(FileNotFoundError):
+        run_native(ROOT / "interop" / "rebase-proof" / "target" / "release" / "does-not-exist", ["sum", "1"])
+
+
+def test_typed_result_is_the_contract_surface():
+    result = run_proof(["sum", "10", "20"])
+    assert isinstance(result, NativeResult)
+    assert result.duration_ms >= 0
+    assert result.stderr == ""