ADR-0010: device binding interface, schemas and conformance kit

.github/workflows/ci.yml

@@ -18,6 +18,8 @@ jobs:
         run: test -z "$(gofmt -l .)" || { gofmt -l .; exit 1; }
       - run: go vet ./...
       - run: go test -race ./...
+      - name: device binding conformance kit (reference fake binding, B1 and B2)
+        run: python3 conformance/binding/conformance.py --level B2 && python3 conformance/binding/conformance.py --level B1
       - name: chunker benchmark (reported, not gated yet)
         run: go test -run xxx -bench . -benchtime 3x ./pkg/chunker
       - name: build linux/amd64 CLI

cmd/onggi/store.go

@@ -375,7 +375,7 @@ func diff(args []string, stdout, stderr io.Writer) int {
 	fs.SetOutput(stderr)
 	sf := addStoreFlags(fs)
 	asJSON := fs.Bool("json", false, "print the diff as JSON")
-	ranges := fs.Bool("ranges", false, "list the changed byte ranges of each file (with --json: [offset, length] pairs)")
+	ranges := fs.Bool("ranges", false, "list the changed byte ranges of each file; with --json also the range plan of each file, schema onggi.switchplan.v1 (ADR-0010)")
 	if err := fs.Parse(args); err != nil || fs.NArg() != 2 {
 		fmt.Fprintln(stderr, "usage: onggi diff --store S [--json] [--ranges] <base ref|sha256:...> <ref|sha256:...>")
 		return 2

conformance/binding/README.md

@@ -0,0 +1,47 @@
+# Device binding conformance kit
+
+Checks a device binding against the interface of [ADR-0010](../../docs/adr/0010-device-binding-contract.md).
+Python 3.6+, standard library only, Apache-2.0.
+
+```bash
+python3 conformance.py                          # the reference fake binding, level B2
+python3 conformance.py --level B1               # the fake binding acting as B1 (plan always reloads)
+python3 conformance.py --level B2 -- ./my-binding --device 0   # your binding
+```
+
+Your binding is run as `COMMAND <operation> [args]` with JSON on stdin and stdout and a non-zero exit on error:
+
+| operation | stdin | stdout |
+|---|---|---|
+| `describe ARTIFACT_DIR` | — | `onggi.profile.v1` |
+| `resident` | — | `onggi.resident.v1` |
+| `plan` | `onggi.switchplan.v1` | `onggi.updateplan.v1` |
+| `apply --target DIR` | `onggi.updateplan.v1` | `onggi.verification.v1` |
+| `rollback` | — | `onggi.verification.v1` |
+| `conformance-load ARTIFACT_DIR MANIFEST` | — | — (kit hook: load the base artifact the normal way and remember MANIFEST as its identity) |
+
+Schemas and examples: [`docs/schemas/`](../../docs/schemas/).
+
+## What is in here
+
+- `golden/*.switchplan.json`: what `onggi diff --ranges --json` prints for four synthetic switches of a 2 MiB
+  `model.bin` and a `config.json`: `identical`, `inplace` (40 KiB overwritten, config changed), `shifted` (17,952 B
+  inserted in front, as a larger TensorRT-LLM batch does) and `rearranged` (the same bytes in 32 KiB blocks in another
+  order). `go test ./pkg/store -run GoldenSwitchPlans` fails when onggi's output drifts from them (`-update`
+  rewrites them).
+- `golden_data.py`: generates the bytes of those artifacts (SHA-256 in counter mode, the same in Go).
+- `fake_binding.py`: the reference binding on a simulated device (a directory of 4 KiB rows). It maps changed runs to
+  rows, answers `reload` when the base is not resident or more than 64 runs or 25 % of the bytes change, stages and
+  journals before writing, reads back to verify, and rolls back from the journal.
+- `conformance.py`: the tests, per golden switch:
+  1. `describe` returns `onggi.profile.v1`; after `conformance-load`, `resident` names the base.
+  2. `plan` is pure: the same answer twice and the resident state unchanged.
+  3. The update plan names the switch plan's base and target; at B1 it is always `reload`; at B2 the `identical` and
+     `inplace` switches are applied as units, and units cover every byte that is not `same`.
+  4. A plan for a base that is not resident is refused (`reload` or an error).
+  5. B2: `apply` returns digests equal to the target's bytes (all target files for a reload), the target is then
+     resident, `rollback` returns digests equal to the base's bytes, the base is resident again, and a second
+     rollback fails.
+
+A real binding computes the digests from what it reads back from the device, mapped to artifact byte order. The kit
+does not check timing or draining; those belong to the runtime's own tests.

conformance/binding/conformance.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+# Copyright 2026 The onggi authors. SPDX-License-Identifier: Apache-2.0
+"""Conformance tests for an onggi device binding (ADR-0010). Python 3.6+, standard library only.
+
+    python3 conformance.py [--level B1|B2] [-- BINDING COMMAND...]
+
+Without a command it tests the reference fake binding. The binding command is run as
+`COMMAND <operation> [args]` with JSON on stdin and stdout; see README.md.
+"""
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+import golden_data  # noqa: E402
+
+FAILS = []
+
+
+def check(cond, what):
+    print(("ok    " if cond else "FAIL  ") + what)
+    if not cond:
+        FAILS.append(what)
+    return cond
+
+
+def run(cmd, op, *args, stdin=None, expect_ok=True):
+    p = subprocess.run(cmd + [op] + list(args), input=None if stdin is None else json.dumps(stdin).encode(),
+                       stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if expect_ok and p.returncode != 0:
+        raise RuntimeError("%s %s: exit %d: %s" % (op, " ".join(args), p.returncode, p.stderr.decode()[-500:]))
+    try:
+        return p.returncode, json.loads(p.stdout.decode()) if p.stdout.strip() else None
+    except ValueError:
+        raise RuntimeError("%s: stdout is not JSON: %r" % (op, p.stdout[:200]))
+
+
+def sha(b):
+    return hashlib.sha256(b).hexdigest()
+
+
+def covers(update, switch):
+    """Every byte that is not `same` in the switch plan lies in a unit of the same file."""
+    units = {}
+    for u in update["units"]:
+        units.setdefault(u["file"], []).append((u["offset"], u["offset"] + u["length"]))
+    for f in switch["files"]:
+        for r in f["ranges"]:
+            if r["kind"] == "same":
+                continue
+            pos, end = r["offset"], r["offset"] + r["length"]
+            for a, b in sorted(units.get(f["path"], [])):
+                if a <= pos < b:
+                    pos = b
+            if pos < end:
+                return False
+    return True
+
+
+def verify_units(ver, files, what):
+    ok = ver.get("schema") == "onggi.verification.v1" and ver.get("ok") is True
+    for u in ver.get("units", []):
+        want = files.get(u["file"], b"")[u["offset"]:u["offset"] + u["length"]]
+        ok = ok and len(want) == u["length"] and u["sha256"] == sha(want)
+    return check(ok, what)
+
+
+def main(argv):
+    level, cmd = "B2", None
+    if "--" in argv:
+        cmd = argv[argv.index("--") + 1:]
+        argv = argv[:argv.index("--")]
+    if "--level" in argv:
+        level = argv[argv.index("--level") + 1]
+    work = tempfile.mkdtemp(prefix="onggi-binding-")
+    if not cmd:
+        os.environ["ONGGI_FAKE_DEVICE"] = os.path.join(work, "device")
+        if level == "B1":
+            os.environ["ONGGI_FAKE_LEVEL"] = "B1"
+        cmd = [sys.executable, os.path.join(HERE, "fake_binding.py")]
+    for name in golden_data.CASES:
+        base, target = golden_data.case(name)
+        bdir, tdir = os.path.join(work, name, "base"), os.path.join(work, name, "target")
+        golden_data.write(bdir, base)
+        golden_data.write(tdir, target)
+        with open(os.path.join(HERE, "golden", name + ".switchplan.json")) as f:
+            sp = json.load(f)
+        print("== %s (%s)" % (name, level))
+
+        _, prof = run(cmd, "describe", tdir)
+        check(prof.get("schema") == "onggi.profile.v1", "describe returns onggi.profile.v1")
+
+        run(cmd, "conformance-load", bdir, sp["base_manifest"])
+        _, res = run(cmd, "resident")
+        check(res.get("schema") == "onggi.resident.v1" and res.get("manifest") == sp["base_manifest"],
+              "resident names the loaded base")
+
+        _, up1 = run(cmd, "plan", stdin=sp)
+        _, up2 = run(cmd, "plan", stdin=sp)
+        _, res2 = run(cmd, "resident")
+        check(up1 == up2 and res2 == res, "plan is pure: same answer twice, resident state unchanged")
+        check(up1.get("schema") == "onggi.updateplan.v1" and up1.get("mode") in ("units", "reload"),
+              "plan returns onggi.updateplan.v1 with mode units or reload")
+        check(up1["base_manifest"] == sp["base_manifest"] and up1["manifest"] == sp["manifest"],
+              "update plan names the switch plan's base and target")
+        if level == "B1":
+            check(up1["mode"] == "reload", "B1: plan always answers reload")
+        elif name in ("identical", "inplace"):
+            check(up1["mode"] == "units", "B2: a switch that keeps 97 % in place is applied as units")
+        if up1["mode"] == "units":
+            check(covers(up1, sp), "units cover every byte that is not at the same offset")
+
+        wrong = dict(sp, base_manifest="sha256:" + "0" * 64)
+        rc, upw = run(cmd, "plan", stdin=wrong, expect_ok=False)
+        check(rc != 0 or (upw or {}).get("mode") == "reload", "plan against a base that is not resident refuses (reload)")
+
+        if level == "B1":
+            continue
+        _, ver = run(cmd, "apply", "--target", tdir, stdin=up1)
+        verify_units(ver, target, "apply verifies: unit digests equal the target bytes")
+        if up1["mode"] == "reload":
+            check(sorted(u["file"] for u in ver["units"]) == sorted(target), "reload writes every target file")
+        _, res3 = run(cmd, "resident")
+        check(res3.get("manifest") == sp["manifest"], "after apply the target is resident")
+
+        _, rb = run(cmd, "rollback")
+        check(rb.get("op") == "rollback", "rollback returns a verification")
+        verify_units(rb, base, "rollback restores: unit digests equal the base bytes")
+        _, res4 = run(cmd, "resident")
+        check(res4.get("manifest") == sp["base_manifest"], "after rollback the base is resident")
+        rc, _ = run(cmd, "rollback", expect_ok=False)
+        check(rc != 0, "a second rollback has nothing to undo and fails")
+
+    print("\n%d failed" % len(FAILS))
+    return 1 if FAILS else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))