Draft: Clef reader and budgeted owner-thread hidden rows

.github/workflows/no-gpu.yml

@@ -24,6 +24,10 @@ jobs:
       # The kernel contract's limits gate: every registered launch against M1-M5 and sm_80/86/89/120/121, controls red.
       - name: zig build test-kernel-contract
         run: zig build test-kernel-contract
+      - name: Clef head inspection CLI (CPU only)
+        run: |
+          zig build tf-clef-inspect
+          python3 tools/zig/test_clef_inspect.py zig-out/bin/tf-clef-inspect
       # The frozen server goldens: request parsing, rendering and streaming against recorded answers.
       - name: zig build test-golden
         run: zig build test-golden

THIRD_PARTY_NOTICES.md

@@ -232,3 +232,12 @@ TensorFold. What they follow or copy:
 
 - The Level Zero loader and the Intel compute runtime are loaded at run time and are not bundled; `ocloc` compiles the kernels
   at build time and is not distributed.
+## Clef head checkpoint
+
+`zig/src/families/clef/weights.zig` reads the separate bf16 head format from Cloudflare's
+[`joint_schema_model.py`](https://huggingface.co/Cloudflare/clef-flash/blob/8b2e5fd17c09fd49bc2880b805d5323ec7ff4ff3/joint_schema_model.py),
+Apache License 2.0. The earlier TensorFold work by @outcastofmusic in
+[#241](https://github.com/ashhart/TensorFold/pull/241) and
+[#251](https://github.com/ashhart/TensorFold/pull/251) established the separate head and backbone interface.
+`zig/tests/fixtures/clef/` contains a small synthetic head generated with the release code at seed 317.
+Its source metadata identifies the release revision. It contains no trained model weights or agreement corpus.

build.zig

@@ -161,6 +161,22 @@ pub fn build(b: *std.Build) void {
     b.step("test-cost-rule", "CPU scripted window and interleaved head timing").dependOn(&cost_run.step);
     test_step.dependOn(&cost_run.step);
     merge_build.targets(b, target, optimize); // tf-merge-check and test-merge, host only on every OS
+    const clef_head = b.addTest(.{ .root_module = b.createModule(.{
+        .root_source_file = b.path("zig/clef_checkpoint_cases.zig"),
+        .target = target,
+        .optimize = optimize,
+        .link_libc = true,
+    }) });
+    const clef_head_run = b.addRunArtifact(clef_head);
+    b.step("test-clef-checkpoint", "CPU Clef checkpoint and disposable-pass contracts").dependOn(&clef_head_run.step);
+    test_step.dependOn(&clef_head_run.step);
+    const clef_inspect = b.addExecutable(.{ .name = "tf-clef-inspect", .root_module = b.createModule(.{
+        .root_source_file = b.path("zig/clef_inspect.zig"),
+        .target = target,
+        .optimize = .safe,
+        .link_libc = true,
+    }) });
+    b.step("tf-clef-inspect", "Build CPU-only Clef head and backbone-config inspection").dependOn(&b.addInstallArtifact(clef_inspect, .{}).step);
     switch (target.result.os.tag) {
         .macos => metalTargets(b, target, optimize, draft_ids, build_options, test_step),
         .linux => cuda_build.targets(b, target, optimize, draft_ids, build_options),

tools/zig/test_clef_inspect.py

@@ -0,0 +1,79 @@
+"""CPU CLI receipts distinguish a validated head from a plain or damaged checkpoint."""
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+BINARY = Path(sys.argv.pop(1)).resolve() if len(sys.argv) > 1 and not sys.argv[1].startswith("-") else ROOT / "zig-out/bin/tf-clef-inspect"
+FIXTURE = ROOT / "zig/tests/fixtures/clef"
+CONFIG = {
+    "model_type": "qwen3_5_text", "hidden_size": 16, "intermediate_size": 32,
+    "num_hidden_layers": 4, "num_attention_heads": 2, "num_key_value_heads": 1,
+    "head_dim": 8, "vocab_size": 97, "linear_num_key_heads": 2,
+    "linear_num_value_heads": 4, "linear_key_head_dim": 4,
+    "linear_value_head_dim": 4, "linear_conv_kernel_dim": 4, "eos_token_id": 2,
+}
+
+
+class InspectTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.path = Path(self.tmp.name)
+        (self.path / "config.json").write_text(json.dumps(CONFIG))
+
+    def pair(self):
+        for name in ("joint_head_config.json", "joint_head.safetensors"):
+            shutil.copyfile(FIXTURE / name, self.path / name)
+
+    def run_cli(self):
+        return subprocess.run([str(BINARY), str(self.path), "--check"], text=True, capture_output=True, timeout=30)
+
+    def rejected(self):
+        result = self.run_cli()
+        self.assertNotEqual(0, result.returncode)
+        self.assertEqual("", result.stdout, "a failed check must not emit a success receipt")
+
+    def test_complete_pair_has_verified_payload_hash_and_no_model_readiness_claim(self):
+        self.pair()
+        result = self.run_cli()
+        self.assertEqual(0, result.returncode, result.stderr)
+        record = json.loads(result.stdout)
+        self.assertTrue(record["head_present"] and record["head_validated"])
+        self.assertEqual(122, record["head_tensors"])
+        self.assertEqual(16, record["backbone_hidden"])
+        self.assertEqual(8, record["head_config"]["width"])
+        self.assertEqual(hashlib.sha256((self.path / "joint_head.safetensors").read_bytes()).hexdigest(), record["head_file_sha256"])
+        self.assertFalse(record["backbone_weights_checked"] or record["gpu"] or record["model_ready"])
+
+    def test_plain_pack_has_no_validated_head_or_hash(self):
+        result = self.run_cli()
+        self.assertEqual(0, result.returncode, result.stderr)
+        record = json.loads(result.stdout)
+        self.assertFalse(record["head_present"] or record["head_validated"] or record["model_ready"])
+        self.assertEqual(0, record["head_tensors"])
+        self.assertIsNone(record["head_config"])
+        self.assertIsNone(record["head_file_sha256"])
+
+    def test_incomplete_pair_and_truncated_payload_are_errors(self):
+        shutil.copyfile(FIXTURE / "joint_head_config.json", self.path / "joint_head_config.json")
+        self.rejected()
+        self.pair()
+        payload = self.path / "joint_head.safetensors"
+        payload.write_bytes(payload.read_bytes()[:-2])
+        self.rejected()
+
+    def test_incompatible_backbone_is_an_error(self):
+        self.pair()
+        (self.path / "config.json").write_text(json.dumps({**CONFIG, "hidden_size": 32}))
+        self.rejected()
+
+
+if __name__ == "__main__":
+    unittest.main()

zig/build/qwen27.zig

@@ -111,13 +111,15 @@ pub fn targets(b: *std.Build, target: std.Build.ResolvedTarget, optimize: std.bu
     _ = program(b, "tf-qwen27-dflash-run", "Build the explicit draft prototype and its plain comparison on real weights", "zig/tests/qwen27/dflash_run.zig", target, true, family);
     const engine_only: []const std.Build.Module.Import = &.{ .{ .name = "tensorfold", .module = engine }, .{ .name = "metal", .module = metal } };
     _ = program(b, "tf-qwen27-synthetic", "Build bounded synthetic native hybrid chunk checks, no models", "zig/tests/qwen27/synthetic.zig", target, false, engine_only);
+    _ = program(b, "tf-qwen27-hidden-rows-check", "Build disposable normalized-row and chat/resume checks without running a GPU", "zig/tests/qwen27/hidden_rows.zig", target, false, engine_only);
     _ = program(b, "tf-qwen27-dflash-synthetic", "Build model-free native draft context/forward/tree check", "zig/tests/qwen27/dflash_runtime.zig", target, false, engine_only);
     _ = program(b, "tf-qwen27-widths", "Build the projections at every oQ width before tensor units: float64, row windows, mixed stacks, embedding", "zig/tests/qwen27/widths.zig", target, false, engine_only);
     _ = program(b, "tf-qwen27-dflash-refine", "Build the drafter refinement probe (top-1 by block position, masked versus given)", "zig/tests/qwen27/dflash_refine.zig", target, false, engine_only);
 
     const api = b.createModule(.{ .root_source_file = b.path("zig/src/core/engine_api.zig"), .target = target, .optimize = .ReleaseSafe, .link_libc = true, .imports = &.{.{ .name = "lanes", .module = lanes }} });
     const host = b.createModule(.{ .root_source_file = b.path("zig/src/native/qwen27_served.zig"), .target = target, .optimize = .ReleaseSafe, .link_libc = true, .imports = &.{ .{ .name = "engine_api", .module = api }, .{ .name = "metal", .module = metal }, .{ .name = "tensorfold", .module = engine } } });
     const served: []const std.Build.Module.Import = &.{ .{ .name = "tensorfold", .module = engine }, .{ .name = "metal", .module = metal }, .{ .name = "engine_api", .module = api }, .{ .name = "qwen27_served", .module = host } };
+    _ = program(b, "tf-qwen27-hidden-rows-served-check", "Build queued capture versus solo and concurrent chat checks without running a GPU", "zig/tests/qwen27/hidden_rows_served.zig", target, false, served);
     _ = program(b, "tf-qwen27-served-check", "Build synthetic CLI versus served-host token check", "zig/tests/qwen27/served.zig", target, false, served);
     _ = program(b, "tf-qwen27-dflash-served-check", "Build synthetic drafted CLI versus served-host token check", "zig/tests/qwen27/dflash_served.zig", target, false, served);
     _ = program(b, "tf-qwen27-lanes-check", "Build synthetic lane-core versus plain greedy token and state check", "zig/tests/qwen27/lanes_check.zig", target, false, served);

zig/clef_checkpoint_cases.zig

@@ -0,0 +1,7 @@
+//! CPU-only Clef head admission and disposable-pass contracts; no device or scoring arithmetic.
+test {
+    _ = @import("src/families/clef/weights.zig");
+    _ = @import("src/families/clef/checkpoint.zig");
+    _ = @import("src/families/qwen27/ane_handoff.zig");
+    _ = @import("src/families/qwen27/hidden_rows_plan.zig");
+}

zig/clef_inspect.zig

@@ -0,0 +1,44 @@
+//! CPU inspection of a separate Clef head against its backbone config; no backbone payload or device is opened.
+const std = @import("std");
+const checkpoint = @import("src/families/clef/checkpoint.zig");
+const qwen = @import("src/families/qwen27/config.zig");
+
+fn digest(bytes: []const u8) [64]u8 {
+    var sum: [32]u8 = undefined;
+    std.crypto.hash.sha2.Sha256.hash(bytes, &sum, .{});
+    return std.fmt.bytesToHex(sum, .lower);
+}
+
+pub fn main(init: std.process.Init) !void {
+    const a = init.arena.allocator();
+    const io = init.io;
+    const args = try init.minimal.args.toSlice(a);
+    if (args.len != 3 or !std.mem.eql(u8, args[2], "--check")) return error.CheckModeRequired;
+    const path = try std.fs.path.join(a, &.{ args[1], "config.json" });
+    const text = try std.Io.Dir.cwd().readFileAlloc(io, path, a, .limited(1 << 20));
+    const c = try qwen.parse(a, text, null);
+    var head = try checkpoint.loadOptional(init.gpa, io, args[1], c.hidden);
+    defer if (head) |*h| h.deinit(io);
+    const config_sha = digest(text);
+    var head_sha: [64]u8 = undefined;
+    const sha: ?[]const u8 = if (head) |h| blk: {
+        head_sha = digest(h.file.map.memory);
+        break :blk &head_sha;
+    } else null;
+    const record = try std.json.Stringify.valueAlloc(a, .{
+        .kind = "clef_head_checkpoint",
+        .version = 1,
+        .head_present = head != null,
+        .head_validated = head != null,
+        .head_tensors = if (head) |h| h.file.names.count() else @as(usize, 0),
+        .head_config = if (head) |h| h.config else null,
+        .head_file_sha256 = sha,
+        .backbone_hidden = c.hidden,
+        .backbone_config_sha256 = @as([]const u8, &config_sha),
+        .backbone_weights_checked = false,
+        .gpu = false,
+        .model_ready = false,
+    }, .{});
+    try std.Io.File.stdout().writeStreamingAll(io, record);
+    try std.Io.File.stdout().writeStreamingAll(io, "\n");
+}

zig/src/core/lane_host.zig

@@ -11,6 +11,7 @@ const admit = @import("lane_host_admit.zig");
 const recalls = @import("lane_host_recall.zig");
 const ends = @import("lane_host_end.zig");
 const rows_host = @import("lane_host_rows.zig");
+const calls = @import("lane_host_calls.zig");
 const Allocator = std.mem.Allocator;
 const Id = api.Id;
 const Request = api.Request;
@@ -36,6 +37,8 @@ pub const LaneHost = struct {
     closing: bool = false,
     stale: bool = false, // the learner's last step: a put-back has not reached the device, so replies are refused
     thread: ?std.Thread = null,
+    owner_thread: ?std.Thread.Id = null,
+    calls: calls.Queue = .{},
     decoded: std.ArrayList(status.Mark) = .empty, // tokens a round landed, for the 2 s decode rate
     prefill_rate: f64 = 0,
     prefill_at: i96 = 0,
@@ -141,6 +144,11 @@ pub const LaneHost = struct {
         } };
     }
 
+    /// Family work waits for the engine thread; ctx must remain valid until this returns, including cancellation.
+    pub fn callOwner(h: *LaneHost, ctx: *anyopaque, operation: calls.Run) calls.Error!void {
+        return calls.submit(h, ctx, operation);
+    }
+
     /// The family's queue as a keepalive target, when it set one.
     fn keepaliveFn(ctx: *anyopaque) ?api.keepalive.Target {
         return self(ctx).keepalive_target;
@@ -207,9 +215,9 @@ pub const LaneHost = struct {
         return job.host.foregroundWaiting() or job.host.core.activeCount() > 0;
     }
 
-    /// A foreground request is queued (they queue ahead of every background one); the caller holds the lock.
+    /// A foreground request or family call waits; the caller holds the lock.
     pub fn foregroundWaiting(h: *const LaneHost) bool {
-        return h.queued.items.len > 0 and !h.queued.items[0].request.background;
+        return h.calls.first != null or (h.queued.items.len > 0 and !h.queued.items[0].request.background);
     }
 
     /// Waits `warm_quiet` for other work (a request, a decoding stream, a cancel): true if none came.
@@ -495,10 +503,14 @@ pub const LaneHost = struct {
     }
 
     fn run(h: *LaneHost) void {
+        h.lock();
+        h.owner_thread = std.Thread.getCurrentId();
+        h.unlock();
         while (true) {
             ends.takeCancels(h);
             rows_host.answer(h);
             learn.refuse(h);
+            calls.one(h);
             while (h.admitOne()) {}
             h.fillOne();
             status.noteLive(h);
@@ -525,7 +537,7 @@ pub const LaneHost = struct {
                     continue;
                 }
                 if (rows_host.idle(h)) continue;
-                if (h.cancels.items.len == 0 and (h.queued.items.len == 0 or h.admitted.items.len >= h.info_.lanes))
+                if (h.calls.first == null and h.cancels.items.len == 0 and (h.queued.items.len == 0 or h.admitted.items.len >= h.info_.lanes))
                     h.wake.waitTimeout(h.io, &h.mutex, .{ .duration = .{ .raw = .fromMilliseconds(100), .clock = .awake } }) catch {};
                 h.unlock();
                 continue;

zig/src/core/lane_host_calls.zig

@@ -0,0 +1,111 @@
+//! Family operations run on the engine thread between rounds, with caller-owned completion storage.
+const std = @import("std");
+const LaneHost = @import("lane_host.zig").LaneHost;
+
+pub const Error = error{ Closed, NotStarted, ReentrantOwnerCall, StaleModel };
+pub const Run = *const fn (*anyopaque) void;
+
+const Call = struct {
+    next: ?*Call = null,
+    io: std.Io,
+    ctx: *anyopaque,
+    run: Run,
+    mutex: std.Io.Mutex = .init,
+    wake: std.Io.Condition = .init,
+    done: bool = false,
+    failure: ?Error = null,
+
+    fn finish(c: *Call, failure: ?Error) void {
+        const io = c.io;
+        c.mutex.lockUncancelable(io);
+        c.failure = failure;
+        c.done = true;
+        c.wake.signal(io);
+        c.mutex.unlock(io);
+        // The caller can now return and destroy this node. Never touch c again.
+    }
+
+    fn wait(c: *Call) Error!void {
+        const io = c.io;
+        c.mutex.lockUncancelable(io);
+        defer c.mutex.unlock(io);
+        while (!c.done) c.wake.waitUncancelable(io, &c.mutex);
+        if (c.failure) |err| return err;
+    }
+};
+
+/// Intrusive nodes stay on their callers' stacks until their own completion is observed.
+pub const Queue = struct {
+    first: ?*Call = null,
+    last: ?*Call = null,
+
+    fn append(q: *Queue, c: *Call) void {
+        if (q.last) |last| last.next = c else q.first = c;
+        q.last = c;
+    }
+
+    fn pop(q: *Queue) ?*Call {
+        const c = q.first orelse return null;
+        q.first = c.next;
+        if (q.first == null) q.last = null;
+        c.next = null;
+        return c;
+    }
+};
+
+/// Once accepted, even a canceled caller waits until the owner has finished using ctx.
+pub fn submit(h: *LaneHost, ctx: *anyopaque, run: Run) Error!void {
+    var c = Call{ .io = h.io, .ctx = ctx, .run = run };
+    h.lock();
+    if (h.closing) {
+        h.unlock();
+        return error.Closed;
+    }
+    if (h.stale) {
+        h.unlock();
+        return error.StaleModel;
+    }
+    if (h.thread == null) {
+        h.unlock();
+        return error.NotStarted;
+    }
+    if (h.owner_thread == std.Thread.getCurrentId()) {
+        h.unlock();
+        return error.ReentrantOwnerCall;
+    }
+    h.calls.append(&c);
+    h.wake.signal(h.io);
+    h.unlock();
+    // No host access after enqueue: stop may free it while this caller observes its own completion.
+    try c.wait();
+}
+
+/// At most one operation per ordinary loop iteration; a running lesson reaches its safe end first.
+pub fn one(h: *LaneHost) void {
+    h.lock();
+    if (h.stale) {
+        const pending = detach(&h.calls);
+        h.unlock();
+        return reject(pending, error.StaleModel);
+    }
+    const c = if (h.closing or h.lesson_open) null else h.calls.pop();
+    h.unlock();
+    const call = c orelse return;
+    call.run(call.ctx);
+    call.finish(null);
+}
+
+/// Detach pending work under the host lock, then finish it without that lock or host storage.
+pub fn detach(q: *Queue) ?*Call {
+    const first = q.first;
+    q.* = .{};
+    return first;
+}
+
+pub fn reject(first: ?*Call, failure: Error) void {
+    var next = first;
+    while (next) |c| {
+        next = c.next;
+        c.finish(failure);
+    }
+}